diff --git a/CHANGELOG.md b/CHANGELOG.md index b53819eb496..a42d9764d59 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/guides/ha-pair-handling.md b/docs/guides/ha-pair-handling.md index fd1a389b4a6..ec7fb9ad108 100644 --- a/docs/guides/ha-pair-handling.md +++ b/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 diff --git a/pkg/api/handlers.go b/pkg/api/handlers.go index b3a700d84cb..6e814d26b72 100644 --- a/pkg/api/handlers.go +++ b/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) diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index cf3b7ee1a75..7ef8fac26b6 100644 --- a/pkg/api/handlers_test.go +++ b/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")) diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go index d787a6b2429..657c4eec189 100644 --- a/pkg/cortex/modules.go +++ b/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, ) diff --git a/pkg/querier/remote_read.go b/pkg/querier/remote_read.go index 5067f1a8f51..b7929812446 100644 --- a/pkg/querier/remote_read.go +++ b/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 diff --git a/pkg/querier/remote_read_test.go b/pkg/querier/remote_read_test.go index b3e7c0c28b4..eee8f2c5531 100644 --- a/pkg/querier/remote_read_test.go +++ b/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) + }) + } +}