query: Support chain deduplication algorithm

Signed-off-by: Michał Mazur <[email protected]>

Author: lachruzam

CHANGELOG.md

@@ -24,6 +24,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#7353](https://github.com/thanos-io/thanos/pull/7353) [#8045](https://github.com/thanos-io/thanos/pull/8045) Receiver/StoreGateway: Add `--matcher-cache-size` option to enable caching for regex matchers in series calls.
 - [#8017](https://github.com/thanos-io/thanos/pull/8017) Store Gateway: Use native histogram for binary reader load and download duration and fixed download duration metric. #8017
 - [#8131](https://github.com/thanos-io/thanos/pull/8131) Store Gateway: Optimize regex matchers for .* and .+. #8131
+- [#7808](https://github.com/thanos-io/thanos/pull/7808) Query: Support chain deduplication algorithm.
 
 ### Changed
 

cmd/thanos/query.go

@@ -26,6 +26,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/compact/downsample"
 	"github.com/thanos-io/thanos/pkg/component"
+	"github.com/thanos-io/thanos/pkg/dedup"
 	"github.com/thanos-io/thanos/pkg/discovery/dns"
 	"github.com/thanos-io/thanos/pkg/exemplars"
 	"github.com/thanos-io/thanos/pkg/extkingpin"
@@ -98,6 +99,11 @@ func registerQuery(app *extkingpin.App) {
 		Default(string(query.ExternalLabels), string(query.StoreType)).
 		Enums(string(query.ExternalLabels), string(query.StoreType))
 
+	deduplicationFunc := cmd.Flag("deduplication.func", "Experimental. Deduplication algorithm for merging overlapping series. "+
+		"Possible values are: \"penalty\", \"chain\". If no value is specified, penalty based deduplication algorithm will be used. "+
+		"When set to chain, the default compact deduplication merger is used, which performs 1:1 deduplication for samples. At least one replica label has to be set via --query.replica-label flag.").
+		Default(dedup.AlgorithmPenalty).Enum(dedup.AlgorithmPenalty, dedup.AlgorithmChain)
+
 	queryReplicaLabels := cmd.Flag("query.replica-label", "Labels to treat as a replica indicator along which data is deduplicated. Still you will be able to query without deduplication using 'dedup=false' parameter. Data includes time series, recording rules, and alerting rules. Flag may be specified multiple times as well as a comma separated list of labels.").
 		Strings()
 	queryPartitionLabels := cmd.Flag("query.partition-label", "Labels that partition the leaf queriers. This is used to scope down the labelsets of leaf queriers when using the distributed query mode. If set, these labels must form a partition of the leaf queriers. Partition labels must not intersect with replica labels. Every TSDB of a leaf querier must have these labels. This is useful when there are multiple external labels that are irrelevant for the partition as it allows the distributed engine to ignore them for some optimizations. If this is empty then all labels are used as partition labels.").Strings()
@@ -300,6 +306,7 @@ func registerQuery(app *extkingpin.App) {
 			*dynamicLookbackDelta,
 			time.Duration(*defaultEvaluationInterval),
 			time.Duration(*storeResponseTimeout),
+			*deduplicationFunc,
 			*queryReplicaLabels,
 			*queryPartitionLabels,
 			selectorLset,
@@ -361,6 +368,7 @@ func runQuery(
 	dynamicLookbackDelta bool,
 	defaultEvaluationInterval time.Duration,
 	storeResponseTimeout time.Duration,
+	deduplicationFunc string,
 	queryReplicaLabels []string,
 	queryPartitionLabels []string,
 	selectorLset labels.Labels,
@@ -422,6 +430,7 @@ func runQuery(
 			seriesProxy,
 			maxConcurrentSelects,
 			queryTimeout,
+			deduplicationFunc,
 		)
 		remoteEndpointsCreator = query.NewRemoteEndpointsCreator(
 			logger,

docs/components/query.md

@@ -103,6 +103,15 @@ thanos query \
 
 This logic can also be controlled via parameter on QueryAPI. More details below.
 
+### Deduplication Algorithms
+
+Thanos Querier supports different algorithms for deduplicating overlapping series. You can choose the deduplication algorithm using the `--deduplication.func` flag. The available options are:
+
+* `penalty` (default): This is the default deduplication algorithm used by Thanos. It fills gaps only after a certain penalty window. This helps avoid flapping between replicas due to minor differences or delays.
+* `chain`: This algorithm performs 1:1 deduplication for samples. It merges all available data points from the replicas without applying any penalty. This is useful in deployments based on receivers, where each replica is populated by the same data. In such cases, using the penalty algorithm may cause gaps even when data is available in other replicas.
+
+Note that deduplication of HA groups is not supported by the `chain` algorithm.
+
 ## Thanos PromQL Engine (experimental)
 
 By default, Thanos querier comes with standard Prometheus PromQL engine. However, when `--query.promql-engine=thanos` is specified, Thanos will use [experimental Thanos PromQL engine](http://github.com/thanos-community/promql-engine) which is a drop-in, efficient implementation of PromQL engine with query planner and optimizers.
@@ -297,6 +306,16 @@ Flags:
       --auto-gomemlimit.ratio=0.9
                                  The ratio of reserved GOMEMLIMIT memory to the
                                  detected maximum container or system memory.
+      --deduplication.func=penalty
+                                 Experimental. Deduplication algorithm for
+                                 merging overlapping series. Possible values
+                                 are: "penalty", "chain". If no value is
+                                 specified, penalty based deduplication
+                                 algorithm will be used. When set to chain, the
+                                 default compact deduplication merger is used,
+                                 which performs 1:1 deduplication for samples.
+                                 At least one replica label has to be set via
+                                 --query.replica-label flag.
       --enable-auto-gomemlimit   Enable go runtime to automatically limit memory
                                  consumption.
       --endpoint=<endpoint> ...  (Deprecated): Addresses of statically

pkg/api/query/grpc_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/thanos-io/thanos/pkg/api/query/querypb"
 	"github.com/thanos-io/thanos/pkg/component"
+	"github.com/thanos-io/thanos/pkg/dedup"
 	"github.com/thanos-io/thanos/pkg/extpromql"
 	"github.com/thanos-io/thanos/pkg/query"
 	"github.com/thanos-io/thanos/pkg/store"
@@ -31,7 +32,7 @@ func TestGRPCQueryAPIWithQueryPlan(t *testing.T) {
 	logger := log.NewNopLogger()
 	reg := prometheus.NewRegistry()
 	proxy := store.NewProxyStore(logger, reg, func() []store.Client { return nil }, component.Store, nil, 1*time.Minute, store.LazyRetrieval)
-	queryableCreator := query.NewQueryableCreator(logger, reg, proxy, 1, 1*time.Minute)
+	queryableCreator := query.NewQueryableCreator(logger, reg, proxy, 1, 1*time.Minute, dedup.AlgorithmPenalty)
 	remoteEndpointsCreator := query.NewRemoteEndpointsCreator(logger, func() []query.Client { return nil }, nil, 1*time.Minute, true, true)
 	lookbackDeltaFunc := func(i int64) time.Duration { return 5 * time.Minute }
 	api := NewGRPCAPI(time.Now, nil, queryableCreator, remoteEndpointsCreator, queryFactory, querypb.EngineType_thanos, lookbackDeltaFunc, 0)
@@ -75,7 +76,7 @@ func TestGRPCQueryAPIErrorHandling(t *testing.T) {
 	logger := log.NewNopLogger()
 	reg := prometheus.NewRegistry()
 	proxy := store.NewProxyStore(logger, reg, func() []store.Client { return nil }, component.Store, nil, 1*time.Minute, store.LazyRetrieval)
-	queryableCreator := query.NewQueryableCreator(logger, reg, proxy, 1, 1*time.Minute)
+	queryableCreator := query.NewQueryableCreator(logger, reg, proxy, 1, 1*time.Minute, dedup.AlgorithmPenalty)
 	remoteEndpointsCreator := query.NewRemoteEndpointsCreator(logger, func() []query.Client { return nil }, nil, 1*time.Minute, true, true)
 	lookbackDeltaFunc := func(i int64) time.Duration { return 5 * time.Minute }
 	tests := []struct {

pkg/api/query/v1_test.go

@@ -53,6 +53,7 @@ import (
 	baseAPI "github.com/thanos-io/thanos/pkg/api"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/component"
+	"github.com/thanos-io/thanos/pkg/dedup"
 	"github.com/thanos-io/thanos/pkg/gate"
 	"github.com/thanos-io/thanos/pkg/query"
 	"github.com/thanos-io/thanos/pkg/rules/rulespb"
@@ -213,7 +214,7 @@ func TestQueryEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout, dedup.AlgorithmPenalty),
 		remoteEndpointsCreate: emptyRemoteEndpointsCreate,
 		queryCreate:           queryFactory,
 		lookbackDeltaCreate:   func(m int64) time.Duration { return time.Duration(0) },
@@ -661,7 +662,7 @@ func TestQueryExplainEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout, dedup.AlgorithmPenalty),
 		remoteEndpointsCreate: emptyRemoteEndpointsCreate,
 		queryCreate:           queryFactory,
 		defaultEngine:         PromqlEnginePrometheus,
@@ -725,7 +726,7 @@ func TestQueryAnalyzeEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout, dedup.AlgorithmPenalty),
 		remoteEndpointsCreate: emptyRemoteEndpointsCreate,
 		queryCreate:           queryFactory,
 		defaultEngine:         PromqlEnginePrometheus,
@@ -894,7 +895,7 @@ func TestMetadataEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout, dedup.AlgorithmPenalty),
 		remoteEndpointsCreate: emptyRemoteEndpointsCreate,
 		queryCreate:           queryFactory,
 		defaultEngine:         PromqlEnginePrometheus,
@@ -911,7 +912,7 @@ func TestMetadataEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:          query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:          query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout, dedup.AlgorithmPenalty),
 		remoteEndpointsCreate:    emptyRemoteEndpointsCreate,
 		queryCreate:              queryFactory,
 		defaultEngine:            PromqlEnginePrometheus,

pkg/dedup/iter.go

@@ -15,9 +15,15 @@ import (
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 )
 
+const (
+	AlgorithmPenalty = "penalty"
+	AlgorithmChain   = "chain"
+)
+
 type dedupSeriesSet struct {
-	set       storage.SeriesSet
-	isCounter bool
+	set               storage.SeriesSet
+	isCounter         bool
+	deduplicationFunc string
 
 	replicas []storage.Series
 
@@ -102,9 +108,9 @@ func (o *overlapSplitSet) Err() error {
 
 // NewSeriesSet returns seriesSet that deduplicates the same series.
 // The series in series set are expected be sorted by all labels.
-func NewSeriesSet(set storage.SeriesSet, f string) storage.SeriesSet {
+func NewSeriesSet(set storage.SeriesSet, f string, deduplicationFunc string) storage.SeriesSet {
 	// TODO: remove dependency on knowing whether it is a counter.
-	s := &dedupSeriesSet{set: set, isCounter: isCounter(f), f: f}
+	s := &dedupSeriesSet{set: set, isCounter: isCounter(f), f: f, deduplicationFunc: deduplicationFunc}
 	s.ok = s.set.Next()
 	if s.ok {
 		s.peek = s.set.At()
@@ -154,7 +160,11 @@ func (s *dedupSeriesSet) At() storage.Series {
 	repl := make([]storage.Series, len(s.replicas))
 	copy(repl, s.replicas)
 
-	return newDedupSeries(s.lset, repl, s.f)
+	if s.deduplicationFunc == AlgorithmChain {
+		return seriesWithLabels{Series: storage.ChainedSeriesMerge(repl...), lset: s.lset}
+	} else {
+		return newDedupSeries(s.lset, repl, s.f)
+	}
 }
 
 func (s *dedupSeriesSet) Err() error {

pkg/dedup/iter_test.go

@@ -544,7 +544,151 @@ func TestDedupSeriesSet(t *testing.T) {
 			if tcase.isCounter {
 				f = "rate"
 			}
-			dedupSet := NewSeriesSet(&mockedSeriesSet{series: tcase.input}, f)
+			dedupSet := NewSeriesSet(&mockedSeriesSet{series: tcase.input}, f, AlgorithmPenalty)
+			var ats []storage.Series
+			for dedupSet.Next() {
+				ats = append(ats, dedupSet.At())
+			}
+			testutil.Ok(t, dedupSet.Err())
+			testutil.Equals(t, len(tcase.exp), len(ats))
+
+			for i, s := range ats {
+				testutil.Equals(t, tcase.exp[i].lset, s.Labels(), "labels mismatch for series %v", i)
+				res := expandSeries(t, s.Iterator(nil))
+				testutil.Equals(t, tcase.exp[i].samples, res, "values mismatch for series :%v", i)
+			}
+		})
+	}
+}
+
+func TestDedupSeriesSet_Chain(t *testing.T) {
+	for _, tcase := range []struct {
+		name  string
+		input []series
+		exp   []series
+	}{
+		{
+			name: "Single dedup label - exact match",
+			input: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+			},
+			exp: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+			},
+		},
+		{
+			name: "Single dedup label - gap in one series",
+			input: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+			},
+			exp: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+			},
+		},
+		{
+			name: "Single dedup label - gaps in two series",
+			input: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}},
+				},
+			},
+			exp: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+			},
+		},
+		{
+			name: "Multi dedup label - exact match",
+			input: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {20000, 102}, {30000, 103}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {20000, 102}, {30000, 103}},
+				},
+			},
+			exp: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {20000, 102}, {30000, 103}},
+				},
+			},
+		},
+		{
+			name: "Multi dedup label - gaps in two series",
+			input: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {20000, 102}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {30000, 103}},
+				},
+			},
+			exp: []series{
+				{
+					lset:    labels.FromStrings("a", "1", "c", "3"),
+					samples: []sample{{10000, 1}, {20000, 2}, {30000, 3}},
+				},
+				{
+					lset:    labels.FromStrings("a", "2", "c", "3"),
+					samples: []sample{{10000, 101}, {20000, 102}, {30000, 103}},
+				},
+			},
+		},
+	} {
+		t.Run(tcase.name, func(t *testing.T) {
+			dedupSet := NewSeriesSet(&mockedSeriesSet{series: tcase.input}, "", AlgorithmChain)
 			var ats []storage.Series
 			for dedupSet.Next() {
 				ats = append(ats, dedupSet.At())