From 020cb44770a914109b8de1aaf661d9d519d678dd Mon Sep 17 00:00:00 2001 From: Yuchen Wang Date: Thu, 22 Jan 2026 13:02:13 -0800 Subject: [PATCH 1/3] add new group replica response strategy --- cmd/thanos/query.go | 23 +++++ cmd/thanos/rule.go | 1 + pkg/query/endpointset.go | 85 +++++++--------- pkg/query/endpointset_test.go | 54 ++++------ pkg/query/querier.go | 47 +++++++-- pkg/query/querier_test.go | 79 +++++++++++++++ pkg/store/proxy.go | 182 +++++++++++++++++++++++++++++----- pkg/store/proxy_test.go | 144 +++++++++++++++++++++++++++ pr.md | 105 ++++++++++++++++++++ 9 files changed, 608 insertions(+), 112 deletions(-) create mode 100644 pr.md diff --git a/cmd/thanos/query.go b/cmd/thanos/query.go index f82ed44963b..892b2ed7629 100644 --- a/cmd/thanos/query.go +++ b/cmd/thanos/query.go @@ -200,6 +200,12 @@ func registerQuery(app *extkingpin.App) { enableGroupReplicaPartialStrategy := cmd.Flag("query.group-replica-strategy", "Enable group-replica partial response strategy."). Default("false").Bool() + groupReplicaGroupLabel := cmd.Flag("query.group-replica.group-label", "External label name that identifies the group for group-replica partial response strategy. Stores with the same group label value hold replicated data. Must be set together with --query.group-replica.quorum-label."). + Default("").String() + + groupReplicaQuorumLabel := cmd.Flag("query.group-replica.quorum-label", "External label name whose value specifies the minimum number of healthy stores required per group. Must be set together with --query.group-replica.group-label. Stores without these labels or with invalid quorum values (<1) are treated as must-success stores."). + Default("").String() + enableRulePartialResponse := cmd.Flag("rule.partial-response", "Enable partial response for rules endpoint. --no-rule.partial-response for disabling."). Hidden().Default("true").Bool() @@ -268,6 +274,11 @@ func registerQuery(app *extkingpin.App) { return errors.Wrap(err, "parse federation labels") } + // Validate group-replica label flags: both must be set together or neither. + if (*groupReplicaGroupLabel == "") != (*groupReplicaQuorumLabel == "") { + return errors.New("--query.group-replica.group-label and --query.group-replica.quorum-label must be set together or neither") + } + for _, feature := range *featureList { if feature == promqlAtModifier { level.Warn(logger).Log("msg", "This option for --enable-feature is now permanently enabled and therefore a no-op.", "option", promqlAtModifier) @@ -408,6 +419,8 @@ func registerQuery(app *extkingpin.App) { *enforceTenancy, *tenantLabel, *enableGroupReplicaPartialStrategy, + *groupReplicaGroupLabel, + *groupReplicaQuorumLabel, *rewriteAggregationLabelStrategy, *rewriteAggregationLabelTo, *lazyRetrievalMaxBufferedResponses, @@ -499,6 +512,8 @@ func runQuery( enforceTenancy bool, tenantLabel string, groupReplicaPartialResponseStrategy bool, + groupReplicaGroupLabel string, + groupReplicaQuorumLabel string, rewriteAggregationLabelStrategy string, rewriteAggregationLabelTo string, lazyRetrievalMaxBufferedResponses int, @@ -612,6 +627,9 @@ func runQuery( if len(exclusiveExternalLabels) > 0 { options = append(options, store.WithExclusiveExternalLabels(exclusiveExternalLabels)) } + if groupReplicaGroupLabel != "" && groupReplicaQuorumLabel != "" { + options = append(options, store.WithGroupReplicaLabels(groupReplicaGroupLabel, groupReplicaQuorumLabel)) + } // Parse and sanitize the provided replica labels flags. queryReplicaLabels = strutil.ParseFlagLabels(queryReplicaLabels) @@ -639,6 +657,7 @@ func runQuery( endpointInfoTimeout, // ignoreErrors when group_replica partial response strategy is enabled. groupReplicaPartialResponseStrategy, + groupReplicaQuorumLabel, queryConnMetricLabels..., ) @@ -652,6 +671,8 @@ func runQuery( ) opts := query.Options{ GroupReplicaPartialResponseStrategy: groupReplicaPartialResponseStrategy, + GroupReplicaGroupLabel: groupReplicaGroupLabel, + GroupReplicaQuorumLabel: groupReplicaQuorumLabel, DeduplicationFunc: queryDeduplicationFunc, RewriteAggregationLabelStrategy: rewriteAggregationLabelStrategy, RewriteAggregationLabelTo: rewriteAggregationLabelTo, @@ -966,6 +987,7 @@ func prepareEndpointSet( unhealthyStoreTimeout time.Duration, endpointInfoTimeout time.Duration, ignoreStoreErrors bool, + quorumLabelName string, queryConnMetricLabels ...string, ) *query.EndpointSet { endpointSet := query.NewEndpointSet( @@ -1011,6 +1033,7 @@ func prepareEndpointSet( dialOpts, unhealthyStoreTimeout, endpointInfoTimeout, + quorumLabelName, queryConnMetricLabels..., ) diff --git a/cmd/thanos/rule.go b/cmd/thanos/rule.go index 41d996e0200..16235c2894e 100644 --- a/cmd/thanos/rule.go +++ b/cmd/thanos/rule.go @@ -444,6 +444,7 @@ func runRule( 5*time.Minute, 5*time.Second, false, + "", // quorumLabelName - not used in rule component ) // Periodically update the GRPC addresses from query config by resolving them using DNS SD if necessary. diff --git a/pkg/query/endpointset.go b/pkg/query/endpointset.go index 02c9f096b4d..fc25e0f1d90 100644 --- a/pkg/query/endpointset.go +++ b/pkg/query/endpointset.go @@ -10,7 +10,6 @@ import ( "math" "regexp" "sort" - "strings" "sync" "time" "unicode/utf8" @@ -155,15 +154,13 @@ type endpointSetNodeCollector struct { mtx sync.Mutex storeNodes map[string]map[string]int storePerExtLset map[string]int - storeNodesAddr map[string]map[string]int - storeNodesKeys map[string]map[string]int + storesByQuorum map[string]int logger log.Logger connectionsDesc *prometheus.Desc labels []string - connectionsWithAddr *prometheus.Desc - connectionsWithKeys *prometheus.Desc + endpointGroupsDesc *prometheus.Desc } func newEndpointSetNodeCollector(logger log.Logger, labels ...string) *endpointSetNodeCollector { @@ -172,23 +169,19 @@ func newEndpointSetNodeCollector(logger log.Logger, labels ...string) *endpointS } desc := "Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier." return &endpointSetNodeCollector{ - logger: logger, - storeNodes: map[string]map[string]int{}, + logger: logger, + storeNodes: map[string]map[string]int{}, + storesByQuorum: map[string]int{}, connectionsDesc: prometheus.NewDesc( "thanos_store_nodes_grpc_connections", desc, labels, nil, ), labels: labels, - connectionsWithAddr: prometheus.NewDesc( - "thanos_store_nodes_grpc_connections_addr", - desc, - []string{string(ReplicaKey), "addr"}, nil, - ), - connectionsWithKeys: prometheus.NewDesc( - "thanos_store_nodes_grpc_connections_keys", - desc, - []string{string(GroupKey), string(ReplicaKey)}, nil, + endpointGroupsDesc: prometheus.NewDesc( + "thanos_query_endpoint_groups", + "Number of discovered store API endpoints by quorum label value. Endpoints without quorum label have quorum=0.", + []string{"quorum"}, nil, ), } } @@ -208,8 +201,7 @@ func truncateExtLabels(s string, threshold int) string { } func (c *endpointSetNodeCollector) Update( nodes map[string]map[string]int, - nodesAddr map[string]map[string]int, - nodesKeys map[string]map[string]int, + storesByQuorum map[string]int, ) { storeNodes := make(map[string]map[string]int, len(nodes)) storePerExtLset := map[string]int{} @@ -227,14 +219,12 @@ func (c *endpointSetNodeCollector) Update( defer c.mtx.Unlock() c.storeNodes = storeNodes c.storePerExtLset = storePerExtLset - c.storeNodesAddr = nodesAddr - c.storeNodesKeys = nodesKeys + c.storesByQuorum = storesByQuorum } func (c *endpointSetNodeCollector) Describe(ch chan<- *prometheus.Desc) { ch <- c.connectionsDesc - ch <- c.connectionsWithAddr - ch <- c.connectionsWithKeys + ch <- c.endpointGroupsDesc } func (c *endpointSetNodeCollector) Collect(ch chan<- prometheus.Metric) { @@ -261,21 +251,11 @@ func (c *endpointSetNodeCollector) Collect(ch chan<- prometheus.Metric) { } } } - for replicaKey, occurrencesPerAddr := range c.storeNodesAddr { - for addr, occurrences := range occurrencesPerAddr { - ch <- prometheus.MustNewConstMetric( - c.connectionsWithAddr, prometheus.GaugeValue, - float64(occurrences), - replicaKey, addr) - } - } - for groupKey, occurrencesPerReplicaKey := range c.storeNodesKeys { - for replicaKeys, occurrences := range occurrencesPerReplicaKey { - ch <- prometheus.MustNewConstMetric( - c.connectionsWithKeys, prometheus.GaugeValue, - float64(occurrences), - groupKey, replicaKeys) - } + for quorum, count := range c.storesByQuorum { + ch <- prometheus.MustNewConstMetric( + c.endpointGroupsDesc, prometheus.GaugeValue, + float64(count), + quorum) } } @@ -292,6 +272,10 @@ type EndpointSet struct { endpointInfoTimeout time.Duration unhealthyEndpointTimeout time.Duration + // quorumLabelName is the external label name whose value specifies the quorum requirement. + // Used for the thanos_query_endpoint_groups metric. + quorumLabelName string + updateMtx sync.Mutex endpointsMtx sync.RWMutex @@ -314,6 +298,7 @@ func NewEndpointSet( dialOpts []grpc.DialOption, unhealthyEndpointTimeout time.Duration, endpointInfoTimeout time.Duration, + quorumLabelName string, endpointMetricLabels ...string, ) *EndpointSet { endpointsMetric := newEndpointSetNodeCollector(logger, endpointMetricLabels...) @@ -337,6 +322,7 @@ func NewEndpointSet( dialOpts: dialOpts, endpointInfoTimeout: endpointInfoTimeout, unhealthyEndpointTimeout: unhealthyEndpointTimeout, + quorumLabelName: quorumLabelName, endpointSpec: func() map[string]*GRPCEndpointSpec { specs := make(map[string]*GRPCEndpointSpec) for _, s := range endpointSpecs() { @@ -443,14 +429,7 @@ func (e *EndpointSet) Update(ctx context.Context) { // Update stats. stats := newEndpointAPIStats() - statsAddr := make(map[string]map[string]int) - statsKeys := make(map[string]map[string]int) - bumpCounter := func(key1, key2 string, mp map[string]map[string]int) { - if _, ok := mp[key1]; !ok { - mp[key1] = make(map[string]int) - } - mp[key1][key2]++ - } + storesByQuorum := make(map[string]int) for addr, er := range e.endpoints { if !er.isQueryable() { continue @@ -466,11 +445,21 @@ func (e *EndpointSet) Update(ctx context.Context) { "address", addr, "extLset", extLset, "duplicates", fmt.Sprintf("%v", stats[component.Sidecar.String()][extLset]+stats[component.Rule.String()][extLset]+1)) } stats[er.ComponentType().String()][extLset]++ - bumpCounter(er.replicaKey, strings.Split(addr, ":")[0], statsAddr) - bumpCounter(er.groupKey, er.replicaKey, statsKeys) + + // Track endpoints by quorum label value for thanos_query_endpoint_groups metric. + quorumValue := "0" // Default for endpoints without quorum label + if e.quorumLabelName != "" { + for _, lset := range er.LabelSets() { + if v := lset.Get(e.quorumLabelName); v != "" { + quorumValue = v + break + } + } + } + storesByQuorum[quorumValue]++ } - e.endpointsMetric.Update(stats, statsAddr, statsKeys) + e.endpointsMetric.Update(stats, storesByQuorum) } func (e *EndpointSet) updateEndpoint(ctx context.Context, spec *GRPCEndpointSpec, er *endpointRef) { diff --git a/pkg/query/endpointset_test.go b/pkg/query/endpointset_test.go index 04a4033a0d2..551f701d1ce 100644 --- a/pkg/query/endpointset_test.go +++ b/pkg/query/endpointset_test.go @@ -243,18 +243,14 @@ func TestTruncateExtLabels(t *testing.T) { func TestEndpointSetUpdate(t *testing.T) { t.Parallel() + const metricsMetaEndpointGroups = ` + # HELP thanos_query_endpoint_groups Number of discovered store API endpoints by quorum label value. Endpoints without quorum label have quorum=0. + # TYPE thanos_query_endpoint_groups gauge + ` const metricsMeta = ` # HELP thanos_store_nodes_grpc_connections Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier. # TYPE thanos_store_nodes_grpc_connections gauge ` - const metricsMetaAddr = ` - # HELP thanos_store_nodes_grpc_connections_addr Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier. - # TYPE thanos_store_nodes_grpc_connections_addr gauge - ` - const metricsMetaKeys = ` - # HELP thanos_store_nodes_grpc_connections_keys Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier. - # TYPE thanos_store_nodes_grpc_connections_keys gauge - ` testCases := []struct { name string endpoints []testEndpointMeta @@ -279,15 +275,12 @@ func TestEndpointSetUpdate(t *testing.T) { connLabels: []string{"store_type"}, expectedEndpoints: 1, - expectedConnMetrics: metricsMeta + + expectedConnMetrics: metricsMetaEndpointGroups + ` - thanos_store_nodes_grpc_connections{store_type="sidecar"} 1 - ` + metricsMetaAddr + - ` - thanos_store_nodes_grpc_connections_addr{addr="127.0.0.1",replica_key=""} 1 - ` + metricsMetaKeys + + thanos_query_endpoint_groups{quorum="0"} 1 + ` + metricsMeta + ` - thanos_store_nodes_grpc_connections_keys{group_key="",replica_key=""} 1 + thanos_store_nodes_grpc_connections{store_type="sidecar"} 1 `, }, { @@ -339,15 +332,12 @@ func TestEndpointSetUpdate(t *testing.T) { strict: true, connLabels: []string{"store_type"}, expectedEndpoints: 1, - expectedConnMetrics: metricsMeta + + expectedConnMetrics: metricsMetaEndpointGroups + ` - thanos_store_nodes_grpc_connections{store_type="sidecar"} 1 - ` + metricsMetaAddr + - ` - thanos_store_nodes_grpc_connections_addr{addr="127.0.0.1",replica_key=""} 1 - ` + metricsMetaKeys + + thanos_query_endpoint_groups{quorum="0"} 1 + ` + metricsMeta + ` - thanos_store_nodes_grpc_connections_keys{group_key="",replica_key=""} 1 + thanos_store_nodes_grpc_connections{store_type="sidecar"} 1 `, }, { @@ -369,14 +359,10 @@ func TestEndpointSetUpdate(t *testing.T) { }, }, expectedEndpoints: 1, - expectedConnMetrics: metricsMeta + ` + expectedConnMetrics: metricsMetaEndpointGroups + ` + thanos_query_endpoint_groups{quorum="0"} 1 + ` + metricsMeta + ` thanos_store_nodes_grpc_connections{external_labels="{lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val}",store_type="sidecar"} 1 - ` + metricsMetaAddr + - ` - thanos_store_nodes_grpc_connections_addr{addr="127.0.0.1",replica_key=""} 1 - ` + metricsMetaKeys + - ` - thanos_store_nodes_grpc_connections_keys{group_key="",replica_key=""} 1 `, }, } @@ -705,7 +691,7 @@ func TestEndpointSetUpdate_AvailabilityScenarios(t *testing.T) { } return specs }, - testGRPCOpts, time.Minute, 2*time.Second) + testGRPCOpts, time.Minute, 2*time.Second, "") defer endpointSet.Close() // Initial update. @@ -1078,7 +1064,7 @@ func TestEndpointSet_Update_NoneAvailable(t *testing.T) { } return specs }, - testGRPCOpts, time.Minute, 2*time.Second) + testGRPCOpts, time.Minute, 2*time.Second, "") defer endpointSet.Close() // Should not matter how many of these we run. @@ -1189,7 +1175,7 @@ func TestEndpoint_Update_QuerierStrict(t *testing.T) { NewGRPCEndpointSpec(discoveredEndpointAddr[1], false), NewGRPCEndpointSpec(discoveredEndpointAddr[2], true), } - }, testGRPCOpts, time.Minute, 1*time.Second) + }, testGRPCOpts, time.Minute, 1*time.Second, "") defer endpointSet.Close() // Initial update. @@ -1370,7 +1356,7 @@ func TestEndpointSet_APIs_Discovery(t *testing.T) { return tc.states[currentState].endpointSpec() }, - testGRPCOpts, time.Minute, 2*time.Second) + testGRPCOpts, time.Minute, 2*time.Second, "") defer endpointSet.Close() @@ -1562,7 +1548,7 @@ func makeEndpointSet(discoveredEndpointAddr []string, strict bool, now nowFunc, } return specs }, - testGRPCOpts, time.Minute, time.Second, metricLabels...) + testGRPCOpts, time.Minute, time.Second, "", metricLabels...) return endpointSet } diff --git a/pkg/query/querier.go b/pkg/query/querier.go index 64bab65a022..8ca73c2a3b1 100644 --- a/pkg/query/querier.go +++ b/pkg/query/querier.go @@ -59,6 +59,8 @@ type QueryableCreator func( type Options struct { GroupReplicaPartialResponseStrategy bool + GroupReplicaGroupLabel string + GroupReplicaQuorumLabel string DeduplicationFunc string RewriteAggregationLabelStrategy string RewriteAggregationLabelTo string @@ -170,6 +172,12 @@ type querier struct { shardInfo *storepb.ShardInfo seriesStatsReporter seriesStatsReporter + // groupReplicaGroupLabel and groupReplicaQuorumLabel are the external label names + // used for label-based group/quorum identification. These labels are stripped from + // query results when set. + groupReplicaGroupLabel string + groupReplicaQuorumLabel string + aggregationLabelRewriter *AggregationLabelRewriter } @@ -248,6 +256,8 @@ func newQuerierWithOpts( skipChunks: skipChunks, shardInfo: shardInfo, seriesStatsReporter: seriesStatsReporter, + groupReplicaGroupLabel: opts.GroupReplicaGroupLabel, + groupReplicaQuorumLabel: opts.GroupReplicaQuorumLabel, aggregationLabelRewriter: aggregationLabelRewriter, } @@ -257,6 +267,28 @@ func (q *querier) isDedupEnabled() bool { return q.deduplicate && len(q.replicaLabels) > 0 } +// getWithoutReplicaLabels returns the list of labels to strip from query results. +// This includes replica labels for deduplication and group/quorum labels for the +// GROUP_REPLICA partial response strategy. +func (q *querier) getWithoutReplicaLabels() []string { + var labels []string + + // Add replica labels if deduplication is enabled + if q.isDedupEnabled() { + labels = append(labels, q.replicaLabels...) + } + + // Add group/quorum labels if configured (for GROUP_REPLICA strategy) + if q.groupReplicaGroupLabel != "" { + labels = append(labels, q.groupReplicaGroupLabel) + } + if q.groupReplicaQuorumLabel != "" { + labels = append(labels, q.groupReplicaQuorumLabel) + } + + return labels +} + type seriesServer struct { // This field just exist to pseudo-implement the unused methods of the interface. storepb.Store_SeriesServer @@ -411,9 +443,10 @@ func (q *querier) selectFn(ctx context.Context, hints *storage.SelectHints, ms . PartialResponseStrategy: q.partialResponseStrategy, SkipChunks: q.skipChunks, } - if q.isDedupEnabled() { - // Soft ask to sort without replica labels and push them at the end of labelset. - req.WithoutReplicaLabels = q.replicaLabels + // Soft ask to sort without replica labels and push them at the end of labelset. + // This includes dedup replica labels and group/replica labels for the GROUP_REPLICA strategy. + if labels := q.getWithoutReplicaLabels(); len(labels) > 0 { + req.WithoutReplicaLabels = labels } if err := q.proxy.Series(&req, resp); err != nil { @@ -471,8 +504,8 @@ func (q *querier) LabelValues(ctx context.Context, name string, hints *storage.L Limit: int64(hints.Limit), } - if q.isDedupEnabled() { - req.WithoutReplicaLabels = q.replicaLabels + if labels := q.getWithoutReplicaLabels(); len(labels) > 0 { + req.WithoutReplicaLabels = labels } resp, err := q.proxy.LabelValues(ctx, req) @@ -514,8 +547,8 @@ func (q *querier) LabelNames(ctx context.Context, hints *storage.LabelHints, mat Limit: int64(hints.Limit), } - if q.isDedupEnabled() { - req.WithoutReplicaLabels = q.replicaLabels + if labels := q.getWithoutReplicaLabels(); len(labels) > 0 { + req.WithoutReplicaLabels = labels } resp, err := q.proxy.LabelNames(ctx, req) diff --git a/pkg/query/querier_test.go b/pkg/query/querier_test.go index 8c77ee93c50..9fd1540a4ec 100644 --- a/pkg/query/querier_test.go +++ b/pkg/query/querier_test.go @@ -1279,3 +1279,82 @@ func storeSeriesResponse(t testing.TB, lset labels.Labels, smplChunks ...[]sampl } return storepb.NewSeriesResponse(&s) } + +func TestQuerier_GetWithoutReplicaLabels(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + deduplicate bool + replicaLabels []string + groupReplicaGroupLabel string + groupReplicaQuorumLabel string + expectedLabels []string + }{ + { + name: "no dedup, no group replica labels - empty result", + deduplicate: false, + replicaLabels: []string{"replica"}, + groupReplicaGroupLabel: "", + groupReplicaQuorumLabel: "", + expectedLabels: nil, + }, + { + name: "dedup enabled - only replica labels", + deduplicate: true, + replicaLabels: []string{"replica", "prometheus_replica"}, + groupReplicaGroupLabel: "", + groupReplicaQuorumLabel: "", + expectedLabels: []string{"replica", "prometheus_replica"}, + }, + { + name: "no dedup, group replica labels configured - only group and quorum labels", + deduplicate: false, + replicaLabels: []string{"replica"}, + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + expectedLabels: []string{"receive_group", "quorum"}, + }, + { + name: "dedup enabled and group replica labels - both combined", + deduplicate: true, + replicaLabels: []string{"replica"}, + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + expectedLabels: []string{"replica", "receive_group", "quorum"}, + }, + { + name: "dedup enabled but no replica labels - only group and quorum labels", + deduplicate: true, + replicaLabels: []string{}, + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + expectedLabels: []string{"receive_group", "quorum"}, + }, + { + name: "only group label configured (partial) - only group label", + deduplicate: false, + replicaLabels: []string{}, + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "", + expectedLabels: []string{"receive_group"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + q := &querier{ + deduplicate: tc.deduplicate, + replicaLabels: tc.replicaLabels, + groupReplicaGroupLabel: tc.groupReplicaGroupLabel, + groupReplicaQuorumLabel: tc.groupReplicaQuorumLabel, + } + + got := q.getWithoutReplicaLabels() + + if tc.expectedLabels == nil { + testutil.Equals(t, 0, len(got)) + } else { + testutil.Equals(t, tc.expectedLabels, got) + } + }) + } +} diff --git a/pkg/store/proxy.go b/pkg/store/proxy.go index 434cb220e9b..7f4782c202b 100644 --- a/pkg/store/proxy.go +++ b/pkg/store/proxy.go @@ -7,6 +7,7 @@ import ( "context" "fmt" "math" + "strconv" "strings" "sync" "time" @@ -111,6 +112,12 @@ type ProxyStore struct { forwardPartialStrategy bool exclusiveExternalLabels []string matcherCache storecache.MatchersCache + + // groupReplicaGroupLabel and groupReplicaQuorumLabel are the external label names + // used for label-based group/quorum identification in GROUP_REPLICA partial response strategy. + // When both are set, the strategy uses these labels from LabelSets() instead of DNS-based parsing. + groupReplicaGroupLabel string + groupReplicaQuorumLabel string } type proxyStoreMetrics struct { @@ -265,6 +272,15 @@ func WithMatcherCache(cache storecache.MatchersCache) ProxyStoreOption { } } +// WithGroupReplicaLabels sets the external label names for label-based group/quorum +// identification in GROUP_REPLICA partial response strategy. +func WithGroupReplicaLabels(groupLabel, quorumLabel string) ProxyStoreOption { + return func(s *ProxyStore) { + s.groupReplicaGroupLabel = groupLabel + s.groupReplicaQuorumLabel = quorumLabel + } +} + // 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( @@ -306,6 +322,43 @@ func NewProxyStore( return s } +// getGroupKey returns the group key for a store client. +// If groupReplicaGroupLabel is configured, it looks up the label value from LabelSets(). +// Otherwise, it falls back to the DNS-based GroupKey() method. +// For stores without the configured label, it returns empty string to indicate a must-success store. +func (s *ProxyStore) getGroupKey(st Client) string { + if s.groupReplicaGroupLabel != "" { + for _, lset := range st.LabelSets() { + if v := lset.Get(s.groupReplicaGroupLabel); v != "" { + return v + } + } + // Store doesn't have the label - return empty to indicate must-success store + return "" + } + return st.GroupKey() +} + +// getQuorum returns the quorum requirement for a store client. +// If groupReplicaQuorumLabel is configured, it parses the label value as an integer. +// Returns 0 if the label is missing, empty, or invalid (non-integer or <1). +// A return value of 0 indicates the store should be treated as a "must-success" store. +func (s *ProxyStore) getQuorum(st Client) int { + if s.groupReplicaQuorumLabel == "" { + return 0 + } + for _, lset := range st.LabelSets() { + if v := lset.Get(s.groupReplicaQuorumLabel); v != "" { + quorum, err := strconv.Atoi(v) + if err != nil || quorum < 1 { + return 0 // Invalid quorum, treat as must-success + } + return quorum + } + } + return 0 // Label not found, treat as must-success +} + func (s *ProxyStore) LabelSet() []labelpb.ZLabelSet { stores := s.stores() if len(stores) == 0 { @@ -461,11 +514,22 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. stores []Client storeLabelSets []labels.Labels ) - // groupReplicaStores[groupKey][replicaKey] = number of stores with the groupKey and replicaKey + // For label-based quorum strategy: + // - groupStores[groupKey] = total stores in group + // - groupQuorum[groupKey] = quorum requirement for group + // - groupFailed[groupKey] = failed stores in group + // - mustSuccessStores = stores without labels or invalid quorum (must all succeed) + groupStores := make(map[string]int) + groupQuorum := make(map[string]int) + groupFailed := make(map[string]int) + mustSuccessStores := make(map[Client]bool) + totalFailedStores := 0 + + // For legacy DNS-based strategy (when label flags not set): + // groupReplicaStores[groupKey][replicaKey] = number of stores + // failedStores[groupKey][replicaKey] = number of failures groupReplicaStores := make(map[string]map[string]int) - // failedStores[groupKey][replicaKey] = number of store failures failedStores := make(map[string]map[string]int) - totalFailedStores := 0 bumpCounter := func(key1, key2 string, mp map[string]map[string]int) { if _, ok := mp[key1]; !ok { mp[key1] = make(map[string]int) @@ -473,6 +537,9 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. mp[key1][key2]++ } + // Check if we're using label-based quorum strategy + useLabelBasedStrategy := s.groupReplicaGroupLabel != "" && s.groupReplicaQuorumLabel != "" + stores, storeLabelSets, storeDebugMsgs := s.matchingStores(ctx, originalRequest.MinTime, originalRequest.MaxTime, matchers) s.metrics.storesPerQueryAfterFiltering.Set(float64(len(stores))) @@ -481,7 +548,22 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. s.metrics.storesPerQueryAfterEELFiltering.Set(float64(len(stores))) for _, st := range stores { - bumpCounter(st.GroupKey(), st.ReplicaKey(), groupReplicaStores) + if useLabelBasedStrategy { + groupKey := s.getGroupKey(st) + quorum := s.getQuorum(st) + if groupKey == "" || quorum == 0 { + // No valid group/quorum labels - must-success store + mustSuccessStores[st] = true + } else { + groupStores[groupKey]++ + if _, exists := groupQuorum[groupKey]; !exists { + groupQuorum[groupKey] = quorum + } + } + } else { + // Legacy DNS-based strategy + bumpCounter(st.GroupKey(), st.ReplicaKey(), groupReplicaStores) + } } if len(stores) == 0 { level.Debug(reqLogger).Log("err", ErrorNoStoresMatched, "stores", strings.Join(storeDebugMsgs, ";")) @@ -515,38 +597,81 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. storeResponses := make([]respSet, 0, len(stores)) checkGroupReplicaErrors := func(st Client, err error) error { - if len(failedStores[st.GroupKey()]) > 1 { + if useLabelBasedStrategy { + // Label-based quorum strategy + if mustSuccessStores[st] { + // Must-success store failed - abort immediately + addr, _ := st.Addr() + msg := "Must-success store failed (no valid group/quorum labels)" + level.Error(reqLogger).Log("msg", msg, "store", addr) + return fmt.Errorf("%s store=%s: %w", msg, addr, err) + } + + groupKey := s.getGroupKey(st) + groupFailed[groupKey]++ + healthy := groupStores[groupKey] - groupFailed[groupKey] + quorum := groupQuorum[groupKey] + + if healthy < quorum { + msg := "Group does not meet quorum requirement" + level.Error(reqLogger).Log( + "msg", msg, + "group", groupKey, + "healthy", healthy, + "quorum", quorum, + "total", groupStores[groupKey], + "failed", groupFailed[groupKey], + ) + return fmt.Errorf("%s group=%s healthy=%d quorum=%d: %w", msg, groupKey, healthy, quorum, err) + } + return nil + } + + // Legacy DNS-based strategy + groupKey := st.GroupKey() + replicaKey := st.ReplicaKey() + if len(failedStores[groupKey]) > 1 { msg := "Multiple replicas have failures for the same group" - group := st.GroupKey() - replicas := fmt.Sprintf("%+v", failedStores[group]) + replicas := fmt.Sprintf("%+v", failedStores[groupKey]) level.Error(reqLogger).Log( "msg", msg, - "group", group, + "group", groupKey, "replicas", replicas, ) - return fmt.Errorf("%s group=%s replicas=%s: %w", msg, group, replicas, err) + return fmt.Errorf("%s group=%s replicas=%s: %w", msg, groupKey, replicas, err) } - if len(groupReplicaStores[st.GroupKey()]) == 1 && failedStores[st.GroupKey()][st.ReplicaKey()] > 1 { + if len(groupReplicaStores[groupKey]) == 1 && failedStores[groupKey][replicaKey] > 1 { msg := "A group with single replica has multiple failures" - group := st.GroupKey() - replicas := fmt.Sprintf("%+v", failedStores[group]) + replicas := fmt.Sprintf("%+v", failedStores[groupKey]) level.Error(reqLogger).Log( "msg", msg, - "group", group, + "group", groupKey, "replicas", replicas, ) - return fmt.Errorf("%s group=%s replicas=%s: %w", msg, group, replicas, err) + return fmt.Errorf("%s group=%s replicas=%s: %w", msg, groupKey, replicas, err) } return nil } logGroupReplicaErrors := func() { - if len(failedStores) > 0 { - level.Warn(s.logger).Log("msg", "Group/replica errors", - "errors", fmt.Sprintf("%+v", failedStores), - "total_failed_stores", totalFailedStores, - ) - s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) + if useLabelBasedStrategy { + if len(groupFailed) > 0 { + level.Warn(s.logger).Log("msg", "Group/quorum errors", + "mode", "label-based", + "group_failures", fmt.Sprintf("%+v", groupFailed), + "total_failed_stores", totalFailedStores, + ) + s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) + } + } else { + if len(failedStores) > 0 { + level.Warn(s.logger).Log("msg", "Group/replica errors", + "mode", "dns-based", + "errors", fmt.Sprintf("%+v", failedStores), + "total_failed_stores", totalFailedStores, + ) + s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) + } } } defer logGroupReplicaErrors() @@ -609,10 +734,21 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. grpcErrorCode = extractGRPCCode(err) } - level.Warn(s.logger).Log("msg", "Store failure", "group", st.GroupKey(), "replica", st.ReplicaKey(), "err", err) - s.metrics.storeFailureCount.WithLabelValues(st.GroupKey(), st.ReplicaKey()).Inc() - bumpCounter(st.GroupKey(), st.ReplicaKey(), failedStores) totalFailedStores++ + if useLabelBasedStrategy { + groupKey := s.getGroupKey(st) + addr, _ := st.Addr() + level.Warn(s.logger).Log("msg", "Store failure", "group", groupKey, "store", addr, "err", err) + // Note: We don't record to storeFailureCount metric in label-based mode + // because the existing metric uses (group, replica) labels which have + // different semantics than (group, addr). + } else { + groupKey := st.GroupKey() + replicaKey := st.ReplicaKey() + level.Warn(s.logger).Log("msg", "Store failure", "group", groupKey, "replica", replicaKey, "err", err) + s.metrics.storeFailureCount.WithLabelValues(groupKey, replicaKey).Inc() + bumpCounter(groupKey, replicaKey, failedStores) + } if originalRequest.PartialResponseStrategy == storepb.PartialResponseStrategy_GROUP_REPLICA { if checkGroupReplicaErrors(st, err) != nil { return err diff --git a/pkg/store/proxy_test.go b/pkg/store/proxy_test.go index 5c30008e219..bab8e797bbc 100644 --- a/pkg/store/proxy_test.go +++ b/pkg/store/proxy_test.go @@ -3556,3 +3556,147 @@ func TestDedupRespHeap_QuorumChunkDedup(t *testing.T) { } } + +func TestProxyStore_GetGroupKeyQuorum(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + groupReplicaGroupLabel string + groupReplicaQuorumLabel string + client *storetestutil.TestClient + expectedGroupKey string + expectedQuorum int + }{ + { + name: "no labels configured - fallback to DNS-based group key", + groupReplicaGroupLabel: "", + groupReplicaQuorumLabel: "", + client: &storetestutil.TestClient{ + Name: "receive-rep0-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-rep0", + ReplicaKeyStr: "receive-rep0-0", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "2")}, + }, + expectedGroupKey: "receive-rep0", + expectedQuorum: 0, // No quorum label configured, returns 0 + }, + { + name: "labels configured - use label-based group key and quorum", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "receive-rep0-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-rep0", + ReplicaKeyStr: "receive-rep0-0", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "2")}, + }, + expectedGroupKey: "receive-0", + expectedQuorum: 2, + }, + { + name: "labels configured but store missing labels - must-success store", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "bucket-store.svc.cluster.local:10901", + GroupKeyStr: "bucket-store", + ReplicaKeyStr: "bucket-store", + ExtLset: []labels.Labels{labels.FromStrings("cluster", "us-east")}, // no receive_group/quorum + }, + expectedGroupKey: "", // Empty string indicates must-success store + expectedQuorum: 0, // No quorum label, must-success store + }, + { + name: "labels configured - multiple label sets, first match used", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "receive-rep1-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-rep1", + ReplicaKeyStr: "receive-rep1-0", + ExtLset: []labels.Labels{ + labels.FromStrings("receive_group", "receive-0", "quorum", "3"), + labels.FromStrings("cluster", "us-east"), + }, + }, + expectedGroupKey: "receive-0", + expectedQuorum: 3, + }, + { + name: "labels configured - store has only group label (no quorum)", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "partial-store.svc.cluster.local:10901", + GroupKeyStr: "partial-store", + ReplicaKeyStr: "partial-store", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0")}, // no quorum label + }, + expectedGroupKey: "receive-0", + expectedQuorum: 0, // No quorum label, must-success store within group + }, + { + name: "invalid quorum value (non-integer) - must-success store", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "receive-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-0", + ReplicaKeyStr: "receive-0", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "invalid")}, + }, + expectedGroupKey: "receive-0", + expectedQuorum: 0, // Invalid quorum value, treat as must-success + }, + { + name: "quorum value less than 1 - must-success store", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "receive-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-0", + ReplicaKeyStr: "receive-0", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "0")}, + }, + expectedGroupKey: "receive-0", + expectedQuorum: 0, // quorum < 1, treat as must-success + }, + { + name: "negative quorum value - must-success store", + groupReplicaGroupLabel: "receive_group", + groupReplicaQuorumLabel: "quorum", + client: &storetestutil.TestClient{ + Name: "receive-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-0", + ReplicaKeyStr: "receive-0", + ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "-1")}, + }, + expectedGroupKey: "receive-0", + expectedQuorum: 0, // Negative quorum, treat as must-success + }, + } { + t.Run(tc.name, func(t *testing.T) { + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + + // Create ProxyStore with the configured labels + proxy := NewProxyStore( + logger, + reg, + func() []Client { return []Client{tc.client} }, + component.Query, + labels.EmptyLabels(), + 0, + EagerRetrieval, + WithGroupReplicaLabels(tc.groupReplicaGroupLabel, tc.groupReplicaQuorumLabel), + ) + + gotGroupKey := proxy.getGroupKey(tc.client) + gotQuorum := proxy.getQuorum(tc.client) + + testutil.Equals(t, tc.expectedGroupKey, gotGroupKey) + testutil.Equals(t, tc.expectedQuorum, gotQuorum) + }) + } +} diff --git a/pr.md b/pr.md new file mode 100644 index 00000000000..6005a522898 --- /dev/null +++ b/pr.md @@ -0,0 +1,105 @@ +# Add Label-Based Group Replica Response Strategy + +## Summary + +This PR enhances the `GROUP_REPLICA` partial response strategy to support label-based group and quorum identification, enabling more flexible failure tolerance for replicated data setups like `aligned_ketama` hashring. + +## New Flags + +- `--query.group-replica.group-label`: External label name identifying the group (stores with same value hold replicated data) +- `--query.group-replica.quorum-label`: External label name whose value specifies minimum healthy stores required per group + +## How It Works + +``` +┌─────────────────────────────────────────────────────────────────────────┐ +│ Thanos Query │ +│ │ +│ Flags: │ +│ --query.group-replica.group-label=receive_group │ +│ --query.group-replica.quorum-label=quorum │ +│ --query.partial-response.strategy=GROUP_REPLICA │ +└─────────────────────────────────────────────────────────────────────────┘ + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ + │ Receive-0 │ │ Receive-1 │ │ Receive-2 │ + │ │ │ │ │ │ + │ Labels: │ │ Labels: │ │ Labels: │ + │ receive_group│ │ receive_group│ │ receive_group│ + │ ="group-A" │ │ ="group-A" │ │ ="group-A" │ + │ quorum="2" │ │ quorum="2" │ │ quorum="2" │ + └──────────────┘ └──────────────┘ └──────────────┘ + │ │ │ + └───────────────┴───────────────┘ + │ + Group "group-A" (quorum=2) + Needs 2 of 3 stores healthy +``` + +## Behavior + +| Scenario | Result | +|----------|--------| +| Group has `>= quorum` healthy stores | Query succeeds for that group | +| Group has `< quorum` healthy stores | Query aborts with error | +| Store missing labels or invalid quorum | Treated as "must-success" (any failure aborts) | +| Flags not configured | Falls back to legacy DNS-based strategy | + +## Example Configuration + +### Receive pods with external labels: + +```yaml +# receive-0 in AZ-1 +- --label=receive_group="ordinal-0" +- --label=quorum="2" + +# receive-1 in AZ-2 (replica of receive-0) +- --label=receive_group="ordinal-0" +- --label=quorum="2" + +# receive-2 in AZ-3 (replica of receive-0) +- --label=receive_group="ordinal-0" +- --label=quorum="2" +``` + +### Query configuration: + +```yaml +- --query.partial-response.strategy=GROUP_REPLICA +- --query.group-replica.group-label=receive_group +- --query.group-replica.quorum-label=quorum +``` + +### Failure scenarios: + +``` +Group "ordinal-0" has 3 stores, quorum=2: + + ✓ receive-0 (AZ-1) - healthy + ✗ receive-1 (AZ-2) - failed + ✓ receive-2 (AZ-3) - healthy + + Result: 2 >= 2 (quorum met) → Query succeeds +``` + +``` +Group "ordinal-0" has 3 stores, quorum=2: + + ✗ receive-0 (AZ-1) - failed + ✗ receive-1 (AZ-2) - failed + ✓ receive-2 (AZ-3) - healthy + + Result: 1 < 2 (quorum not met) → Query aborts +``` + +## Label Stripping + +Both `group-label` and `quorum-label` are automatically stripped from query results (similar to replica labels with deduplication). + +## Backward Compatibility + +- When flags are not set, the existing DNS-based `GROUP_REPLICA` behavior is preserved +- No changes required for existing deployments From 10e45923378def8288b393d4f9478e2f6f91fcf6 Mon Sep 17 00:00:00 2001 From: Yuchen Wang Date: Tue, 27 Jan 2026 15:59:04 -0800 Subject: [PATCH 2/3] add non-series external labels --- cmd/thanos/query.go | 2 -- cmd/thanos/receive.go | 17 +++++++-- pkg/query/querier.go | 32 ++++------------- pkg/query/querier_test.go | 73 +++++++++++---------------------------- pkg/receive/multitsdb.go | 15 +++++++- pkg/store/tsdb.go | 34 ++++++++++++++++-- 6 files changed, 87 insertions(+), 86 deletions(-) diff --git a/cmd/thanos/query.go b/cmd/thanos/query.go index 892b2ed7629..6257fe149eb 100644 --- a/cmd/thanos/query.go +++ b/cmd/thanos/query.go @@ -671,8 +671,6 @@ func runQuery( ) opts := query.Options{ GroupReplicaPartialResponseStrategy: groupReplicaPartialResponseStrategy, - GroupReplicaGroupLabel: groupReplicaGroupLabel, - GroupReplicaQuorumLabel: groupReplicaQuorumLabel, DeduplicationFunc: queryDeduplicationFunc, RewriteAggregationLabelStrategy: rewriteAggregationLabelStrategy, RewriteAggregationLabelTo: rewriteAggregationLabelTo, diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go index 1df000f4436..f7d76f9b83c 100644 --- a/cmd/thanos/receive.go +++ b/cmd/thanos/receive.go @@ -73,6 +73,11 @@ func registerReceive(app *extkingpin.App) { return errors.Wrap(err, "parse labels") } + infoOnlyLset, err := parseFlagLabels(conf.infoOnlyLblStrs) + if err != nil { + return errors.Wrap(err, "parse info-only labels") + } + if !model.LabelName.IsValid(model.LabelName(conf.tenantLabelName)) { return errors.Errorf("unsupported format for tenant label name, got %s", conf.tenantLabelName) } @@ -117,6 +122,7 @@ func registerReceive(app *extkingpin.App) { logFilterMethods, tsdbOpts, lset, + infoOnlyLset, component.Receive, metadata.HashFunc(conf.hashFunc), receiveMode, @@ -135,6 +141,7 @@ func runReceive( logFilterMethods []string, tsdbOpts *tsdb.Options, lset labels.Labels, + infoOnlyLset labels.Labels, comp component.SourceStoreAPI, hashFunc metadata.HashFunc, receiveMode receive.ReceiverMode, @@ -249,6 +256,9 @@ func runReceive( } multiTSDBOptions = append(multiTSDBOptions, receive.WithMatchersCache(cache)) } + if infoOnlyLset.Len() > 0 { + multiTSDBOptions = append(multiTSDBOptions, receive.WithInfoOnlyLabels(infoOnlyLset)) + } dbs := receive.NewMultiTSDB( conf.dataDir, @@ -931,8 +941,9 @@ type receiveConfig struct { rwClientSkipVerify bool rwServerTlsMinVersion string - dataDir string - labelStrs []string + dataDir string + labelStrs []string + infoOnlyLblStrs []string objStoreConfig *extflag.PathOrContent retention *model.Duration @@ -1035,6 +1046,8 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) { cmd.Flag("label", "External labels to announce. This flag will be removed in the future when handling multiple tsdb instances is added.").PlaceHolder("key=\"value\"").StringsVar(&rc.labelStrs) + cmd.Flag("label-info-only", "Labels to expose only via InfoAPI (for store discovery), but NOT merge into series responses. Useful for metadata like replica group/quorum that query component needs for routing but shouldn't be in query results.").PlaceHolder("key=\"value\"").StringsVar(&rc.infoOnlyLblStrs) + rc.objStoreConfig = extkingpin.RegisterCommonObjStoreFlags(cmd, "", false) rc.retention = extkingpin.ModelDuration(cmd.Flag("tsdb.retention", "How long to retain raw samples on local storage. 0d - disables the retention policy (i.e. infinite retention). For more details on how retention is enforced for individual tenants, please refer to the Tenant lifecycle management section in the Receive documentation: https://thanos.io/tip/components/receive.md/#tenant-lifecycle-management").Default("15d")) diff --git a/pkg/query/querier.go b/pkg/query/querier.go index 8ca73c2a3b1..250344a92a3 100644 --- a/pkg/query/querier.go +++ b/pkg/query/querier.go @@ -59,8 +59,6 @@ type QueryableCreator func( type Options struct { GroupReplicaPartialResponseStrategy bool - GroupReplicaGroupLabel string - GroupReplicaQuorumLabel string DeduplicationFunc string RewriteAggregationLabelStrategy string RewriteAggregationLabelTo string @@ -172,12 +170,6 @@ type querier struct { shardInfo *storepb.ShardInfo seriesStatsReporter seriesStatsReporter - // groupReplicaGroupLabel and groupReplicaQuorumLabel are the external label names - // used for label-based group/quorum identification. These labels are stripped from - // query results when set. - groupReplicaGroupLabel string - groupReplicaQuorumLabel string - aggregationLabelRewriter *AggregationLabelRewriter } @@ -256,8 +248,6 @@ func newQuerierWithOpts( skipChunks: skipChunks, shardInfo: shardInfo, seriesStatsReporter: seriesStatsReporter, - groupReplicaGroupLabel: opts.GroupReplicaGroupLabel, - groupReplicaQuorumLabel: opts.GroupReplicaQuorumLabel, aggregationLabelRewriter: aggregationLabelRewriter, } @@ -268,25 +258,15 @@ func (q *querier) isDedupEnabled() bool { } // getWithoutReplicaLabels returns the list of labels to strip from query results. -// This includes replica labels for deduplication and group/quorum labels for the -// GROUP_REPLICA partial response strategy. +// This includes replica labels for deduplication. +// Note: group/quorum labels for GROUP_REPLICA strategy are NOT included here because +// they are configured as info-only labels on receivers, meaning they are only exposed +// via InfoAPI for store discovery but NOT merged into series responses. func (q *querier) getWithoutReplicaLabels() []string { - var labels []string - - // Add replica labels if deduplication is enabled if q.isDedupEnabled() { - labels = append(labels, q.replicaLabels...) - } - - // Add group/quorum labels if configured (for GROUP_REPLICA strategy) - if q.groupReplicaGroupLabel != "" { - labels = append(labels, q.groupReplicaGroupLabel) + return q.replicaLabels } - if q.groupReplicaQuorumLabel != "" { - labels = append(labels, q.groupReplicaQuorumLabel) - } - - return labels + return nil } type seriesServer struct { diff --git a/pkg/query/querier_test.go b/pkg/query/querier_test.go index 9fd1540a4ec..d1a6530588d 100644 --- a/pkg/query/querier_test.go +++ b/pkg/query/querier_test.go @@ -1283,69 +1283,38 @@ func storeSeriesResponse(t testing.TB, lset labels.Labels, smplChunks ...[]sampl func TestQuerier_GetWithoutReplicaLabels(t *testing.T) { t.Parallel() + // Note: group/quorum labels for GROUP_REPLICA strategy are no longer included in + // WithoutReplicaLabels because they are configured as info-only labels on receivers, + // meaning they are only exposed via InfoAPI and NOT merged into series responses. for _, tc := range []struct { - name string - deduplicate bool - replicaLabels []string - groupReplicaGroupLabel string - groupReplicaQuorumLabel string - expectedLabels []string + name string + deduplicate bool + replicaLabels []string + expectedLabels []string }{ { - name: "no dedup, no group replica labels - empty result", - deduplicate: false, - replicaLabels: []string{"replica"}, - groupReplicaGroupLabel: "", - groupReplicaQuorumLabel: "", - expectedLabels: nil, - }, - { - name: "dedup enabled - only replica labels", - deduplicate: true, - replicaLabels: []string{"replica", "prometheus_replica"}, - groupReplicaGroupLabel: "", - groupReplicaQuorumLabel: "", - expectedLabels: []string{"replica", "prometheus_replica"}, - }, - { - name: "no dedup, group replica labels configured - only group and quorum labels", - deduplicate: false, - replicaLabels: []string{"replica"}, - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - expectedLabels: []string{"receive_group", "quorum"}, - }, - { - name: "dedup enabled and group replica labels - both combined", - deduplicate: true, - replicaLabels: []string{"replica"}, - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - expectedLabels: []string{"replica", "receive_group", "quorum"}, + name: "no dedup - empty result", + deduplicate: false, + replicaLabels: []string{"replica"}, + expectedLabels: nil, }, { - name: "dedup enabled but no replica labels - only group and quorum labels", - deduplicate: true, - replicaLabels: []string{}, - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - expectedLabels: []string{"receive_group", "quorum"}, + name: "dedup enabled - only replica labels", + deduplicate: true, + replicaLabels: []string{"replica", "prometheus_replica"}, + expectedLabels: []string{"replica", "prometheus_replica"}, }, { - name: "only group label configured (partial) - only group label", - deduplicate: false, - replicaLabels: []string{}, - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "", - expectedLabels: []string{"receive_group"}, + name: "dedup enabled but no replica labels - empty result", + deduplicate: true, + replicaLabels: []string{}, + expectedLabels: nil, }, } { t.Run(tc.name, func(t *testing.T) { q := &querier{ - deduplicate: tc.deduplicate, - replicaLabels: tc.replicaLabels, - groupReplicaGroupLabel: tc.groupReplicaGroupLabel, - groupReplicaQuorumLabel: tc.groupReplicaQuorumLabel, + deduplicate: tc.deduplicate, + replicaLabels: tc.replicaLabels, } got := q.getWithoutReplicaLabels() diff --git a/pkg/receive/multitsdb.go b/pkg/receive/multitsdb.go index 0eaa96bc2d1..580c1ce98bb 100644 --- a/pkg/receive/multitsdb.go +++ b/pkg/receive/multitsdb.go @@ -53,6 +53,7 @@ type MultiTSDB struct { tsdbOpts *tsdb.Options tenantLabelName string labels labels.Labels + infoOnlyLabels labels.Labels // Labels only exposed via InfoAPI, not merged into series bucket objstore.Bucket mtx *sync.RWMutex @@ -110,6 +111,14 @@ func WithMatchersCache(cache storecache.MatchersCache) MultiTSDBOption { } } +// WithInfoOnlyLabels sets labels that are only exposed via InfoAPI (for store discovery), +// but NOT merged into series responses. Useful for metadata like replica group/quorum. +func WithInfoOnlyLabels(lset labels.Labels) MultiTSDBOption { + return func(s *MultiTSDB) { + s.infoOnlyLabels = lset + } +} + // NewMultiTSDB creates new MultiTSDB. // NOTE: Passed labels must be sorted lexicographically (alphabetically). func NewMultiTSDB( @@ -846,7 +855,11 @@ func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant if t.matcherCache != nil { options = append(options, store.WithMatcherCacheInstance(t.matcherCache)) } - tenant.set(store.NewTSDBStore(logger, s, component.Receive, lset, options...), s, ship, exemplars.NewTSDB(s, lset)) + tsdbStore := store.NewTSDBStore(logger, s, component.Receive, lset, options...) + if t.infoOnlyLabels.Len() > 0 { + tsdbStore.SetInfoOnlyLset(t.infoOnlyLabels) + } + tenant.set(tsdbStore, s, ship, exemplars.NewTSDB(s, lset)) t.addTenantLocked(tenantID, tenant) // need to update the client list once store is ready & client != nil level.Info(logger).Log("msg", "TSDB is now ready") return nil diff --git a/pkg/store/tsdb.go b/pkg/store/tsdb.go index 4d6b8c9ef2d..5d16ba9bd8f 100644 --- a/pkg/store/tsdb.go +++ b/pkg/store/tsdb.go @@ -72,6 +72,7 @@ type TSDBStore struct { matcherCache storecache.MatchersCache extLset labels.Labels + infoOnlyLset labels.Labels // Labels only exposed via InfoAPI, not merged into series startStoreFilterUpdate bool storeFilter filter.StoreFilter mtx sync.RWMutex @@ -169,6 +170,22 @@ func (s *TSDBStore) SetExtLset(extLset labels.Labels) { s.extLset = extLset } +// SetInfoOnlyLset sets labels that are only exposed via InfoAPI (LabelSet), +// but NOT merged into series responses. Useful for metadata like replica group/quorum. +func (s *TSDBStore) SetInfoOnlyLset(infoOnlyLset labels.Labels) { + s.mtx.Lock() + defer s.mtx.Unlock() + + s.infoOnlyLset = infoOnlyLset +} + +func (s *TSDBStore) getInfoOnlyLset() labels.Labels { + s.mtx.RLock() + defer s.mtx.RUnlock() + + return s.infoOnlyLset +} + func (s *TSDBStore) getExtLset() labels.Labels { s.mtx.RLock() defer s.mtx.RUnlock() @@ -177,10 +194,21 @@ func (s *TSDBStore) getExtLset() labels.Labels { } func (s *TSDBStore) LabelSet() []labelpb.ZLabelSet { - labels := labelpb.ZLabelSetsFromPromLabels(s.getExtLset()) + // Combine extLset and infoOnlyLset for InfoAPI exposure. + // infoOnlyLset labels are NOT merged into series responses. + combined := s.getExtLset() + if infoOnly := s.getInfoOnlyLset(); infoOnly.Len() > 0 { + builder := labels.NewBuilder(combined) + infoOnly.Range(func(l labels.Label) { + builder.Set(l.Name, l.Value) + }) + combined = builder.Labels() + } + + zlabels := labelpb.ZLabelSetsFromPromLabels(combined) labelSets := []labelpb.ZLabelSet{} - if len(labels) > 0 { - labelSets = append(labelSets, labels...) + if len(zlabels) > 0 { + labelSets = append(labelSets, zlabels...) } return labelSets From a8f58fcddf1f07ae6138942fb90f3cd3a4b0e128 Mon Sep 17 00:00:00 2001 From: Yuchen Wang Date: Wed, 28 Jan 2026 15:45:13 -0800 Subject: [PATCH 3/3] add replica_group and quorum to proto --- cmd/thanos/query.go | 21 --- cmd/thanos/receive.go | 24 ++- cmd/thanos/rule.go | 1 - pkg/info/infopb/rpc.pb.go | 158 +++++++++++++----- pkg/info/infopb/rpc.proto | 12 ++ pkg/query/endpointset.go | 53 +++--- pkg/query/endpointset_test.go | 10 +- pkg/receive/multitsdb.go | 26 +-- pkg/store/proxy.go | 231 ++++++++++----------------- pkg/store/proxy_test.go | 176 ++++++++------------ pkg/store/storepb/replica_info.go | 26 +++ pkg/store/storepb/testutil/client.go | 34 +++- pkg/store/tsdb.go | 39 +++-- 13 files changed, 414 insertions(+), 397 deletions(-) create mode 100644 pkg/store/storepb/replica_info.go diff --git a/cmd/thanos/query.go b/cmd/thanos/query.go index 6257fe149eb..f82ed44963b 100644 --- a/cmd/thanos/query.go +++ b/cmd/thanos/query.go @@ -200,12 +200,6 @@ func registerQuery(app *extkingpin.App) { enableGroupReplicaPartialStrategy := cmd.Flag("query.group-replica-strategy", "Enable group-replica partial response strategy."). Default("false").Bool() - groupReplicaGroupLabel := cmd.Flag("query.group-replica.group-label", "External label name that identifies the group for group-replica partial response strategy. Stores with the same group label value hold replicated data. Must be set together with --query.group-replica.quorum-label."). - Default("").String() - - groupReplicaQuorumLabel := cmd.Flag("query.group-replica.quorum-label", "External label name whose value specifies the minimum number of healthy stores required per group. Must be set together with --query.group-replica.group-label. Stores without these labels or with invalid quorum values (<1) are treated as must-success stores."). - Default("").String() - enableRulePartialResponse := cmd.Flag("rule.partial-response", "Enable partial response for rules endpoint. --no-rule.partial-response for disabling."). Hidden().Default("true").Bool() @@ -274,11 +268,6 @@ func registerQuery(app *extkingpin.App) { return errors.Wrap(err, "parse federation labels") } - // Validate group-replica label flags: both must be set together or neither. - if (*groupReplicaGroupLabel == "") != (*groupReplicaQuorumLabel == "") { - return errors.New("--query.group-replica.group-label and --query.group-replica.quorum-label must be set together or neither") - } - for _, feature := range *featureList { if feature == promqlAtModifier { level.Warn(logger).Log("msg", "This option for --enable-feature is now permanently enabled and therefore a no-op.", "option", promqlAtModifier) @@ -419,8 +408,6 @@ func registerQuery(app *extkingpin.App) { *enforceTenancy, *tenantLabel, *enableGroupReplicaPartialStrategy, - *groupReplicaGroupLabel, - *groupReplicaQuorumLabel, *rewriteAggregationLabelStrategy, *rewriteAggregationLabelTo, *lazyRetrievalMaxBufferedResponses, @@ -512,8 +499,6 @@ func runQuery( enforceTenancy bool, tenantLabel string, groupReplicaPartialResponseStrategy bool, - groupReplicaGroupLabel string, - groupReplicaQuorumLabel string, rewriteAggregationLabelStrategy string, rewriteAggregationLabelTo string, lazyRetrievalMaxBufferedResponses int, @@ -627,9 +612,6 @@ func runQuery( if len(exclusiveExternalLabels) > 0 { options = append(options, store.WithExclusiveExternalLabels(exclusiveExternalLabels)) } - if groupReplicaGroupLabel != "" && groupReplicaQuorumLabel != "" { - options = append(options, store.WithGroupReplicaLabels(groupReplicaGroupLabel, groupReplicaQuorumLabel)) - } // Parse and sanitize the provided replica labels flags. queryReplicaLabels = strutil.ParseFlagLabels(queryReplicaLabels) @@ -657,7 +639,6 @@ func runQuery( endpointInfoTimeout, // ignoreErrors when group_replica partial response strategy is enabled. groupReplicaPartialResponseStrategy, - groupReplicaQuorumLabel, queryConnMetricLabels..., ) @@ -985,7 +966,6 @@ func prepareEndpointSet( unhealthyStoreTimeout time.Duration, endpointInfoTimeout time.Duration, ignoreStoreErrors bool, - quorumLabelName string, queryConnMetricLabels ...string, ) *query.EndpointSet { endpointSet := query.NewEndpointSet( @@ -1031,7 +1011,6 @@ func prepareEndpointSet( dialOpts, unhealthyStoreTimeout, endpointInfoTimeout, - quorumLabelName, queryConnMetricLabels..., ) diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go index f7d76f9b83c..fcb7d71731f 100644 --- a/cmd/thanos/receive.go +++ b/cmd/thanos/receive.go @@ -73,11 +73,6 @@ func registerReceive(app *extkingpin.App) { return errors.Wrap(err, "parse labels") } - infoOnlyLset, err := parseFlagLabels(conf.infoOnlyLblStrs) - if err != nil { - return errors.Wrap(err, "parse info-only labels") - } - if !model.LabelName.IsValid(model.LabelName(conf.tenantLabelName)) { return errors.Errorf("unsupported format for tenant label name, got %s", conf.tenantLabelName) } @@ -122,7 +117,6 @@ func registerReceive(app *extkingpin.App) { logFilterMethods, tsdbOpts, lset, - infoOnlyLset, component.Receive, metadata.HashFunc(conf.hashFunc), receiveMode, @@ -141,7 +135,6 @@ func runReceive( logFilterMethods []string, tsdbOpts *tsdb.Options, lset labels.Labels, - infoOnlyLset labels.Labels, comp component.SourceStoreAPI, hashFunc metadata.HashFunc, receiveMode receive.ReceiverMode, @@ -256,8 +249,8 @@ func runReceive( } multiTSDBOptions = append(multiTSDBOptions, receive.WithMatchersCache(cache)) } - if infoOnlyLset.Len() > 0 { - multiTSDBOptions = append(multiTSDBOptions, receive.WithInfoOnlyLabels(infoOnlyLset)) + if conf.replicaGroup != "" { + multiTSDBOptions = append(multiTSDBOptions, receive.WithReplicaGroup(conf.replicaGroup, conf.quorum)) } dbs := receive.NewMultiTSDB( @@ -459,6 +452,8 @@ func runReceive( SupportsSharding: true, SupportsWithoutReplicaLabels: true, TsdbInfos: proxy.TSDBInfos(), + ReplicaGroup: conf.replicaGroup, + Quorum: int32(conf.quorum), }, nil } return nil, errors.New("Not ready") @@ -941,9 +936,10 @@ type receiveConfig struct { rwClientSkipVerify bool rwServerTlsMinVersion string - dataDir string - labelStrs []string - infoOnlyLblStrs []string + dataDir string + labelStrs []string + replicaGroup string + quorum int objStoreConfig *extflag.PathOrContent retention *model.Duration @@ -1046,7 +1042,9 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) { cmd.Flag("label", "External labels to announce. This flag will be removed in the future when handling multiple tsdb instances is added.").PlaceHolder("key=\"value\"").StringsVar(&rc.labelStrs) - cmd.Flag("label-info-only", "Labels to expose only via InfoAPI (for store discovery), but NOT merge into series responses. Useful for metadata like replica group/quorum that query component needs for routing but shouldn't be in query results.").PlaceHolder("key=\"value\"").StringsVar(&rc.infoOnlyLblStrs) + cmd.Flag("receive.replica-group", "Replica group identifier for GROUP_REPLICA partial response strategy. Stores with the same replica_group value hold replicated data. When set, the query layer can tolerate failures as long as enough replicas in the group are healthy.").Default("").StringVar(&rc.replicaGroup) + + cmd.Flag("receive.quorum", "Minimum number of healthy stores required per replica group for GROUP_REPLICA partial response strategy. Only meaningful when receive.replica-group is set. If 0, the store is treated as must-success.").Default("0").IntVar(&rc.quorum) rc.objStoreConfig = extkingpin.RegisterCommonObjStoreFlags(cmd, "", false) diff --git a/cmd/thanos/rule.go b/cmd/thanos/rule.go index 16235c2894e..41d996e0200 100644 --- a/cmd/thanos/rule.go +++ b/cmd/thanos/rule.go @@ -444,7 +444,6 @@ func runRule( 5*time.Minute, 5*time.Second, false, - "", // quorumLabelName - not used in rule component ) // Periodically update the GRPC addresses from query config by resolving them using DNS SD if necessary. diff --git a/pkg/info/infopb/rpc.pb.go b/pkg/info/infopb/rpc.pb.go index c88be63cfc1..5712b877be9 100644 --- a/pkg/info/infopb/rpc.pb.go +++ b/pkg/info/infopb/rpc.pb.go @@ -124,6 +124,16 @@ type StoreInfo struct { SupportsWithoutReplicaLabels bool `protobuf:"varint,5,opt,name=supports_without_replica_labels,json=supportsWithoutReplicaLabels,proto3" json:"supports_without_replica_labels,omitempty"` // TSDBInfos holds metadata for all TSDBs exposed by the store. TsdbInfos []TSDBInfo `protobuf:"bytes,6,rep,name=tsdb_infos,json=tsdbInfos,proto3" json:"tsdb_infos"` + // Replica topology hints for GROUP_REPLICA partial response strategy. + // Stores with the same replica_group value hold replicated data. + // When set, the query layer can tolerate failures as long as enough + // replicas in the group are healthy (determined by quorum). + // If empty, the store is treated as a "must-success" store. + ReplicaGroup string `protobuf:"bytes,7,opt,name=replica_group,json=replicaGroup,proto3" json:"replica_group,omitempty"` + // Minimum number of healthy stores required per replica group. + // Only meaningful when replica_group is set. + // If 0 or not set, the store is treated as a "must-success" store. + Quorum int32 `protobuf:"varint,8,opt,name=quorum,proto3" json:"quorum,omitempty"` } func (m *StoreInfo) Reset() { *m = StoreInfo{} } @@ -400,44 +410,46 @@ func init() { func init() { proto.RegisterFile("info/infopb/rpc.proto", fileDescriptor_a1214ec45d2bf952) } var fileDescriptor_a1214ec45d2bf952 = []byte{ - // 589 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x94, 0xdf, 0x8a, 0xda, 0x40, - 0x14, 0xc6, 0x8d, 0x7f, 0xe3, 0x71, 0xdd, 0xee, 0x0e, 0xbb, 0x25, 0x4a, 0x89, 0x12, 0xf6, 0x42, - 0x68, 0x31, 0x60, 0xa1, 0x94, 0xf6, 0xaa, 0x6e, 0x85, 0x6e, 0xe9, 0x42, 0x1b, 0x85, 0xc2, 0xde, - 0x84, 0xa8, 0xb3, 0x1a, 0x48, 0x32, 0x63, 0x66, 0xa4, 0xfa, 0x16, 0x7d, 0x95, 0xbe, 0x85, 0x97, - 0x7b, 0xd9, 0xab, 0xd2, 0xea, 0x43, 0xf4, 0xb6, 0xcc, 0x4c, 0x62, 0x0d, 0xdd, 0xbd, 0xe9, 0x8d, - 0x66, 0xe6, 0xfb, 0x9d, 0xc9, 0x39, 0xdf, 0x39, 0x13, 0x38, 0xf7, 0xa3, 0x5b, 0x62, 0x8b, 0x1f, - 0x3a, 0xb6, 0x63, 0x3a, 0xe9, 0xd2, 0x98, 0x70, 0x82, 0x6a, 0x7c, 0xee, 0x45, 0x84, 0x75, 0x85, - 0xd0, 0x6c, 0x30, 0x4e, 0x62, 0x6c, 0x07, 0xde, 0x18, 0x07, 0x74, 0x6c, 0xf3, 0x35, 0xc5, 0x4c, - 0x71, 0xcd, 0xb3, 0x19, 0x99, 0x11, 0xf9, 0x68, 0x8b, 0x27, 0xb5, 0x6b, 0xd5, 0xa1, 0x76, 0x15, - 0xdd, 0x12, 0x07, 0x2f, 0x96, 0x98, 0x71, 0xeb, 0x5b, 0x01, 0x8e, 0xd4, 0x9a, 0x51, 0x12, 0x31, - 0x8c, 0x5e, 0x00, 0xc8, 0xc3, 0x5c, 0x86, 0x39, 0x33, 0xb4, 0x76, 0xa1, 0x53, 0xeb, 0x9d, 0x76, - 0x93, 0x57, 0xde, 0x7c, 0x10, 0xd2, 0x10, 0xf3, 0x7e, 0x71, 0xf3, 0xa3, 0x95, 0x73, 0xaa, 0x41, - 0xb2, 0x66, 0xe8, 0x02, 0xea, 0x97, 0x24, 0xa4, 0x24, 0xc2, 0x11, 0x1f, 0xad, 0x29, 0x36, 0xf2, - 0x6d, 0xad, 0x53, 0x75, 0xb2, 0x9b, 0xe8, 0x19, 0x94, 0x64, 0xc2, 0x46, 0xa1, 0xad, 0x75, 0x6a, - 0xbd, 0xc7, 0xdd, 0x83, 0x5a, 0xba, 0x43, 0xa1, 0xc8, 0x64, 0x14, 0x24, 0xe8, 0x78, 0x19, 0x60, - 0x66, 0x14, 0xef, 0xa1, 0x1d, 0xa1, 0x28, 0x5a, 0x42, 0xe8, 0x1d, 0x3c, 0x0a, 0x31, 0x8f, 0xfd, - 0x89, 0x1b, 0x62, 0xee, 0x4d, 0x3d, 0xee, 0x19, 0x25, 0x19, 0xd7, 0xca, 0xc4, 0x5d, 0x4b, 0xe6, - 0x3a, 0x41, 0xe4, 0x01, 0xc7, 0x61, 0x66, 0x0f, 0xf5, 0xa0, 0xc2, 0xbd, 0x78, 0x26, 0x0c, 0x28, - 0xcb, 0x13, 0x8c, 0xcc, 0x09, 0x23, 0xa5, 0xc9, 0xd0, 0x14, 0x44, 0x2f, 0xa1, 0x8a, 0x57, 0x38, - 0xa4, 0x81, 0x17, 0x33, 0xa3, 0x22, 0xa3, 0x9a, 0x99, 0xa8, 0x41, 0xaa, 0xca, 0xb8, 0xbf, 0x30, - 0xb2, 0xa1, 0xb4, 0x58, 0xe2, 0x78, 0x6d, 0xe8, 0x32, 0xaa, 0x91, 0x89, 0xfa, 0x24, 0x94, 0x37, - 0x1f, 0xaf, 0x54, 0xa1, 0x92, 0xb3, 0x7e, 0x6b, 0x50, 0xdd, 0x7b, 0x85, 0x1a, 0xa0, 0x87, 0x7e, - 0xe4, 0x72, 0x3f, 0xc4, 0x86, 0xd6, 0xd6, 0x3a, 0x05, 0xa7, 0x12, 0xfa, 0xd1, 0xc8, 0x0f, 0xb1, - 0x94, 0xbc, 0x95, 0x92, 0xf2, 0x89, 0xe4, 0xad, 0xa4, 0xf4, 0x14, 0x4e, 0xd9, 0x92, 0x52, 0x12, - 0x73, 0xe6, 0xb2, 0xb9, 0x17, 0x4f, 0xfd, 0x68, 0x26, 0x9b, 0xa2, 0x3b, 0x27, 0xa9, 0x30, 0x4c, - 0xf6, 0xd1, 0x00, 0x5a, 0x7b, 0xf8, 0x8b, 0xcf, 0xe7, 0x64, 0xc9, 0xdd, 0x18, 0xd3, 0xc0, 0x9f, - 0x78, 0xae, 0x9c, 0x00, 0x26, 0x9d, 0xd6, 0x9d, 0x27, 0x29, 0xf6, 0x59, 0x51, 0x8e, 0x82, 0xe4, - 0xd4, 0x30, 0xf4, 0x0a, 0x80, 0xb3, 0xe9, 0xd8, 0x15, 0x85, 0x09, 0x67, 0xc5, 0x68, 0x9d, 0x67, - 0x9d, 0x1d, 0xbe, 0xed, 0x8b, 0xa2, 0xd2, 0xf1, 0x12, 0xb8, 0x58, 0xb3, 0xf7, 0x45, 0xbd, 0x78, - 0x52, 0xb2, 0x6a, 0x50, 0xdd, 0xb7, 0xdd, 0x3a, 0x03, 0xf4, 0x6f, 0x2f, 0xc5, 0x7c, 0x1f, 0xf4, - 0xc7, 0x1a, 0x40, 0x3d, 0x63, 0xfc, 0xff, 0xd9, 0x65, 0x1d, 0xc3, 0xd1, 0x61, 0x27, 0xac, 0x05, - 0xe8, 0x69, 0xae, 0xc8, 0x86, 0x72, 0x62, 0x82, 0x26, 0x1b, 0xf8, 0xe0, 0x6d, 0x49, 0xb0, 0x4c, - 0x0a, 0xf9, 0x87, 0x53, 0x28, 0x64, 0x52, 0xe8, 0x5d, 0x42, 0x51, 0xbe, 0xee, 0x75, 0xf2, 0x9f, - 0x9d, 0xc9, 0x83, 0x3b, 0xdd, 0x6c, 0xdc, 0xa3, 0xa8, 0xdb, 0xdd, 0xbf, 0xd8, 0xfc, 0x32, 0x73, - 0x9b, 0xad, 0xa9, 0xdd, 0x6d, 0x4d, 0xed, 0xe7, 0xd6, 0xd4, 0xbe, 0xee, 0xcc, 0xdc, 0xdd, 0xce, - 0xcc, 0x7d, 0xdf, 0x99, 0xb9, 0x9b, 0xb2, 0xfa, 0xd6, 0x8c, 0xcb, 0xf2, 0x53, 0xf1, 0xfc, 0x4f, - 0x00, 0x00, 0x00, 0xff, 0xff, 0xf1, 0x5f, 0x0a, 0x2f, 0x81, 0x04, 0x00, 0x00, + // 621 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x94, 0xdf, 0x6a, 0xdb, 0x3e, + 0x14, 0xc7, 0xe3, 0xfc, 0x6b, 0x72, 0xd2, 0xf4, 0xd7, 0x8a, 0xb6, 0x38, 0xe1, 0x87, 0x1b, 0xbc, + 0x5e, 0x04, 0x36, 0x62, 0xc8, 0x60, 0x8c, 0xed, 0x6a, 0xed, 0xca, 0xd6, 0xb1, 0xc2, 0xe6, 0x14, + 0x06, 0xbd, 0x31, 0x4e, 0xab, 0xa6, 0x06, 0xdb, 0x52, 0x24, 0x99, 0xb5, 0x6f, 0xb1, 0x57, 0x19, + 0xec, 0x21, 0x7a, 0xd9, 0xcb, 0x5d, 0x8d, 0xad, 0x7d, 0x91, 0xa1, 0x23, 0x3b, 0x8b, 0x59, 0x7b, + 0xb3, 0x9b, 0xc4, 0x3a, 0xdf, 0xcf, 0x91, 0xce, 0x3f, 0x09, 0xb6, 0xa2, 0xf4, 0x9c, 0x79, 0xfa, + 0x87, 0x4f, 0x3d, 0xc1, 0x4f, 0x47, 0x5c, 0x30, 0xc5, 0x48, 0x47, 0x5d, 0x84, 0x29, 0x93, 0x23, + 0x2d, 0xf4, 0x7b, 0x52, 0x31, 0x41, 0xbd, 0x38, 0x9c, 0xd2, 0x98, 0x4f, 0x3d, 0x75, 0xc5, 0xa9, + 0x34, 0x5c, 0x7f, 0x73, 0xc6, 0x66, 0x0c, 0x3f, 0x3d, 0xfd, 0x65, 0xac, 0x6e, 0x17, 0x3a, 0x87, + 0xe9, 0x39, 0xf3, 0xe9, 0x3c, 0xa3, 0x52, 0xb9, 0x5f, 0x6b, 0xb0, 0x6a, 0xd6, 0x92, 0xb3, 0x54, + 0x52, 0xf2, 0x0c, 0x00, 0x37, 0x0b, 0x24, 0x55, 0xd2, 0xb6, 0x06, 0xb5, 0x61, 0x67, 0xbc, 0x31, + 0xca, 0x8f, 0x3c, 0x79, 0xaf, 0xa5, 0x09, 0x55, 0x7b, 0xf5, 0xeb, 0x1f, 0x3b, 0x15, 0xbf, 0x1d, + 0xe7, 0x6b, 0x49, 0x76, 0xa1, 0xbb, 0xcf, 0x12, 0xce, 0x52, 0x9a, 0xaa, 0xe3, 0x2b, 0x4e, 0xed, + 0xea, 0xc0, 0x1a, 0xb6, 0xfd, 0xb2, 0x91, 0x3c, 0x81, 0x06, 0x06, 0x6c, 0xd7, 0x06, 0xd6, 0xb0, + 0x33, 0xde, 0x1e, 0x2d, 0xe5, 0x32, 0x9a, 0x68, 0x05, 0x83, 0x31, 0x90, 0xa6, 0x45, 0x16, 0x53, + 0x69, 0xd7, 0xef, 0xa1, 0x7d, 0xad, 0x18, 0x1a, 0x21, 0xf2, 0x16, 0xfe, 0x4b, 0xa8, 0x12, 0xd1, + 0x69, 0x90, 0x50, 0x15, 0x9e, 0x85, 0x2a, 0xb4, 0x1b, 0xe8, 0xb7, 0x53, 0xf2, 0x3b, 0x42, 0xe6, + 0x28, 0x47, 0x70, 0x83, 0xb5, 0xa4, 0x64, 0x23, 0x63, 0x58, 0x51, 0xa1, 0x98, 0xe9, 0x02, 0x34, + 0x71, 0x07, 0xbb, 0xb4, 0xc3, 0xb1, 0xd1, 0xd0, 0xb5, 0x00, 0xc9, 0x73, 0x68, 0xd3, 0x4b, 0x9a, + 0xf0, 0x38, 0x14, 0xd2, 0x5e, 0x41, 0xaf, 0x7e, 0xc9, 0xeb, 0xa0, 0x50, 0xd1, 0xef, 0x0f, 0x4c, + 0x3c, 0x68, 0xcc, 0x33, 0x2a, 0xae, 0xec, 0x16, 0x7a, 0xf5, 0x4a, 0x5e, 0x1f, 0xb5, 0xf2, 0xea, + 0xc3, 0xa1, 0x49, 0x14, 0x39, 0xf7, 0x5b, 0x15, 0xda, 0x8b, 0x5a, 0x91, 0x1e, 0xb4, 0x92, 0x28, + 0x0d, 0x54, 0x94, 0x50, 0xdb, 0x1a, 0x58, 0xc3, 0x9a, 0xbf, 0x92, 0x44, 0xe9, 0x71, 0x94, 0x50, + 0x94, 0xc2, 0x4b, 0x23, 0x55, 0x73, 0x29, 0xbc, 0x44, 0xe9, 0x31, 0x6c, 0xc8, 0x8c, 0x73, 0x26, + 0x94, 0x0c, 0xe4, 0x45, 0x28, 0xce, 0xa2, 0x74, 0x86, 0x4d, 0x69, 0xf9, 0xeb, 0x85, 0x30, 0xc9, + 0xed, 0xe4, 0x00, 0x76, 0x16, 0xf0, 0xe7, 0x48, 0x5d, 0xb0, 0x4c, 0x05, 0x82, 0xf2, 0x38, 0x3a, + 0x0d, 0x03, 0x9c, 0x00, 0x89, 0x95, 0x6e, 0xf9, 0xff, 0x17, 0xd8, 0x27, 0x43, 0xf9, 0x06, 0xc2, + 0xa9, 0x91, 0xe4, 0x05, 0x80, 0x92, 0x67, 0xd3, 0x40, 0x27, 0xa6, 0x2b, 0xab, 0x47, 0x6b, 0xab, + 0x5c, 0xd9, 0xc9, 0xeb, 0x3d, 0x9d, 0x54, 0x31, 0x5e, 0x1a, 0xd7, 0x6b, 0x49, 0x1e, 0x41, 0xb7, + 0x38, 0x71, 0x26, 0x58, 0xc6, 0xb1, 0xc4, 0x6d, 0x7f, 0x35, 0x37, 0xbe, 0xd1, 0x36, 0xb2, 0x0d, + 0xcd, 0x79, 0xc6, 0x44, 0x96, 0x60, 0x29, 0x1b, 0x7e, 0xbe, 0x7a, 0x57, 0x6f, 0xd5, 0xd7, 0x1b, + 0x6e, 0x07, 0xda, 0x8b, 0x99, 0x71, 0x37, 0x81, 0xfc, 0x3d, 0x08, 0xfa, 0x72, 0x2c, 0x35, 0xd7, + 0x3d, 0x80, 0x6e, 0xa9, 0x6b, 0xff, 0x56, 0x6b, 0x77, 0x0d, 0x56, 0x97, 0xdb, 0xe8, 0xce, 0xa1, + 0x55, 0x24, 0x4a, 0x3c, 0x68, 0xe6, 0x15, 0xb4, 0xb0, 0xfb, 0x0f, 0x5e, 0xb5, 0x1c, 0x2b, 0x85, + 0x50, 0x7d, 0x38, 0x84, 0x5a, 0x29, 0x84, 0xf1, 0x3e, 0xd4, 0xf1, 0xb8, 0x97, 0xf9, 0x7f, 0x79, + 0xa0, 0x97, 0x1e, 0x84, 0x7e, 0xef, 0x1e, 0xc5, 0x3c, 0x0d, 0x7b, 0xbb, 0xd7, 0xbf, 0x9c, 0xca, + 0xf5, 0xad, 0x63, 0xdd, 0xdc, 0x3a, 0xd6, 0xcf, 0x5b, 0xc7, 0xfa, 0x72, 0xe7, 0x54, 0x6e, 0xee, + 0x9c, 0xca, 0xf7, 0x3b, 0xa7, 0x72, 0xd2, 0x34, 0x0f, 0xd5, 0xb4, 0x89, 0xef, 0xcc, 0xd3, 0xdf, + 0x01, 0x00, 0x00, 0xff, 0xff, 0xaf, 0x38, 0x92, 0xb8, 0xbe, 0x04, 0x00, 0x00, } // Reference imports to suppress errors if they are not otherwise used. @@ -681,6 +693,18 @@ func (m *StoreInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l + if m.Quorum != 0 { + i = encodeVarintRpc(dAtA, i, uint64(m.Quorum)) + i-- + dAtA[i] = 0x40 + } + if len(m.ReplicaGroup) > 0 { + i -= len(m.ReplicaGroup) + copy(dAtA[i:], m.ReplicaGroup) + i = encodeVarintRpc(dAtA, i, uint64(len(m.ReplicaGroup))) + i-- + dAtA[i] = 0x3a + } if len(m.TsdbInfos) > 0 { for iNdEx := len(m.TsdbInfos) - 1; iNdEx >= 0; iNdEx-- { { @@ -983,6 +1007,13 @@ func (m *StoreInfo) Size() (n int) { n += 1 + l + sovRpc(uint64(l)) } } + l = len(m.ReplicaGroup) + if l > 0 { + n += 1 + l + sovRpc(uint64(l)) + } + if m.Quorum != 0 { + n += 1 + sovRpc(uint64(m.Quorum)) + } return n } @@ -1583,6 +1614,57 @@ func (m *StoreInfo) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 7: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ReplicaGroup", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowRpc + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthRpc + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthRpc + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ReplicaGroup = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 8: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Quorum", wireType) + } + m.Quorum = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowRpc + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Quorum |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipRpc(dAtA[iNdEx:]) diff --git a/pkg/info/infopb/rpc.proto b/pkg/info/infopb/rpc.proto index 9f0db3709da..43179c25de1 100644 --- a/pkg/info/infopb/rpc.proto +++ b/pkg/info/infopb/rpc.proto @@ -64,6 +64,18 @@ message StoreInfo { // TSDBInfos holds metadata for all TSDBs exposed by the store. repeated TSDBInfo tsdb_infos = 6 [(gogoproto.nullable) = false]; + + // Replica topology hints for GROUP_REPLICA partial response strategy. + // Stores with the same replica_group value hold replicated data. + // When set, the query layer can tolerate failures as long as enough + // replicas in the group are healthy (determined by quorum). + // If empty, the store is treated as a "must-success" store. + string replica_group = 7; + + // Minimum number of healthy stores required per replica group. + // Only meaningful when replica_group is set. + // If 0 or not set, the store is treated as a "must-success" store. + int32 quorum = 8; } // RulesInfo holds the metadata related to Rules API exposed by the component. diff --git a/pkg/query/endpointset.go b/pkg/query/endpointset.go index fc25e0f1d90..c77a1cd6246 100644 --- a/pkg/query/endpointset.go +++ b/pkg/query/endpointset.go @@ -272,10 +272,6 @@ type EndpointSet struct { endpointInfoTimeout time.Duration unhealthyEndpointTimeout time.Duration - // quorumLabelName is the external label name whose value specifies the quorum requirement. - // Used for the thanos_query_endpoint_groups metric. - quorumLabelName string - updateMtx sync.Mutex endpointsMtx sync.RWMutex @@ -298,7 +294,6 @@ func NewEndpointSet( dialOpts []grpc.DialOption, unhealthyEndpointTimeout time.Duration, endpointInfoTimeout time.Duration, - quorumLabelName string, endpointMetricLabels ...string, ) *EndpointSet { endpointsMetric := newEndpointSetNodeCollector(logger, endpointMetricLabels...) @@ -322,7 +317,6 @@ func NewEndpointSet( dialOpts: dialOpts, endpointInfoTimeout: endpointInfoTimeout, unhealthyEndpointTimeout: unhealthyEndpointTimeout, - quorumLabelName: quorumLabelName, endpointSpec: func() map[string]*GRPCEndpointSpec { specs := make(map[string]*GRPCEndpointSpec) for _, s := range endpointSpecs() { @@ -446,16 +440,10 @@ func (e *EndpointSet) Update(ctx context.Context) { } stats[er.ComponentType().String()][extLset]++ - // Track endpoints by quorum label value for thanos_query_endpoint_groups metric. - quorumValue := "0" // Default for endpoints without quorum label - if e.quorumLabelName != "" { - for _, lset := range er.LabelSets() { - if v := lset.Get(e.quorumLabelName); v != "" { - quorumValue = v - break - } - } - } + // Track endpoints by quorum value for thanos_query_endpoint_groups metric. + // Quorum is now a first-class field from StoreInfo. + ri := er.ReplicaInfo() + quorumValue := fmt.Sprintf("%d", ri.Quorum) storesByQuorum[quorumValue]++ } @@ -534,8 +522,8 @@ func (e *EndpointSet) GetStoreClients() []store.Client { StoreClient: storepb.NewStoreClient(er.cc), addr: er.addr, metadata: er.metadata, - groupKey: er.GroupKey(), - replicaKey: er.ReplicaKey(), + groupKey: er.groupKey, + replicaKey: er.replicaKey, status: er.status, }) er.mtx.RUnlock() @@ -663,12 +651,31 @@ type endpointRef struct { logger log.Logger } -func (er *endpointRef) GroupKey() string { - return er.groupKey -} +// ReplicaInfo returns replica topology hints for GROUP_REPLICA partial response strategy. +// It unifies DNS-based grouping (legacy) with first-class StoreInfo fields, +// preferring StoreInfo fields when available. +func (er *endpointRef) ReplicaInfo() store.ReplicaInfo { + er.mtx.RLock() + defer er.mtx.RUnlock() + + // Prefer first-class fields from StoreInfo if available + if er.metadata != nil && er.metadata.Store != nil { + if rg := er.metadata.Store.ReplicaGroup; rg != "" { + return store.ReplicaInfo{ + Group: rg, + Replica: er.replicaKey, // Use DNS-based replica key for identification + Quorum: int(er.metadata.Store.Quorum), + } + } + } -func (er *endpointRef) ReplicaKey() string { - return er.replicaKey + // Fall back to DNS-based grouping (legacy) + // Quorum=0 means must-success semantics + return store.ReplicaInfo{ + Group: er.groupKey, + Replica: er.replicaKey, + Quorum: 0, + } } // newEndpointRef creates a new endpointRef with a gRPC channel to the given the IP address. diff --git a/pkg/query/endpointset_test.go b/pkg/query/endpointset_test.go index 551f701d1ce..b8bd507f903 100644 --- a/pkg/query/endpointset_test.go +++ b/pkg/query/endpointset_test.go @@ -691,7 +691,7 @@ func TestEndpointSetUpdate_AvailabilityScenarios(t *testing.T) { } return specs }, - testGRPCOpts, time.Minute, 2*time.Second, "") + testGRPCOpts, time.Minute, 2*time.Second) defer endpointSet.Close() // Initial update. @@ -1064,7 +1064,7 @@ func TestEndpointSet_Update_NoneAvailable(t *testing.T) { } return specs }, - testGRPCOpts, time.Minute, 2*time.Second, "") + testGRPCOpts, time.Minute, 2*time.Second) defer endpointSet.Close() // Should not matter how many of these we run. @@ -1175,7 +1175,7 @@ func TestEndpoint_Update_QuerierStrict(t *testing.T) { NewGRPCEndpointSpec(discoveredEndpointAddr[1], false), NewGRPCEndpointSpec(discoveredEndpointAddr[2], true), } - }, testGRPCOpts, time.Minute, 1*time.Second, "") + }, testGRPCOpts, time.Minute, 1*time.Second) defer endpointSet.Close() // Initial update. @@ -1356,7 +1356,7 @@ func TestEndpointSet_APIs_Discovery(t *testing.T) { return tc.states[currentState].endpointSpec() }, - testGRPCOpts, time.Minute, 2*time.Second, "") + testGRPCOpts, time.Minute, 2*time.Second) defer endpointSet.Close() @@ -1548,7 +1548,7 @@ func makeEndpointSet(discoveredEndpointAddr []string, strict bool, now nowFunc, } return specs }, - testGRPCOpts, time.Minute, time.Second, "", metricLabels...) + testGRPCOpts, time.Minute, time.Second, metricLabels...) return endpointSet } diff --git a/pkg/receive/multitsdb.go b/pkg/receive/multitsdb.go index 580c1ce98bb..a8e48ddb8eb 100644 --- a/pkg/receive/multitsdb.go +++ b/pkg/receive/multitsdb.go @@ -53,7 +53,8 @@ type MultiTSDB struct { tsdbOpts *tsdb.Options tenantLabelName string labels labels.Labels - infoOnlyLabels labels.Labels // Labels only exposed via InfoAPI, not merged into series + replicaGroup string // Replica group identifier for GROUP_REPLICA partial response strategy + quorum int32 // Minimum healthy stores required per group bucket objstore.Bucket mtx *sync.RWMutex @@ -111,11 +112,12 @@ func WithMatchersCache(cache storecache.MatchersCache) MultiTSDBOption { } } -// WithInfoOnlyLabels sets labels that are only exposed via InfoAPI (for store discovery), -// but NOT merged into series responses. Useful for metadata like replica group/quorum. -func WithInfoOnlyLabels(lset labels.Labels) MultiTSDBOption { +// WithReplicaGroup sets the replica group identifier and quorum for GROUP_REPLICA +// partial response strategy. Stores with the same replica_group value hold replicated data. +func WithReplicaGroup(replicaGroup string, quorum int) MultiTSDBOption { return func(s *MultiTSDB) { - s.infoOnlyLabels = lset + s.replicaGroup = replicaGroup + s.quorum = int32(quorum) } } @@ -284,12 +286,10 @@ func (l *localClient) LabelValues(ctx context.Context, in *storepb.LabelValuesRe return l.store.LabelValues(ctx, in) } -func (l *localClient) GroupKey() string { - return "" -} - -func (l *localClient) ReplicaKey() string { - return "" +// ReplicaInfo returns replica topology hints for GROUP_REPLICA partial response strategy. +// It delegates to the underlying TSDBStore which has the first-class replica_group and quorum fields. +func (l *localClient) ReplicaInfo() store.ReplicaInfo { + return l.store.ReplicaInfo() } func (l *localClient) Matches(matchers []*labels.Matcher) bool { @@ -856,8 +856,8 @@ func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant options = append(options, store.WithMatcherCacheInstance(t.matcherCache)) } tsdbStore := store.NewTSDBStore(logger, s, component.Receive, lset, options...) - if t.infoOnlyLabels.Len() > 0 { - tsdbStore.SetInfoOnlyLset(t.infoOnlyLabels) + if t.replicaGroup != "" { + tsdbStore.SetReplicaInfo(t.replicaGroup, t.quorum) } tenant.set(tsdbStore, s, ship, exemplars.NewTSDB(s, lset)) t.addTenantLocked(tenantID, tenant) // need to update the client list once store is ready & client != nil diff --git a/pkg/store/proxy.go b/pkg/store/proxy.go index 7f4782c202b..98a1a355b5c 100644 --- a/pkg/store/proxy.go +++ b/pkg/store/proxy.go @@ -7,7 +7,6 @@ import ( "context" "fmt" "math" - "strconv" "strings" "sync" "time" @@ -50,6 +49,10 @@ const metricNameLabel = "__name__" // This can happen with Query servers trees and external labels. var ErrorNoStoresMatched = errors.New("No StoreAPIs matched for this query") +// ReplicaInfo is an alias to storepb.ReplicaInfo for convenience. +// It contains replica topology hints for GROUP_REPLICA partial response strategy. +type ReplicaInfo = storepb.ReplicaInfo + // Client holds meta information about a store. type Client interface { // StoreClient to access the store. @@ -78,14 +81,10 @@ type Client interface { // represents a local client (server-as-client) and has no remote address. Addr() (addr string, isLocalClient bool) - // ReplicaKey returns replica name of the store client. A replica consists of a set of endpoints belong to the - // same replica. E.g, "pantheon-db-rep0", "pantheon-db-rep1", "long-range-store". - ReplicaKey() string - - // GroupKey returns group name of the store client. A group defines a group of replicas that belong to the - // same group. E.g. "pantheon-db" has replicas "pantheon-db-rep0", "pantheon-db-rep1". - // "long-range-store" has only one replica, "long-range-store". - GroupKey() string + // ReplicaInfo returns replica topology hints for GROUP_REPLICA partial response strategy. + // Implementations should unify DNS-based grouping (legacy) with first-class StoreInfo fields, + // preferring StoreInfo fields when available. + ReplicaInfo() ReplicaInfo // Matches returns true if provided label matchers are allowed in the store. Matches(matches []*labels.Matcher) bool @@ -112,12 +111,6 @@ type ProxyStore struct { forwardPartialStrategy bool exclusiveExternalLabels []string matcherCache storecache.MatchersCache - - // groupReplicaGroupLabel and groupReplicaQuorumLabel are the external label names - // used for label-based group/quorum identification in GROUP_REPLICA partial response strategy. - // When both are set, the strategy uses these labels from LabelSets() instead of DNS-based parsing. - groupReplicaGroupLabel string - groupReplicaQuorumLabel string } type proxyStoreMetrics struct { @@ -272,15 +265,6 @@ func WithMatcherCache(cache storecache.MatchersCache) ProxyStoreOption { } } -// WithGroupReplicaLabels sets the external label names for label-based group/quorum -// identification in GROUP_REPLICA partial response strategy. -func WithGroupReplicaLabels(groupLabel, quorumLabel string) ProxyStoreOption { - return func(s *ProxyStore) { - s.groupReplicaGroupLabel = groupLabel - s.groupReplicaQuorumLabel = quorumLabel - } -} - // 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( @@ -322,43 +306,6 @@ func NewProxyStore( return s } -// getGroupKey returns the group key for a store client. -// If groupReplicaGroupLabel is configured, it looks up the label value from LabelSets(). -// Otherwise, it falls back to the DNS-based GroupKey() method. -// For stores without the configured label, it returns empty string to indicate a must-success store. -func (s *ProxyStore) getGroupKey(st Client) string { - if s.groupReplicaGroupLabel != "" { - for _, lset := range st.LabelSets() { - if v := lset.Get(s.groupReplicaGroupLabel); v != "" { - return v - } - } - // Store doesn't have the label - return empty to indicate must-success store - return "" - } - return st.GroupKey() -} - -// getQuorum returns the quorum requirement for a store client. -// If groupReplicaQuorumLabel is configured, it parses the label value as an integer. -// Returns 0 if the label is missing, empty, or invalid (non-integer or <1). -// A return value of 0 indicates the store should be treated as a "must-success" store. -func (s *ProxyStore) getQuorum(st Client) int { - if s.groupReplicaQuorumLabel == "" { - return 0 - } - for _, lset := range st.LabelSets() { - if v := lset.Get(s.groupReplicaQuorumLabel); v != "" { - quorum, err := strconv.Atoi(v) - if err != nil || quorum < 1 { - return 0 // Invalid quorum, treat as must-success - } - return quorum - } - } - return 0 // Label not found, treat as must-success -} - func (s *ProxyStore) LabelSet() []labelpb.ZLabelSet { stores := s.stores() if len(stores) == 0 { @@ -514,18 +461,18 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. stores []Client storeLabelSets []labels.Labels ) - // For label-based quorum strategy: + // For quorum-based strategy (stores with Quorum() > 0 via StoreInfo): // - groupStores[groupKey] = total stores in group // - groupQuorum[groupKey] = quorum requirement for group // - groupFailed[groupKey] = failed stores in group - // - mustSuccessStores = stores without labels or invalid quorum (must all succeed) + // - mustSuccessStores = stores without quorum (Quorum=0) must all succeed groupStores := make(map[string]int) groupQuorum := make(map[string]int) groupFailed := make(map[string]int) mustSuccessStores := make(map[Client]bool) totalFailedStores := 0 - // For legacy DNS-based strategy (when label flags not set): + // For legacy DNS-based strategy (stores with Quorum=0): // groupReplicaStores[groupKey][replicaKey] = number of stores // failedStores[groupKey][replicaKey] = number of failures groupReplicaStores := make(map[string]map[string]int) @@ -537,9 +484,6 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. mp[key1][key2]++ } - // Check if we're using label-based quorum strategy - useLabelBasedStrategy := s.groupReplicaGroupLabel != "" && s.groupReplicaQuorumLabel != "" - stores, storeLabelSets, storeDebugMsgs := s.matchingStores(ctx, originalRequest.MinTime, originalRequest.MaxTime, matchers) s.metrics.storesPerQueryAfterFiltering.Set(float64(len(stores))) @@ -548,21 +492,21 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. s.metrics.storesPerQueryAfterEELFiltering.Set(float64(len(stores))) for _, st := range stores { - if useLabelBasedStrategy { - groupKey := s.getGroupKey(st) - quorum := s.getQuorum(st) - if groupKey == "" || quorum == 0 { - // No valid group/quorum labels - must-success store + ri := st.ReplicaInfo() + if ri.IsMustSuccess() { + // Must-success store (quorum=0) - use legacy DNS-based strategy + bumpCounter(ri.Group, ri.Replica, groupReplicaStores) + } else { + // Quorum-based strategy + if ri.Group == "" { + // No group but has quorum - treat as must-success mustSuccessStores[st] = true } else { - groupStores[groupKey]++ - if _, exists := groupQuorum[groupKey]; !exists { - groupQuorum[groupKey] = quorum + groupStores[ri.Group]++ + if _, exists := groupQuorum[ri.Group]; !exists { + groupQuorum[ri.Group] = ri.Quorum } } - } else { - // Legacy DNS-based strategy - bumpCounter(st.GroupKey(), st.ReplicaKey(), groupReplicaStores) } } if len(stores) == 0 { @@ -597,81 +541,76 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. storeResponses := make([]respSet, 0, len(stores)) checkGroupReplicaErrors := func(st Client, err error) error { - if useLabelBasedStrategy { - // Label-based quorum strategy - if mustSuccessStores[st] { - // Must-success store failed - abort immediately - addr, _ := st.Addr() - msg := "Must-success store failed (no valid group/quorum labels)" - level.Error(reqLogger).Log("msg", msg, "store", addr) - return fmt.Errorf("%s store=%s: %w", msg, addr, err) + ri := st.ReplicaInfo() + if ri.IsMustSuccess() { + // Legacy DNS-based strategy (must-success store) + if len(failedStores[ri.Group]) > 1 { + msg := "Multiple replicas have failures for the same group" + replicas := fmt.Sprintf("%+v", failedStores[ri.Group]) + level.Error(reqLogger).Log( + "msg", msg, + "group", ri.Group, + "replicas", replicas, + ) + return fmt.Errorf("%s group=%s replicas=%s: %w", msg, ri.Group, replicas, err) } - - groupKey := s.getGroupKey(st) - groupFailed[groupKey]++ - healthy := groupStores[groupKey] - groupFailed[groupKey] - quorum := groupQuorum[groupKey] - - if healthy < quorum { - msg := "Group does not meet quorum requirement" + if len(groupReplicaStores[ri.Group]) == 1 && failedStores[ri.Group][ri.Replica] > 1 { + msg := "A group with single replica has multiple failures" + replicas := fmt.Sprintf("%+v", failedStores[ri.Group]) level.Error(reqLogger).Log( "msg", msg, - "group", groupKey, - "healthy", healthy, - "quorum", quorum, - "total", groupStores[groupKey], - "failed", groupFailed[groupKey], + "group", ri.Group, + "replicas", replicas, ) - return fmt.Errorf("%s group=%s healthy=%d quorum=%d: %w", msg, groupKey, healthy, quorum, err) + return fmt.Errorf("%s group=%s replicas=%s: %w", msg, ri.Group, replicas, err) } return nil } - // Legacy DNS-based strategy - groupKey := st.GroupKey() - replicaKey := st.ReplicaKey() - if len(failedStores[groupKey]) > 1 { - msg := "Multiple replicas have failures for the same group" - replicas := fmt.Sprintf("%+v", failedStores[groupKey]) - level.Error(reqLogger).Log( - "msg", msg, - "group", groupKey, - "replicas", replicas, - ) - return fmt.Errorf("%s group=%s replicas=%s: %w", msg, groupKey, replicas, err) + // Quorum-based strategy + if mustSuccessStores[st] { + // Must-success store failed - abort immediately + addr, _ := st.Addr() + msg := "Must-success store failed (no valid replica group)" + level.Error(reqLogger).Log("msg", msg, "store", addr) + return fmt.Errorf("%s store=%s: %w", msg, addr, err) } - if len(groupReplicaStores[groupKey]) == 1 && failedStores[groupKey][replicaKey] > 1 { - msg := "A group with single replica has multiple failures" - replicas := fmt.Sprintf("%+v", failedStores[groupKey]) + + groupFailed[ri.Group]++ + healthy := groupStores[ri.Group] - groupFailed[ri.Group] + requiredQuorum := groupQuorum[ri.Group] + + if healthy < requiredQuorum { + msg := "Group does not meet quorum requirement" level.Error(reqLogger).Log( "msg", msg, - "group", groupKey, - "replicas", replicas, + "group", ri.Group, + "healthy", healthy, + "quorum", requiredQuorum, + "total", groupStores[ri.Group], + "failed", groupFailed[ri.Group], ) - return fmt.Errorf("%s group=%s replicas=%s: %w", msg, groupKey, replicas, err) + return fmt.Errorf("%s group=%s healthy=%d quorum=%d: %w", msg, ri.Group, healthy, requiredQuorum, err) } return nil } logGroupReplicaErrors := func() { - if useLabelBasedStrategy { - if len(groupFailed) > 0 { - level.Warn(s.logger).Log("msg", "Group/quorum errors", - "mode", "label-based", - "group_failures", fmt.Sprintf("%+v", groupFailed), - "total_failed_stores", totalFailedStores, - ) - s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) - } - } else { - if len(failedStores) > 0 { - level.Warn(s.logger).Log("msg", "Group/replica errors", - "mode", "dns-based", - "errors", fmt.Sprintf("%+v", failedStores), - "total_failed_stores", totalFailedStores, - ) - s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) - } + if len(groupFailed) > 0 { + level.Warn(s.logger).Log("msg", "Group/quorum errors", + "mode", "quorum-based", + "group_failures", fmt.Sprintf("%+v", groupFailed), + "total_failed_stores", totalFailedStores, + ) + s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) + } + if len(failedStores) > 0 { + level.Warn(s.logger).Log("msg", "Group/replica errors", + "mode", "dns-based", + "errors", fmt.Sprintf("%+v", failedStores), + "total_failed_stores", totalFailedStores, + ) + s.metrics.failedStoresPerQuery.Set(float64(totalFailedStores)) } } defer logGroupReplicaErrors() @@ -735,19 +674,19 @@ func (s *ProxyStore) Series(originalRequest *storepb.SeriesRequest, srv storepb. } totalFailedStores++ - if useLabelBasedStrategy { - groupKey := s.getGroupKey(st) + ri := st.ReplicaInfo() + if ri.IsMustSuccess() { + // Legacy DNS-based strategy (must-success store) + level.Warn(s.logger).Log("msg", "Store failure", "group", ri.Group, "replica", ri.Replica, "err", err) + s.metrics.storeFailureCount.WithLabelValues(ri.Group, ri.Replica).Inc() + bumpCounter(ri.Group, ri.Replica, failedStores) + } else { + // Quorum-based strategy addr, _ := st.Addr() - level.Warn(s.logger).Log("msg", "Store failure", "group", groupKey, "store", addr, "err", err) - // Note: We don't record to storeFailureCount metric in label-based mode + level.Warn(s.logger).Log("msg", "Store failure", "group", ri.Group, "store", addr, "err", err) + // Note: We don't record to storeFailureCount metric in quorum-based mode // because the existing metric uses (group, replica) labels which have // different semantics than (group, addr). - } else { - groupKey := st.GroupKey() - replicaKey := st.ReplicaKey() - level.Warn(s.logger).Log("msg", "Store failure", "group", groupKey, "replica", replicaKey, "err", err) - s.metrics.storeFailureCount.WithLabelValues(groupKey, replicaKey).Inc() - bumpCounter(groupKey, replicaKey, failedStores) } if originalRequest.PartialResponseStrategy == storepb.PartialResponseStrategy_GROUP_REPLICA { if checkGroupReplicaErrors(st, err) != nil { diff --git a/pkg/store/proxy_test.go b/pkg/store/proxy_test.go index bab8e797bbc..602c349f1d0 100644 --- a/pkg/store/proxy_test.go +++ b/pkg/store/proxy_test.go @@ -3560,143 +3560,95 @@ func TestDedupRespHeap_QuorumChunkDedup(t *testing.T) { func TestProxyStore_GetGroupKeyQuorum(t *testing.T) { t.Parallel() + // Test that getGroupKey and getQuorum use first-class fields from StoreInfo + // when available, otherwise fall back to DNS-based grouping. for _, tc := range []struct { - name string - groupReplicaGroupLabel string - groupReplicaQuorumLabel string - client *storetestutil.TestClient - expectedGroupKey string - expectedQuorum int + name string + client *storetestutil.TestClient + expectedGroup string + expectedReplica string + expectedQuorum int + expectMustSuccess bool }{ { - name: "no labels configured - fallback to DNS-based group key", - groupReplicaGroupLabel: "", - groupReplicaQuorumLabel: "", + name: "no ReplicaGroup set - fallback to DNS-based GroupKey", client: &storetestutil.TestClient{ - Name: "receive-rep0-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-rep0", - ReplicaKeyStr: "receive-rep0-0", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "2")}, + Name: "receive-rep0-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-rep0", + ReplicaKeyStr: "receive-rep0-0", + ReplicaGroupStr: "", // Not set - triggers DNS fallback + QuorumValue: 0, // Ignored when ReplicaGroupStr is empty }, - expectedGroupKey: "receive-rep0", - expectedQuorum: 0, // No quorum label configured, returns 0 + expectedGroup: "receive-rep0", // Falls back to GroupKeyStr + expectedReplica: "receive-rep0-0", + expectedQuorum: 0, // DNS fallback always has quorum=0 + expectMustSuccess: true, }, { - name: "labels configured - use label-based group key and quorum", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", + name: "ReplicaGroup and Quorum set - use first-class fields", client: &storetestutil.TestClient{ - Name: "receive-rep0-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-rep0", - ReplicaKeyStr: "receive-rep0-0", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "2")}, + Name: "receive-rep0-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-rep0", // Ignored when ReplicaGroupStr is set + ReplicaKeyStr: "receive-rep0-0", + ReplicaGroupStr: "receive-0", + QuorumValue: 2, }, - expectedGroupKey: "receive-0", - expectedQuorum: 2, + expectedGroup: "receive-0", // Uses ReplicaGroupStr + expectedReplica: "receive-rep0-0", + expectedQuorum: 2, // Uses QuorumValue + expectMustSuccess: false, }, { - name: "labels configured but store missing labels - must-success store", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", + name: "store without ReplicaGroup/Quorum - must-success via DNS fallback", client: &storetestutil.TestClient{ - Name: "bucket-store.svc.cluster.local:10901", - GroupKeyStr: "bucket-store", - ReplicaKeyStr: "bucket-store", - ExtLset: []labels.Labels{labels.FromStrings("cluster", "us-east")}, // no receive_group/quorum + Name: "bucket-store.svc.cluster.local:10901", + GroupKeyStr: "bucket-store", + ReplicaKeyStr: "bucket-store", + ReplicaGroupStr: "", + QuorumValue: 0, }, - expectedGroupKey: "", // Empty string indicates must-success store - expectedQuorum: 0, // No quorum label, must-success store + expectedGroup: "bucket-store", // Falls back to GroupKeyStr + expectedReplica: "bucket-store", + expectedQuorum: 0, // Must-success store + expectMustSuccess: true, }, { - name: "labels configured - multiple label sets, first match used", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", + name: "ReplicaGroup set but Quorum is 0 - use ReplicaGroup, must-success", client: &storetestutil.TestClient{ - Name: "receive-rep1-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-rep1", - ReplicaKeyStr: "receive-rep1-0", - ExtLset: []labels.Labels{ - labels.FromStrings("receive_group", "receive-0", "quorum", "3"), - labels.FromStrings("cluster", "us-east"), - }, - }, - expectedGroupKey: "receive-0", - expectedQuorum: 3, - }, - { - name: "labels configured - store has only group label (no quorum)", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - client: &storetestutil.TestClient{ - Name: "partial-store.svc.cluster.local:10901", - GroupKeyStr: "partial-store", - ReplicaKeyStr: "partial-store", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0")}, // no quorum label - }, - expectedGroupKey: "receive-0", - expectedQuorum: 0, // No quorum label, must-success store within group - }, - { - name: "invalid quorum value (non-integer) - must-success store", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - client: &storetestutil.TestClient{ - Name: "receive-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-0", - ReplicaKeyStr: "receive-0", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "invalid")}, + Name: "partial-store.svc.cluster.local:10901", + GroupKeyStr: "partial-store", + ReplicaKeyStr: "partial-store", + ReplicaGroupStr: "receive-0", + QuorumValue: 0, // Not set, must-success }, - expectedGroupKey: "receive-0", - expectedQuorum: 0, // Invalid quorum value, treat as must-success + expectedGroup: "receive-0", // Uses ReplicaGroupStr + expectedReplica: "partial-store", + expectedQuorum: 0, // Must-success store + expectMustSuccess: true, }, { - name: "quorum value less than 1 - must-success store", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", + name: "high quorum value", client: &storetestutil.TestClient{ - Name: "receive-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-0", - ReplicaKeyStr: "receive-0", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "0")}, + Name: "receive-0.receive.svc.cluster.local:10901", + GroupKeyStr: "receive-0", + ReplicaKeyStr: "receive-0", + ReplicaGroupStr: "group-a", + QuorumValue: 5, }, - expectedGroupKey: "receive-0", - expectedQuorum: 0, // quorum < 1, treat as must-success - }, - { - name: "negative quorum value - must-success store", - groupReplicaGroupLabel: "receive_group", - groupReplicaQuorumLabel: "quorum", - client: &storetestutil.TestClient{ - Name: "receive-0.receive.svc.cluster.local:10901", - GroupKeyStr: "receive-0", - ReplicaKeyStr: "receive-0", - ExtLset: []labels.Labels{labels.FromStrings("receive_group", "receive-0", "quorum", "-1")}, - }, - expectedGroupKey: "receive-0", - expectedQuorum: 0, // Negative quorum, treat as must-success + expectedGroup: "group-a", + expectedReplica: "receive-0", + expectedQuorum: 5, + expectMustSuccess: false, }, } { t.Run(tc.name, func(t *testing.T) { - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - - // Create ProxyStore with the configured labels - proxy := NewProxyStore( - logger, - reg, - func() []Client { return []Client{tc.client} }, - component.Query, - labels.EmptyLabels(), - 0, - EagerRetrieval, - WithGroupReplicaLabels(tc.groupReplicaGroupLabel, tc.groupReplicaQuorumLabel), - ) - - gotGroupKey := proxy.getGroupKey(tc.client) - gotQuorum := proxy.getQuorum(tc.client) + // Test the unified ReplicaInfo() method directly + ri := tc.client.ReplicaInfo() - testutil.Equals(t, tc.expectedGroupKey, gotGroupKey) - testutil.Equals(t, tc.expectedQuorum, gotQuorum) + testutil.Equals(t, tc.expectedGroup, ri.Group) + testutil.Equals(t, tc.expectedReplica, ri.Replica) + testutil.Equals(t, tc.expectedQuorum, ri.Quorum) + testutil.Equals(t, tc.expectMustSuccess, ri.IsMustSuccess()) }) } } diff --git a/pkg/store/storepb/replica_info.go b/pkg/store/storepb/replica_info.go new file mode 100644 index 00000000000..1830c964d54 --- /dev/null +++ b/pkg/store/storepb/replica_info.go @@ -0,0 +1,26 @@ +// Copyright (c) The Thanos Authors. +// Licensed under the Apache License 2.0. + +package storepb + +// ReplicaInfo contains replica topology hints for GROUP_REPLICA partial response strategy. +// It unifies DNS-based grouping (legacy) with first-class StoreInfo fields. +type ReplicaInfo struct { + // Group identifies stores that hold replicated data. Stores with the same Group + // value are considered replicas of each other. + // Examples: "pantheon-db", "long-range-store" + Group string + + // Replica identifies this specific store within a group. + // Examples: "pantheon-db-rep0", "pantheon-db-rep1" + Replica string + + // Quorum is the minimum number of healthy stores required per group. + // A value of 0 means "must-success" - the store must respond successfully. + Quorum int +} + +// IsMustSuccess returns true if this store must respond successfully (quorum=0). +func (ri ReplicaInfo) IsMustSuccess() bool { + return ri.Quorum == 0 +} diff --git a/pkg/store/storepb/testutil/client.go b/pkg/store/storepb/testutil/client.go index 4587c42f375..0d7bf600cdd 100644 --- a/pkg/store/storepb/testutil/client.go +++ b/pkg/store/storepb/testutil/client.go @@ -22,8 +22,16 @@ type TestClient struct { IsLocalStore bool StoreTSDBInfos []infopb.TSDBInfo StoreFilterNotMatches bool - GroupKeyStr string - ReplicaKeyStr string + + // ReplicaInfo fields for GROUP_REPLICA strategy testing. + // GroupKeyStr is used as Group when ReplicaGroupStr is empty (DNS-based fallback). + // ReplicaGroupStr is the first-class StoreInfo field - if set, it takes precedence. + // ReplicaKeyStr is used as Replica in ReplicaInfo. + // QuorumValue is used as Quorum when ReplicaGroupStr is set (0 = must-success). + GroupKeyStr string + ReplicaKeyStr string + ReplicaGroupStr string // First-class StoreInfo field (takes precedence over GroupKeyStr) + QuorumValue int } func (c TestClient) LabelSets() []labels.Labels { return c.ExtLset } @@ -36,5 +44,23 @@ func (c TestClient) Addr() (string, bool) { return c.Name, c.IsLoc func (c TestClient) Matches(matches []*labels.Matcher) bool { return !c.StoreFilterNotMatches } -func (c TestClient) GroupKey() string { return c.GroupKeyStr } -func (c TestClient) ReplicaKey() string { return c.ReplicaKeyStr } +// ReplicaInfo returns replica topology hints for GROUP_REPLICA partial response strategy. +// It mimics the unification logic of endpointRef.ReplicaInfo(): +// - If ReplicaGroupStr is set (first-class StoreInfo field), use it as Group with QuorumValue. +// - Otherwise fall back to GroupKeyStr (DNS-based) with Quorum=0 (must-success). +func (c TestClient) ReplicaInfo() storepb.ReplicaInfo { + // Prefer first-class ReplicaGroupStr if set + if c.ReplicaGroupStr != "" { + return storepb.ReplicaInfo{ + Group: c.ReplicaGroupStr, + Replica: c.ReplicaKeyStr, + Quorum: c.QuorumValue, + } + } + // Fall back to DNS-based grouping + return storepb.ReplicaInfo{ + Group: c.GroupKeyStr, + Replica: c.ReplicaKeyStr, + Quorum: 0, // Must-success for DNS-based stores + } +} diff --git a/pkg/store/tsdb.go b/pkg/store/tsdb.go index 5d16ba9bd8f..b4a5f1ef9fb 100644 --- a/pkg/store/tsdb.go +++ b/pkg/store/tsdb.go @@ -72,7 +72,8 @@ type TSDBStore struct { matcherCache storecache.MatchersCache extLset labels.Labels - infoOnlyLset labels.Labels // Labels only exposed via InfoAPI, not merged into series + replicaGroup string // Replica group identifier for GROUP_REPLICA strategy + quorum int32 // Minimum healthy stores required per group startStoreFilterUpdate bool storeFilter filter.StoreFilter mtx sync.RWMutex @@ -170,20 +171,27 @@ func (s *TSDBStore) SetExtLset(extLset labels.Labels) { s.extLset = extLset } -// SetInfoOnlyLset sets labels that are only exposed via InfoAPI (LabelSet), -// but NOT merged into series responses. Useful for metadata like replica group/quorum. -func (s *TSDBStore) SetInfoOnlyLset(infoOnlyLset labels.Labels) { +// SetReplicaInfo sets the replica group and quorum for GROUP_REPLICA partial response strategy. +func (s *TSDBStore) SetReplicaInfo(replicaGroup string, quorum int32) { s.mtx.Lock() defer s.mtx.Unlock() - s.infoOnlyLset = infoOnlyLset + s.replicaGroup = replicaGroup + s.quorum = quorum } -func (s *TSDBStore) getInfoOnlyLset() labels.Labels { +// ReplicaInfo returns replica topology hints for GROUP_REPLICA partial response strategy. +// Since TSDBStore is used internally by receivers, it returns the first-class fields directly. +// Quorum=0 means must-success semantics. +func (s *TSDBStore) ReplicaInfo() ReplicaInfo { s.mtx.RLock() defer s.mtx.RUnlock() - return s.infoOnlyLset + return ReplicaInfo{ + Group: s.replicaGroup, + Replica: "", // TSDBStore doesn't have a replica identifier + Quorum: int(s.quorum), + } } func (s *TSDBStore) getExtLset() labels.Labels { @@ -194,21 +202,10 @@ func (s *TSDBStore) getExtLset() labels.Labels { } func (s *TSDBStore) LabelSet() []labelpb.ZLabelSet { - // Combine extLset and infoOnlyLset for InfoAPI exposure. - // infoOnlyLset labels are NOT merged into series responses. - combined := s.getExtLset() - if infoOnly := s.getInfoOnlyLset(); infoOnly.Len() > 0 { - builder := labels.NewBuilder(combined) - infoOnly.Range(func(l labels.Label) { - builder.Set(l.Name, l.Value) - }) - combined = builder.Labels() - } - - zlabels := labelpb.ZLabelSetsFromPromLabels(combined) + labels := labelpb.ZLabelSetsFromPromLabels(s.getExtLset()) labelSets := []labelpb.ZLabelSet{} - if len(zlabels) > 0 { - labelSets = append(labelSets, zlabels...) + if len(labels) > 0 { + labelSets = append(labelSets, labels...) } return labelSets