Querier: ignore HA replica label matchers in remote read

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873
 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889
+* [ENHANCEMENT] Querier: Ignore matchers on the HA replica label in remote read requests for tenants with HA deduplication enabled, so Prometheus can remote read with the replica label among its external labels. #3714
 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869
 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880

docs/guides/ha-pair-handling.md

@@ -104,10 +104,13 @@ remote_write:
 ```
 
 When Prometheus is executing remote read queries, it will add the external
-labels to the query. In this case, if it asks for the `__replica__` label,
-Cortex will not return any data.
+labels to the query. The replica label is removed from the samples when HA
+deduplication is enabled, so Cortex ignores matchers on the HA replica label
+(`-distributor.ha-tracker.replica`) in remote read requests for tenants that
+have HA deduplication enabled. For other tenants, if the query asks for the
+`__replica__` label, Cortex will not return any data.
 
-Therefore, the `__replica__` label should only be added for remote write.
+Therefore, the `__replica__` label should preferably only be added for remote write.
 
 ## Accept multiple HA pairs in single request
 Let's assume there are two teams (T1 and T2), and each team operates two Prometheus for the HA (T1.a, T1.b for T1 and

pkg/api/handlers.go

@@ -168,6 +168,7 @@ func NewQuerierHandler(
 	exemplarQueryable storage.ExemplarQueryable,
 	engine engine.QueryEngine,
 	metadataQuerier querier.MetadataQuerier,
+	limits querier.RemoteReadLimits,
 	reg prometheus.Registerer,
 	logger log.Logger,
 ) http.Handler {
@@ -333,7 +334,7 @@ func NewQuerierHandler(
 	// TODO(gotjosh): This custom handler is temporary until we're able to vendor the changes in:
 	// https://github.com/prometheus/prometheus/pull/7125/files
 	router.Path(path.Join(prefix, "/api/v1/metadata")).Handler(querier.MetadataHandler(metadataQuerier))
-	router.Path(path.Join(prefix, "/api/v1/read")).Handler(querier.RemoteReadHandler(queryable, logger))
+	router.Path(path.Join(prefix, "/api/v1/read")).Handler(querier.RemoteReadHandler(queryable, limits, logger))
 	router.Path(path.Join(prefix, "/api/v1/read")).Methods("POST").Handler(promRouter)
 	router.Path(path.Join(prefix, "/api/v1/query")).Methods("GET", "POST").Handler(instantQueryHandler)
 	router.Path(path.Join(prefix, "/api/v1/query_range")).Methods("GET", "POST").Handler(rangedQueryHandler)
@@ -348,7 +349,7 @@ func NewQuerierHandler(
 	// TODO(gotjosh): This custom handler is temporary until we're able to vendor the changes in:
 	// https://github.com/prometheus/prometheus/pull/7125/files
 	router.Path(path.Join(legacyPrefix, "/api/v1/metadata")).Handler(querier.MetadataHandler(metadataQuerier))
-	router.Path(path.Join(legacyPrefix, "/api/v1/read")).Handler(querier.RemoteReadHandler(queryable, logger))
+	router.Path(path.Join(legacyPrefix, "/api/v1/read")).Handler(querier.RemoteReadHandler(queryable, limits, logger))
 	router.Path(path.Join(legacyPrefix, "/api/v1/read")).Methods("POST").Handler(legacyPromRouter)
 	router.Path(path.Join(legacyPrefix, "/api/v1/query")).Methods("GET", "POST").Handler(instantQueryHandler)
 	router.Path(path.Join(legacyPrefix, "/api/v1/query_range")).Methods("GET", "POST").Handler(rangedQueryHandler)

pkg/api/handlers_test.go

@@ -235,7 +235,7 @@ func TestBuildInfoAPI(t *testing.T) {
 			version.Version = tc.version
 			version.Branch = tc.branch
 			version.Revision = tc.revision
-			handler := NewQuerierHandler(cfg, querierConfig, nil, nil, nil, nil, nil, &FakeLogger{})
+			handler := NewQuerierHandler(cfg, querierConfig, nil, nil, nil, nil, nil, nil, &FakeLogger{})
 			writer := httptest.NewRecorder()
 			req := httptest.NewRequest("GET", "/api/v1/status/buildinfo", nil)
 			req = req.WithContext(user.InjectOrgID(req.Context(), "test"))

pkg/cortex/modules.go

@@ -418,6 +418,7 @@ func (t *Cortex) initQuerier() (serv services.Service, err error) {
 		t.ExemplarQueryable,
 		t.QuerierEngine,
 		t.MetadataQuerier,
+		t.OverridesConfig,
 		prometheus.DefaultRegisterer,
 		util_log.Logger,
 	)

pkg/querier/remote_read.go

@@ -1,23 +1,81 @@
 package querier
 
 import (
+	"context"
 	"net/http"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/storage"
 	storecache "github.com/thanos-io/thanos/pkg/store/cache"
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/util"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
+	"github.com/cortexproject/cortex/pkg/util/users"
 )
 
 // Queries are a set of matchers with time ranges - should not get into megabytes
 const maxRemoteReadQuerySize = 1024 * 1024
 
-// RemoteReadHandler handles Prometheus remote read requests.
-func RemoteReadHandler(q storage.Queryable, logger log.Logger) http.Handler {
+// RemoteReadLimits are the per-tenant limits used by the remote read handler.
+type RemoteReadLimits interface {
+	// AcceptHASamples returns whether HA samples deduplication is enabled for the tenant.
+	AcceptHASamples(userID string) bool
+	// HAReplicaLabel returns the label identifying a Prometheus HA replica for the tenant.
+	HAReplicaLabel(userID string) string
+}
+
+// haReplicaLabels returns the HA replica label names configured for the tenants of the request.
+// The replica label is removed by the distributor when HA deduplication is enabled, so it is
+// never present on the stored series.
+func haReplicaLabels(ctx context.Context, limits RemoteReadLimits) map[string]struct{} {
+	if limits == nil {
+		return nil
+	}
+	tenantIDs, err := users.TenantIDs(ctx)
+	if err != nil {
+		return nil
+	}
+	var names map[string]struct{}
+	for _, id := range tenantIDs {
+		if !limits.AcceptHASamples(id) {
+			continue
+		}
+		if name := limits.HAReplicaLabel(id); name != "" {
+			if names == nil {
+				names = map[string]struct{}{}
+			}
+			names[name] = struct{}{}
+		}
+	}
+	return names
+}
+
+// removeHAReplicaMatchers drops the matchers on the given HA replica label names. Prometheus adds
+// its external labels, including the replica label, to remote read matchers. Since that label
+// is not stored when HA deduplication is enabled, keeping it would return no data.
+// If dropping them would leave no matcher at all, the original matchers are returned.
+func removeHAReplicaMatchers(matchers []*labels.Matcher, names map[string]struct{}) []*labels.Matcher {
+	if len(names) == 0 {
+		return matchers
+	}
+	out := make([]*labels.Matcher, 0, len(matchers))
+	for _, m := range matchers {
+		if _, ok := names[m.Name]; !ok {
+			out = append(out, m)
+		}
+	}
+	if len(out) == 0 {
+		return matchers
+	}
+	return out
+}
+
+// RemoteReadHandler handles Prometheus remote read requests. If limits is not nil, matchers on
+// the HA replica label are ignored for tenants having HA deduplication enabled.
+func RemoteReadHandler(q storage.Queryable, limits RemoteReadLimits, logger log.Logger) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		var req client.ReadRequest
@@ -28,6 +86,8 @@ func RemoteReadHandler(q storage.Queryable, logger log.Logger) http.Handler {
 			return
 		}
 
+		replicaLabels := haReplicaLabels(ctx, limits)
+
 		// Fetch samples for all queries in parallel.
 		resp := client.ReadResponse{
 			Results: make([]*client.QueryResponse, len(req.Queries)),
@@ -41,6 +101,8 @@ func RemoteReadHandler(q storage.Queryable, logger log.Logger) http.Handler {
 					return
 				}
 
+				matchers = removeHAReplicaMatchers(matchers, replicaLabels)
+
 				querier, err := q.Querier(int64(from), int64(to))
 				if err != nil {
 					errors <- err

pkg/querier/remote_read_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/util/annotations"
 	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/common/user"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
@@ -40,7 +41,7 @@ func TestRemoteReadHandler(t *testing.T) {
 			},
 		}, nil
 	})
-	handler := RemoteReadHandler(q, log.NewNopLogger())
+	handler := RemoteReadHandler(q, nil, log.NewNopLogger())
 
 	requestBody, err := proto.Marshal(&client.ReadRequest{
 		Queries: []*client.QueryRequest{
@@ -110,3 +111,99 @@ func (m mockQuerier) LabelNames(ctx context.Context, _ *storage.LabelHints, matc
 func (mockQuerier) Close() error {
 	return nil
 }
+
+type mockRemoteReadLimits struct {
+	haEnabled map[string]bool
+	label     string
+}
+
+func (m mockRemoteReadLimits) AcceptHASamples(userID string) bool { return m.haEnabled[userID] }
+func (m mockRemoteReadLimits) HAReplicaLabel(string) string       { return m.label }
+
+type matcherCapturingQuerier struct {
+	mockQuerier
+	got *[]*labels.Matcher
+}
+
+func (m matcherCapturingQuerier) Select(ctx context.Context, sortSeries bool, sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	*m.got = matchers
+	return m.mockQuerier.Select(ctx, sortSeries, sp, matchers...)
+}
+
+func TestRemoteReadHandler_HAReplicaLabel(t *testing.T) {
+	t.Parallel()
+
+	queryMatchers := []*client.LabelMatcher{
+		{Type: client.EQUAL, Name: labels.MetricName, Value: "up"},
+		{Type: client.EQUAL, Name: "cluster", Value: "prom-team1"},
+		{Type: client.EQUAL, Name: "__replica__", Value: "replica1"},
+	}
+
+	for name, tc := range map[string]struct {
+		limits   RemoteReadLimits
+		userID   string
+		matchers []*client.LabelMatcher
+		expected []string
+	}{
+		"HA enabled removes the replica matcher": {
+			limits:   mockRemoteReadLimits{haEnabled: map[string]bool{"user-1": true}, label: "__replica__"},
+			userID:   "user-1",
+			matchers: queryMatchers,
+			expected: []string{`__name__="up"`, `cluster="prom-team1"`},
+		},
+		"HA disabled keeps the replica matcher": {
+			limits:   mockRemoteReadLimits{haEnabled: map[string]bool{}, label: "__replica__"},
+			userID:   "user-1",
+			matchers: queryMatchers,
+			expected: []string{`__name__="up"`, `cluster="prom-team1"`, `__replica__="replica1"`},
+		},
+		"a different replica label name keeps the matcher": {
+			limits:   mockRemoteReadLimits{haEnabled: map[string]bool{"user-1": true}, label: "other"},
+			userID:   "user-1",
+			matchers: queryMatchers,
+			expected: []string{`__name__="up"`, `cluster="prom-team1"`, `__replica__="replica1"`},
+		},
+		"nil limits keeps the replica matcher": {
+			limits:   nil,
+			userID:   "user-1",
+			matchers: queryMatchers,
+			expected: []string{`__name__="up"`, `cluster="prom-team1"`, `__replica__="replica1"`},
+		},
+		"the replica matcher is kept when it is the only one": {
+			limits:   mockRemoteReadLimits{haEnabled: map[string]bool{"user-1": true}, label: "__replica__"},
+			userID:   "user-1",
+			matchers: queryMatchers[2:],
+			expected: []string{`__replica__="replica1"`},
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			var got []*labels.Matcher
+			q := storage.QueryableFunc(func(mint, maxt int64) (storage.Querier, error) {
+				return matcherCapturingQuerier{mockQuerier: mockQuerier{}, got: &got}, nil
+			})
+			handler := RemoteReadHandler(q, tc.limits, log.NewNopLogger())
+
+			requestBody, err := proto.Marshal(&client.ReadRequest{
+				Queries: []*client.QueryRequest{
+					{StartTimestampMs: 0, EndTimestampMs: 10, Matchers: tc.matchers},
+				},
+			})
+			require.NoError(t, err)
+			request, err := http.NewRequest("POST", "/query", bytes.NewReader(snappy.Encode(nil, requestBody)))
+			require.NoError(t, err)
+			request = request.WithContext(user.InjectOrgID(request.Context(), tc.userID))
+
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, request)
+			require.Equal(t, 200, recorder.Result().StatusCode)
+
+			actual := make([]string, 0, len(got))
+			for _, m := range got {
+				actual = append(actual, m.String())
+			}
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}