Ingester: Metadata APIs should honour QueryIngestersWithin when QueryStoreForLabels is enabled

Signed-off-by: 🌲 Harry 🌊 John 🏔 <[email protected]>

Author: harry671003

CHANGELOG.md

@@ -393,6 +393,7 @@ func (t *Cortex) initIngesterService() (serv services.Service, err error) {
 	t.Cfg.Ingester.DistributorShardingStrategy = t.Cfg.Distributor.ShardingStrategy
 	t.Cfg.Ingester.DistributorShardByAllLabels = t.Cfg.Distributor.ShardByAllLabels
 	t.Cfg.Ingester.InstanceLimitsFn = ingesterInstanceLimits(t.RuntimeConfig)
+	t.Cfg.Ingester.QueryStoreForLabels = t.Cfg.Querier.QueryStoreForLabels
 	t.tsdbIngesterConfig()
 
 	t.Ingester, err = ingester.New(t.Cfg.Ingester, t.Overrides, prometheus.DefaultRegisterer, util_log.Logger)

pkg/cortex/modules.go

@@ -106,6 +106,9 @@ type Config struct {
 	DistributorShardingStrategy string `yaml:"-"`
 	DistributorShardByAllLabels bool   `yaml:"-"`
 
+	// Injected at runtime and read from querier config.
+	QueryStoreForLabels bool `yaml:"-"`
+
 	DefaultLimits    InstanceLimits         `yaml:"instance_limits"`
 	InstanceLimitsFn func() *InstanceLimits `yaml:"-"`
 
@@ -1303,7 +1306,7 @@ func (i *Ingester) LabelValues(ctx context.Context, req *client.LabelValuesReque
 		return &client.LabelValuesResponse{}, nil
 	}
 
-	mint, maxt, err := metadataQueryRange(startTimestampMs, endTimestampMs, db)
+	mint, maxt, err := metadataQueryRange(startTimestampMs, endTimestampMs, db, i.cfg.QueryStoreForLabels)
 	if err != nil {
 		return nil, err
 	}
@@ -1364,7 +1367,7 @@ func (i *Ingester) LabelNames(ctx context.Context, req *client.LabelNamesRequest
 		return &client.LabelNamesResponse{}, nil
 	}
 
-	mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db)
+	mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db, i.cfg.QueryStoreForLabels)
 	if err != nil {
 		return nil, err
 	}
@@ -1432,7 +1435,7 @@ func (i *Ingester) MetricsForLabelMatchers(ctx context.Context, req *client.Metr
 		return nil, err
 	}
 
-	mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db)
+	mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db, i.cfg.QueryStoreForLabels)
 	if err != nil {
 		return nil, err
 	}
@@ -2564,7 +2567,12 @@ func (i *Ingester) flushHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 // metadataQueryRange returns the best range to query for metadata queries based on the timerange in the ingester.
-func metadataQueryRange(queryStart, queryEnd int64, db *userTSDB) (mint, maxt int64, err error) {
+func metadataQueryRange(queryStart, queryEnd int64, db *userTSDB, queryStoreForLabels bool) (mint, maxt int64, err error) {
+	if queryStoreForLabels {
+		// If the feature for querying metadata from store-gateway is enabled, then we don't want to manipulate the mint and maxt
+		return
+	}
+
 	// Ingesters are run with limited retention and we don't support querying the store-gateway for labels yet.
 	// This means if someone loads a dashboard that is outside the range of the ingester, and we only return the
 	// data for the timerange requested (which will be empty), the dashboards will break. To fix this we should

pkg/ingester/ingester.go

@@ -1733,10 +1733,11 @@ func Test_Ingester_MetricsForLabelMatchers(t *testing.T) {
 	}
 
 	tests := map[string]struct {
-		from     int64
-		to       int64
-		matchers []*client.LabelMatchers
-		expected []*cortexpb.Metric
+		from                int64
+		to                  int64
+		matchers            []*client.LabelMatchers
+		expected            []*cortexpb.Metric
+		queryStoreForLabels bool
 	}{
 		"should return an empty response if no metric match": {
 			from: math.MinInt64,
@@ -1794,6 +1795,17 @@ func Test_Ingester_MetricsForLabelMatchers(t *testing.T) {
 				{Labels: cortexpb.FromLabelsToLabelAdapters(fixtures[1].lbls)},
 			},
 		},
+		"should filter metrics by time range if queryStoreForLabels is enabled": {
+			from: 100,
+			to:   1000,
+			matchers: []*client.LabelMatchers{{
+				Matchers: []*client.LabelMatcher{
+					{Type: client.EQUAL, Name: model.MetricNameLabel, Value: "test_1"},
+				},
+			}},
+			expected:            []*cortexpb.Metric{},
+			queryStoreForLabels: true,
+		},
 		"should not return duplicated metrics on overlapping matchers": {
 			from: math.MinInt64,
 			to:   math.MaxInt64,
@@ -1860,7 +1872,7 @@ func Test_Ingester_MetricsForLabelMatchers(t *testing.T) {
 				EndTimestampMs:   testData.to,
 				MatchersSet:      testData.matchers,
 			}
-
+			i.cfg.QueryStoreForLabels = testData.queryStoreForLabels
 			res, err := i.MetricsForLabelMatchers(ctx, req)
 			require.NoError(t, err)
 			assert.ElementsMatch(t, testData.expected, res.Metric)

pkg/ingester/ingester_test.go

@@ -37,13 +37,14 @@ type Distributor interface {
 	MetricsMetadata(ctx context.Context) ([]scrape.MetricMetadata, error)
 }
 
-func newDistributorQueryable(distributor Distributor, streaming bool, streamingMetdata bool, iteratorFn chunkIteratorFunc, queryIngestersWithin time.Duration) QueryableWithFilter {
+func newDistributorQueryable(distributor Distributor, streaming bool, streamingMetdata bool, iteratorFn chunkIteratorFunc, queryIngestersWithin time.Duration, queryStoreForLabels bool) QueryableWithFilter {
 	return distributorQueryable{
 		distributor:          distributor,
 		streaming:            streaming,
 		streamingMetdata:     streamingMetdata,
 		iteratorFn:           iteratorFn,
 		queryIngestersWithin: queryIngestersWithin,
+		queryStoreForLabels:  queryStoreForLabels,
 	}
 }
 
@@ -53,6 +54,7 @@ type distributorQueryable struct {
 	streamingMetdata     bool
 	iteratorFn           chunkIteratorFunc
 	queryIngestersWithin time.Duration
+	queryStoreForLabels  bool
 }
 
 func (d distributorQueryable) Querier(ctx context.Context, mint, maxt int64) (storage.Querier, error) {
@@ -65,6 +67,7 @@ func (d distributorQueryable) Querier(ctx context.Context, mint, maxt int64) (st
 		streamingMetadata:    d.streamingMetdata,
 		chunkIterFn:          d.iteratorFn,
 		queryIngestersWithin: d.queryIngestersWithin,
+		queryStoreForLabels:  d.queryStoreForLabels,
 	}, nil
 }
 
@@ -81,6 +84,7 @@ type distributorQuerier struct {
 	streamingMetadata    bool
 	chunkIterFn          chunkIteratorFunc
 	queryIngestersWithin time.Duration
+	queryStoreForLabels  bool
 }
 
 // Select implements storage.Querier interface.
@@ -94,36 +98,20 @@ func (q *distributorQuerier) Select(sortSeries bool, sp *storage.SelectHints, ma
 		minT, maxT = sp.Start, sp.End
 	}
 
-	// If the querier receives a 'series' query, it means only metadata is needed.
-	// For this specific case we shouldn't apply the queryIngestersWithin
-	// time range manipulation, otherwise we'll end up returning no series at all for
-	// older time ranges (while in Cortex we do ignore the start/end and always return
-	// series in ingesters).
-	// Also, in the recent versions of Prometheus, we pass in the hint but with Func set to "series".
-	// See: https://github.com/prometheus/prometheus/pull/8050
-	if sp != nil && sp.Func == "series" {
-		var (
-			ms  []metric.Metric
-			err error
-		)
-
-		if q.streamingMetadata {
-			ms, err = q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(minT), model.Time(maxT), matchers...)
-		} else {
-			ms, err = q.distributor.MetricsForLabelMatchers(ctx, model.Time(minT), model.Time(maxT), matchers...)
-		}
-
-		if err != nil {
-			return storage.ErrSeriesSet(err)
-		}
-		return series.MetricsToSeriesSet(sortSeries, ms)
-	}
-
 	// If queryIngestersWithin is enabled, we do manipulate the query mint to query samples up until
 	// now - queryIngestersWithin, because older time ranges are covered by the storage. This
 	// optimization is particularly important for the blocks storage where the blocks retention in the
 	// ingesters could be way higher than queryIngestersWithin.
 	if q.queryIngestersWithin > 0 {
+		if sp != nil && sp.Func == "series" && !q.queryStoreForLabels {
+			// If the querier receives a 'series' query, it means only metadata is needed.
+			// For this specific case where we don't query the store for labels
+			// we shouldn't apply the queryIngestersWithin time range manipulation.
+			// Otherwise we'll end up returning no series at all for
+			// older time ranges (while in Cortex we do ignore the start/end and always return
+			// series in ingesters).
+			goto SKIP_MINT_MANIPULATION
+		}
 		now := time.Now()
 		origMinT := minT
 		minT = math.Max64(minT, util.TimeToMillis(now.Add(-q.queryIngestersWithin)))
@@ -138,6 +126,27 @@ func (q *distributorQuerier) Select(sortSeries bool, sp *storage.SelectHints, ma
 		}
 	}
 
+SKIP_MINT_MANIPULATION:
+	// In the recent versions of Prometheus, we pass in the hint but with Func set to "series".
+	// See: https://github.com/prometheus/prometheus/pull/8050
+	if sp != nil && sp.Func == "series" {
+		var (
+			ms  []metric.Metric
+			err error
+		)
+
+		if q.streamingMetadata {
+			ms, err = q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(minT), model.Time(maxT), matchers...)
+		} else {
+			ms, err = q.distributor.MetricsForLabelMatchers(ctx, model.Time(minT), model.Time(maxT), matchers...)
+		}
+
+		if err != nil {
+			return storage.ErrSeriesSet(err)
+		}
+		return series.MetricsToSeriesSet(sortSeries, ms)
+	}
+
 	if q.streaming {
 		return q.streamingSelect(ctx, sortSeries, minT, maxT, matchers)
 	}

pkg/querier/distributor_queryable.go

@@ -46,7 +46,7 @@ func TestDistributorQuerier(t *testing.T) {
 		},
 		nil)
 
-	queryable := newDistributorQueryable(d, false, false, nil, 0)
+	queryable := newDistributorQueryable(d, false, false, nil, 0, false)
 	querier, err := queryable.Querier(context.Background(), mint, maxt)
 	require.NoError(t, err)
 
@@ -70,6 +70,7 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 
 	tests := map[string]struct {
 		querySeries          bool
+		queryStoreForLabels  bool
 		queryIngestersWithin time.Duration
 		queryMinT            int64
 		queryMaxT            int64
@@ -112,6 +113,15 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 			expectedMinT:         util.TimeToMillis(now.Add(-100 * time.Minute)),
 			expectedMaxT:         util.TimeToMillis(now.Add(-90 * time.Minute)),
 		},
+		"should manipulate query time range if queryIngestersWithin is enabled and queryStoreForLabels is enabled": {
+			querySeries:          true,
+			queryStoreForLabels:  true,
+			queryIngestersWithin: time.Hour,
+			queryMinT:            util.TimeToMillis(now.Add(-100 * time.Minute)),
+			queryMaxT:            util.TimeToMillis(now.Add(-30 * time.Minute)),
+			expectedMinT:         util.TimeToMillis(now.Add(-60 * time.Minute)),
+			expectedMaxT:         util.TimeToMillis(now.Add(-30 * time.Minute)),
+		},
 	}
 
 	for _, streamingEnabled := range []bool{false, true} {
@@ -124,7 +134,7 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 				distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]metric.Metric{}, nil)
 
 				ctx := user.InjectOrgID(context.Background(), "test")
-				queryable := newDistributorQueryable(distributor, streamingEnabled, streamingEnabled, nil, testData.queryIngestersWithin)
+				queryable := newDistributorQueryable(distributor, streamingEnabled, streamingEnabled, nil, testData.queryIngestersWithin, testData.queryStoreForLabels)
 				querier, err := queryable.Querier(ctx, testData.queryMinT, testData.queryMaxT)
 				require.NoError(t, err)
 
@@ -161,7 +171,7 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 
 func TestDistributorQueryableFilter(t *testing.T) {
 	d := &MockDistributor{}
-	dq := newDistributorQueryable(d, false, false, nil, 1*time.Hour)
+	dq := newDistributorQueryable(d, false, false, nil, 1*time.Hour, true)
 
 	now := time.Now()
 
@@ -209,7 +219,7 @@ func TestIngesterStreaming(t *testing.T) {
 		nil)
 
 	ctx := user.InjectOrgID(context.Background(), "0")
-	queryable := newDistributorQueryable(d, true, true, mergeChunks, 0)
+	queryable := newDistributorQueryable(d, true, true, mergeChunks, 0, true)
 	querier, err := queryable.Querier(ctx, mint, maxt)
 	require.NoError(t, err)
 
@@ -285,7 +295,7 @@ func TestIngesterStreamingMixedResults(t *testing.T) {
 		nil)
 
 	ctx := user.InjectOrgID(context.Background(), "0")
-	queryable := newDistributorQueryable(d, true, true, mergeChunks, 0)
+	queryable := newDistributorQueryable(d, true, true, mergeChunks, 0, true)
 	querier, err := queryable.Querier(ctx, mint, maxt)
 	require.NoError(t, err)
 
@@ -336,7 +346,7 @@ func TestDistributorQuerier_LabelNames(t *testing.T) {
 			d.On("MetricsForLabelMatchersStream", mock.Anything, model.Time(mint), model.Time(maxt), someMatchers).
 				Return(metrics, nil)
 
-			queryable := newDistributorQueryable(d, false, streamingEnabled, nil, 0)
+			queryable := newDistributorQueryable(d, false, streamingEnabled, nil, 0, true)
 			querier, err := queryable.Querier(context.Background(), mint, maxt)
 			require.NoError(t, err)
 

pkg/querier/distributor_queryable_test.go

@@ -146,7 +146,7 @@ func getChunksIteratorFunction(cfg Config) chunkIteratorFunc {
 func New(cfg Config, limits *validation.Overrides, distributor Distributor, stores []QueryableWithFilter, tombstonesLoader purger.TombstonesLoader, reg prometheus.Registerer, logger log.Logger) (storage.SampleAndChunkQueryable, storage.ExemplarQueryable, v1.QueryEngine) {
 	iteratorFunc := getChunksIteratorFunction(cfg)
 
-	distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterStreaming, cfg.IngesterMetadataStreaming, iteratorFunc, cfg.QueryIngestersWithin)
+	distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterStreaming, cfg.IngesterMetadataStreaming, iteratorFunc, cfg.QueryIngestersWithin, cfg.QueryStoreForLabels)
 
 	ns := make([]QueryableWithFilter, len(stores))
 	for ix, s := range stores {

pkg/querier/querier.go

@@ -213,8 +213,8 @@ func TestShouldSortSeriesIfQueryingMultipleQueryables(t *testing.T) {
 
 	distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&unorderedResponse, nil)
 	distributor.On("Query", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(unorderedResponseMatrix, nil)
-	distributorQueryableStreaming := newDistributorQueryable(distributor, true, cfg.IngesterMetadataStreaming, batch.NewChunkMergeIterator, cfg.QueryIngestersWithin)
-	distributorQueryable := newDistributorQueryable(distributor, false, cfg.IngesterMetadataStreaming, batch.NewChunkMergeIterator, cfg.QueryIngestersWithin)
+	distributorQueryableStreaming := newDistributorQueryable(distributor, true, cfg.IngesterMetadataStreaming, batch.NewChunkMergeIterator, cfg.QueryIngestersWithin, cfg.QueryStoreForLabels)
+	distributorQueryable := newDistributorQueryable(distributor, false, cfg.IngesterMetadataStreaming, batch.NewChunkMergeIterator, cfg.QueryIngestersWithin, cfg.QueryStoreForLabels)
 
 	tCases := []struct {
 		name                 string