Merge pull request thanos-io#372 from Shopify/shopify-proxy-dedup

Disable dedup proxy in multi-tsdb

cmd/thanos/receive.go

@@ -319,6 +319,7 @@ func runReceive(
 
 		options := []store.ProxyStoreOption{
 			store.PropagateStoreSelectorMatchers(false),
+			store.WithoutDedup(),
 		}
 		if debugLogging {
 			options = append(options, store.WithProxyStoreDebugLogging())

pkg/store/proxy.go

@@ -90,6 +90,7 @@ type ProxyStore struct {
 	debugLogging              bool
 	propagateSelectorMatchers bool
 	storeSelector             *storeSelector
+	enableDedup               bool
 }
 
 type proxyStoreMetrics struct {
@@ -137,6 +138,13 @@ func PropagateStoreSelectorMatchers(value bool) ProxyStoreOption {
 	}
 }
 
+// WithoutDedup disabled chunk deduplication when streaming series.
+func WithoutDedup() ProxyStoreOption {
+	return func(s *ProxyStore) {
+		s.enableDedup = false
+	}
+}
+
 // NewProxyStore returns a new ProxyStore that uses the given clients that implements storeAPI to fan-in all series to the client.
 // Note that there is no deduplication support. Deduplication should be done on the highest level (just before PromQL).
 func NewProxyStore(
@@ -163,11 +171,13 @@ func NewProxyStore(
 			b := make([]byte, 0, initialBufSize)
 			return &b
 		}},
+
 		responseTimeout:           responseTimeout,
 		metrics:                   metrics,
 		retrievalStrategy:         retrievalStrategy,
 		propagateSelectorMatchers: true,
 		storeSelector:             newStoreSelector(nil),
+		enableDedup:               true,
 	}
 
 	for _, option := range options {
@@ -399,7 +409,10 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb.
 
 	level.Debug(reqLogger).Log("msg", "Series: started fanout streams", "status", strings.Join(storeDebugMsgs, ";"))
 
-	respHeap := NewDedupResponseHeap(NewProxyResponseHeap(storeResponses...))
+	var respHeap seriesStream = NewProxyResponseHeap(storeResponses...)
+	if s.enableDedup {
+		respHeap = NewDedupResponseHeap(respHeap)
+	}
 	for respHeap.Next() {
 		resp := respHeap.At()
 

pkg/store/proxy_heap.go

@@ -26,8 +26,13 @@ import (
 	grpc_opentracing "github.com/thanos-io/thanos/pkg/tracing/tracing_middleware"
 )
 
+type seriesStream interface {
+	Next() bool
+	At() *storepb.SeriesResponse
+}
+
 type dedupResponseHeap struct {
-	h *ProxyResponseHeap
+	h seriesStream
 
 	bufferedSameSeries []*storepb.SeriesResponse
 
@@ -40,7 +45,7 @@ type dedupResponseHeap struct {
 
 // NewDedupResponseHeap returns a wrapper around ProxyResponseHeap that merged duplicated series messages into one.
 // It also deduplicates identical chunks identified by the same checksum from each series message.
-func NewDedupResponseHeap(h *ProxyResponseHeap) *dedupResponseHeap {
+func NewDedupResponseHeap(h seriesStream) *dedupResponseHeap {
 	ok := h.Next()
 	var prev *storepb.SeriesResponse
 	if ok {