Disable dedup proxy in multi-tsdb

The receiver manages independent TSDBs which do not have duplicated series. For this reason it should be safe to disable deduplication of chunks and reduce CPU usage for this path. Signed-off-by: Filip Petkovski <[email protected]>
thanos-io · Oct 1, 2024 · 17a1c92 · 17a1c92
1 parent b31a637
commit 17a1c92
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 3 deletions.
diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go
@@ -323,6 +323,7 @@ func runReceive(
 
 		options := []store.ProxyStoreOption{
 			store.WithProxyStoreDebugLogging(debugLogging),
+			store.WithoutDedup(),
 		}
 
 		proxy := store.NewProxyStore(

diff --git a/pkg/store/proxy.go b/pkg/store/proxy.go
@@ -86,6 +86,7 @@ type ProxyStore struct {
 	retrievalStrategy RetrievalStrategy
 	debugLogging      bool
 	tsdbSelector      *TSDBSelector
+	enableDedup       bool
 }
 
 type proxyStoreMetrics struct {
@@ -126,6 +127,13 @@ func WithTSDBSelector(selector *TSDBSelector) 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(
@@ -156,6 +164,7 @@ func NewProxyStore(
 		metrics:           metrics,
 		retrievalStrategy: retrievalStrategy,
 		tsdbSelector:      DefaultSelector,
+		enableDedup:       true,
 	}
 
 	for _, option := range options {
@@ -307,7 +316,10 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb.
 
 	level.Debug(reqLogger).Log("msg", "Series: started fanout streams", "status", strings.Join(storeDebugMsgs, ";"))
 
-	respHeap := NewResponseDeduplicator(NewProxyResponseLoserTree(storeResponses...))
+	var respHeap seriesStream = NewProxyResponseLoserTree(storeResponses...)
+	if s.enableDedup {
+		respHeap = NewResponseDeduplicator(respHeap)
+	}
 
 	i := 0
 	for respHeap.Next() {

diff --git a/pkg/store/proxy_merge.go b/pkg/store/proxy_merge.go
@@ -26,8 +26,13 @@ import (
 	"github.com/thanos-io/thanos/pkg/tracing"
 )
 
+type seriesStream interface {
+	Next() bool
+	At() *storepb.SeriesResponse
+}
+
 type responseDeduplicator struct {
-	h *losertree.Tree[*storepb.SeriesResponse, respSet]
+	h seriesStream
 
 	bufferedSameSeries []*storepb.SeriesResponse
 
@@ -42,7 +47,7 @@ type responseDeduplicator struct {
 
 // NewResponseDeduplicator returns a wrapper around a loser tree that merges duplicated series messages into one.
 // It also deduplicates identical chunks identified by the same checksum from each series message.
-func NewResponseDeduplicator(h *losertree.Tree[*storepb.SeriesResponse, respSet]) *responseDeduplicator {
+func NewResponseDeduplicator(h seriesStream) *responseDeduplicator {
 	ok := h.Next()
 	var prev *storepb.SeriesResponse
 	if ok {