From 9c8eb5b32ef0504c3a6f4175ffaa7bfc8a2796de Mon Sep 17 00:00:00 2001 From: Charlie Le Date: Wed, 19 Aug 2026 15:06:44 -0700 Subject: [PATCH] Remove the deprecated fifocache.size and querier.ingester-metadata-streaming settings Unlike the flag-only removals, both of these have a YAML config option and one of them changes behaviour, so they are split out here. -.fifocache.size was deprecated in 1.1.0, five years and twenty-one minors ago. A cache configured only through it now starts with no capacity, because NewFifoCache no longer copies it into MaxSizeItems. -querier.ingester-metadata-streaming defaulted to true and its help text has promised since 1.18.0 that the feature would be always on. The querier now always uses the streaming metadata RPCs, so the non-streaming branches in distributor_queryable.go and the three non-streaming methods on the local Distributor interface are gone. The tests that parameterised over streaming on/off collapse accordingly. Also removes the hidden ingester_streaming YAML field, which had no reader at all - it was left behind when -querier.ingester-streaming was deprecated in 1.17.0. Because Cortex decodes config with UnmarshalStrict, that field was still silently accepted until now. Generated config docs and the JSON schema are regenerated. Signed-off-by: Charlie Le --- CHANGELOG.md | 2 + docs/blocks-storage/querier.md | 5 - docs/configuration/config-file-reference.md | 10 - pkg/chunk/cache/fifo_cache.go | 11 - pkg/querier/distributor_queryable.go | 65 +--- pkg/querier/distributor_queryable_test.go | 178 +++++------ pkg/querier/querier.go | 5 +- pkg/querier/querier_test.go | 317 ++++++++++---------- schemas/cortex-config-schema.json | 12 - 9 files changed, 250 insertions(+), 355 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c2517bef4f..109a5b28c8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,8 @@ ## master / unreleased * [ENHANCEMENT] Query Frontend: Log `X-Grafana-User` header in query stats, slow query, and query request logs when Grafana's `send_user_header` is enabled. #7799 * [FEATURE] Engine: Add `-querier.selector-batch-size` and `-ruler.selector-batch-size` flags to configure series batching in the Thanos promQL engine. 0 disables batching. #7763 +* [CHANGE] Remove the deprecated `-.fifocache.size` flag and its `size` YAML field (deprecated in 1.1.0). Use `-.fifocache.max-size-items` or `-.fifocache.max-size-bytes`; a cache configured only via `size` now starts with no capacity. #7791 +* [CHANGE] Querier: Remove the deprecated `-querier.ingester-metadata-streaming` flag and its `ingester_metadata_streaming` YAML field (deprecated in 1.18.0, default `true`). Streaming RPCs are now always used for the metadata APIs. Also removes the dead hidden `ingester_streaming` YAML field left over from `-querier.ingester-streaming`. #7791 * [CHANGE] Remove deprecated CLI flags that have been no-ops for at least two minor releases. All of them were flag-only (no YAML config option) and already had no effect, so the only impact is that passing them now fails at startup. Remove them from your command lines before upgrading. #7790 - `-querier.ingester-streaming` (deprecated in 1.17.0) - `-querier.iterators` (deprecated in 1.17.0) diff --git a/docs/blocks-storage/querier.md b/docs/blocks-storage/querier.md index 3dc3c704b21..eb348487509 100644 --- a/docs/blocks-storage/querier.md +++ b/docs/blocks-storage/querier.md @@ -104,11 +104,6 @@ querier: # CLI flag: -querier.timeout [timeout: | default = 2m] - # Deprecated (This feature will be always on after v1.18): Use streaming RPCs - # for metadata APIs from ingester. - # CLI flag: -querier.ingester-metadata-streaming - [ingester_metadata_streaming: | default = true] - # Use LabelNames ingester RPCs with match params. # CLI flag: -querier.ingester-label-names-with-matchers [ingester_label_names_with_matchers: | default = false] diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index c713af4f124..9b3bc8e0fdf 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3844,11 +3844,6 @@ The `fifo_cache_config` configures the local in-memory cache. # The expiry duration for the cache. # CLI flag: -frontend.fifocache.duration [validity: | default = 0s] - -# Deprecated (use max-size-items or max-size-bytes instead): The number of -# entries to cache. -# CLI flag: -frontend.fifocache.size -[size: | default = 0] ``` ### `flusher_config` @@ -5276,11 +5271,6 @@ The `querier_config` configures the Cortex querier. # CLI flag: -querier.timeout [timeout: | default = 2m] -# Deprecated (This feature will be always on after v1.18): Use streaming RPCs -# for metadata APIs from ingester. -# CLI flag: -querier.ingester-metadata-streaming -[ingester_metadata_streaming: | default = true] - # Use LabelNames ingester RPCs with match params. # CLI flag: -querier.ingester-label-names-with-matchers [ingester_label_names_with_matchers: | default = false] diff --git a/pkg/chunk/cache/fifo_cache.go b/pkg/chunk/cache/fifo_cache.go index 3be8764d04d..60439c54b0a 100644 --- a/pkg/chunk/cache/fifo_cache.go +++ b/pkg/chunk/cache/fifo_cache.go @@ -14,8 +14,6 @@ import ( "github.com/pkg/errors" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" - - "github.com/cortexproject/cortex/pkg/util/flagext" ) const ( @@ -33,8 +31,6 @@ type FifoCacheConfig struct { MaxSizeBytes string `yaml:"max_size_bytes"` MaxSizeItems int `yaml:"max_size_items"` Validity time.Duration `yaml:"validity"` - - DeprecatedSize int `yaml:"size"` } // RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet @@ -42,8 +38,6 @@ func (cfg *FifoCacheConfig) RegisterFlagsWithPrefix(prefix, description string, f.StringVar(&cfg.MaxSizeBytes, prefix+"fifocache.max-size-bytes", "", description+"Maximum memory size of the cache in bytes. A unit suffix (KB, MB, GB) may be applied.") f.IntVar(&cfg.MaxSizeItems, prefix+"fifocache.max-size-items", 0, description+"Maximum number of entries in the cache.") f.DurationVar(&cfg.Validity, prefix+"fifocache.duration", 0, description+"The expiry duration for the cache.") - - f.IntVar(&cfg.DeprecatedSize, prefix+"fifocache.size", 0, "Deprecated (use max-size-items or max-size-bytes instead): "+description+"The number of entries to cache. ") } func (cfg *FifoCacheConfig) Validate() error { @@ -93,11 +87,6 @@ type cacheEntry struct { // NewFifoCache returns a new initialised FifoCache of size. func NewFifoCache(name string, cfg FifoCacheConfig, reg prometheus.Registerer, logger log.Logger) *FifoCache { - if cfg.DeprecatedSize > 0 { - flagext.DeprecatedFlagsUsed.Inc() - level.Warn(logger).Log("msg", "running with DEPRECATED flag fifocache.size, use fifocache.max-size-items or fifocache.max-size-bytes instead", "cache", name) - cfg.MaxSizeItems = cfg.DeprecatedSize - } maxSizeBytes, _ := parsebytes(cfg.MaxSizeBytes) if maxSizeBytes == 0 && cfg.MaxSizeItems == 0 { diff --git a/pkg/querier/distributor_queryable.go b/pkg/querier/distributor_queryable.go index 42f5fb59d4a..d648f39ff4a 100644 --- a/pkg/querier/distributor_queryable.go +++ b/pkg/querier/distributor_queryable.go @@ -34,22 +34,18 @@ const retryMaxBackoff = 5 * time.Millisecond type Distributor interface { QueryStream(ctx context.Context, from, to model.Time, partialDataEnabled bool, matchers ...*labels.Matcher) (*client.QueryStreamResponse, error) QueryExemplars(ctx context.Context, from, to model.Time, matchers ...[]*labels.Matcher) (*client.ExemplarQueryResponse, error) - LabelValuesForLabelName(ctx context.Context, from, to model.Time, label model.LabelName, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) LabelValuesForLabelNameStream(ctx context.Context, from, to model.Time, label model.LabelName, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) - LabelNames(context.Context, model.Time, model.Time, *storage.LabelHints, bool, ...*labels.Matcher) ([]string, error) LabelNamesStream(context.Context, model.Time, model.Time, *storage.LabelHints, bool, ...*labels.Matcher) ([]string, error) - MetricsForLabelMatchers(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) MetricsForLabelMatchersStream(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) MetricsMetadata(ctx context.Context, req *client.MetricsMetadataRequest) ([]scrape.MetricMetadata, error) } -func newDistributorQueryable(distributor Distributor, streamingMetdata bool, labelNamesWithMatchers bool, iteratorFn chunkIteratorFunc, isPartialDataEnabled partialdata.IsCfgEnabledFunc, ingesterQueryMaxAttempts int, limits *validation.Overrides, nowFn func() time.Time) QueryableWithFilter { +func newDistributorQueryable(distributor Distributor, labelNamesWithMatchers bool, iteratorFn chunkIteratorFunc, isPartialDataEnabled partialdata.IsCfgEnabledFunc, ingesterQueryMaxAttempts int, limits *validation.Overrides, nowFn func() time.Time) QueryableWithFilter { if nowFn == nil { nowFn = time.Now } return distributorQueryable{ distributor: distributor, - streamingMetdata: streamingMetdata, labelNamesWithMatchers: labelNamesWithMatchers, iteratorFn: iteratorFn, isPartialDataEnabled: isPartialDataEnabled, @@ -61,7 +57,6 @@ func newDistributorQueryable(distributor Distributor, streamingMetdata bool, lab type distributorQueryable struct { distributor Distributor - streamingMetdata bool labelNamesWithMatchers bool iteratorFn chunkIteratorFunc isPartialDataEnabled partialdata.IsCfgEnabledFunc @@ -75,7 +70,6 @@ func (d distributorQueryable) Querier(mint, maxt int64) (storage.Querier, error) distributor: d.distributor, mint: mint, maxt: maxt, - streamingMetadata: d.streamingMetdata, labelNamesMatchers: d.labelNamesWithMatchers, chunkIterFn: d.iteratorFn, isPartialDataEnabled: d.isPartialDataEnabled, @@ -93,7 +87,6 @@ func (d distributorQueryable) UseQueryable(now time.Time, userID string, _, quer type distributorQuerier struct { distributor Distributor mint, maxt int64 - streamingMetadata bool labelNamesMatchers bool chunkIterFn chunkIteratorFunc isPartialDataEnabled partialdata.IsCfgEnabledFunc @@ -142,16 +135,7 @@ func (q *distributorQuerier) Select(ctx context.Context, sortSeries bool, sp *st // 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 []labels.Labels - err error - ) - - if q.streamingMetadata { - ms, err = q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(minT), model.Time(maxT), sp, partialDataEnabled, matchers...) - } else { - ms, err = q.distributor.MetricsForLabelMatchers(ctx, model.Time(minT), model.Time(maxT), sp, partialDataEnabled, matchers...) - } + ms, err := q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(minT), model.Time(maxT), sp, partialDataEnabled, matchers...) if err != nil && !partialdata.IsPartialDataError(err) { return storage.ErrSeriesSet(err) @@ -251,22 +235,11 @@ func (q *distributorQuerier) queryWithRetry(ctx context.Context, queryFunc func( } func (q *distributorQuerier) LabelValues(ctx context.Context, name string, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) { - var ( - lvs []string - err error - ) - partialDataEnabled := q.partialDataEnabled(ctx) - if q.streamingMetadata { - lvs, err = q.labelsWithRetry(ctx, func() ([]string, error) { - return q.distributor.LabelValuesForLabelNameStream(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...) - }) - } else { - lvs, err = q.labelsWithRetry(ctx, func() ([]string, error) { - return q.distributor.LabelValuesForLabelName(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...) - }) - } + lvs, err := q.labelsWithRetry(ctx, func() ([]string, error) { + return q.distributor.LabelValuesForLabelNameStream(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...) + }) if partialdata.IsPartialDataError(err) { warnings := annotations.Annotations(nil) @@ -286,20 +259,9 @@ func (q *distributorQuerier) LabelNames(ctx context.Context, hints *storage.Labe log, ctx := spanlogger.New(ctx, "distributorQuerier.LabelNames") defer log.Finish() - var ( - ln []string - err error - ) - - if q.streamingMetadata { - ln, err = q.labelsWithRetry(ctx, func() ([]string, error) { - return q.distributor.LabelNamesStream(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...) - }) - } else { - ln, err = q.labelsWithRetry(ctx, func() ([]string, error) { - return q.distributor.LabelNames(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...) - }) - } + ln, err := q.labelsWithRetry(ctx, func() ([]string, error) { + return q.distributor.LabelNamesStream(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...) + }) if partialdata.IsPartialDataError(err) { warnings := annotations.Annotations(nil) @@ -348,16 +310,7 @@ func (q *distributorQuerier) labelNamesWithMatchers(ctx context.Context, hints * log, ctx := spanlogger.New(ctx, "distributorQuerier.labelNamesWithMatchers") defer log.Finish() - var ( - ms []labels.Labels - err error - ) - - if q.streamingMetadata { - ms, err = q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(q.mint), model.Time(q.maxt), labelHintsToSelectHints(hints), partialDataEnabled, matchers...) - } else { - ms, err = q.distributor.MetricsForLabelMatchers(ctx, model.Time(q.mint), model.Time(q.maxt), labelHintsToSelectHints(hints), partialDataEnabled, matchers...) - } + ms, err := q.distributor.MetricsForLabelMatchersStream(ctx, model.Time(q.mint), model.Time(q.maxt), labelHintsToSelectHints(hints), partialDataEnabled, matchers...) if err != nil && !partialdata.IsPartialDataError(err) { return nil, nil, err diff --git a/pkg/querier/distributor_queryable_test.go b/pkg/querier/distributor_queryable_test.go index f4dfa0d4dde..d2398f12943 100644 --- a/pkg/querier/distributor_queryable_test.go +++ b/pkg/querier/distributor_queryable_test.go @@ -80,50 +80,47 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T) }, } - for _, streamingMetadataEnabled := range []bool{false, true} { - for testName, testData := range tests { - t.Run(fmt.Sprintf("%s (streaming metadata enabled: %t)", testName, streamingMetadataEnabled), func(t *testing.T) { - t.Parallel() + for testName, testData := range tests { + t.Run(testName, func(t *testing.T) { + t.Parallel() - distributor := &MockDistributor{} - distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&client.QueryStreamResponse{}, nil) - distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) - distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) + distributor := &MockDistributor{} + distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&client.QueryStreamResponse{}, nil) + distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) - ctx := user.InjectOrgID(context.Background(), "test") + ctx := user.InjectOrgID(context.Background(), "test") - limits := DefaultLimitsConfig() - limits.QueryIngestersWithin = model.Duration(testData.queryIngestersWithin) - overrides := validation.NewOverrides(limits, nil) + limits := DefaultLimitsConfig() + limits.QueryIngestersWithin = model.Duration(testData.queryIngestersWithin) + overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(distributor, streamingMetadataEnabled, true, nil, nil, 1, overrides, nil) - querier, err := queryable.Querier(testData.queryMinT, testData.queryMaxT) - require.NoError(t, err) + queryable := newDistributorQueryable(distributor, true, nil, nil, 1, overrides, nil) + querier, err := queryable.Querier(testData.queryMinT, testData.queryMaxT) + require.NoError(t, err) - start, end, err := validateQueryTimeRange(ctx, "test", testData.queryMinT, testData.queryMaxT, overrides, 0) - require.NoError(t, err) - // Select hints are passed by Prometheus when querying /series. - var hints *storage.SelectHints - if testData.querySeries { - hints = &storage.SelectHints{ - Start: start, - End: end, - Func: "series", - } + start, end, err := validateQueryTimeRange(ctx, "test", testData.queryMinT, testData.queryMaxT, overrides, 0) + require.NoError(t, err) + // Select hints are passed by Prometheus when querying /series. + var hints *storage.SelectHints + if testData.querySeries { + hints = &storage.SelectHints{ + Start: start, + End: end, + Func: "series", } + } - seriesSet := querier.Select(ctx, true, hints) - require.NoError(t, seriesSet.Err()) + seriesSet := querier.Select(ctx, true, hints) + require.NoError(t, seriesSet.Err()) - if testData.expectedMinT == 0 && testData.expectedMaxT == 0 { - assert.Len(t, distributor.Calls, 0) - } else { - require.Len(t, distributor.Calls, 1) - assert.InDelta(t, testData.expectedMinT, int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), float64(15*time.Second.Milliseconds())) - assert.Equal(t, testData.expectedMaxT, int64(distributor.Calls[0].Arguments.Get(2).(model.Time))) - } - }) - } + if testData.expectedMinT == 0 && testData.expectedMaxT == 0 { + assert.Len(t, distributor.Calls, 0) + } else { + require.Len(t, distributor.Calls, 1) + assert.InDelta(t, testData.expectedMinT, int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), float64(15*time.Second.Milliseconds())) + assert.Equal(t, testData.expectedMaxT, int64(distributor.Calls[0].Arguments.Get(2).(model.Time))) + } + }) } } @@ -136,7 +133,7 @@ func TestDistributorQueryableFilter(t *testing.T) { limits.QueryIngestersWithin = model.Duration(1 * time.Hour) overrides := validation.NewOverrides(limits, nil) - dq := newDistributorQueryable(d, false, true, nil, nil, 1, overrides, nil) + dq := newDistributorQueryable(d, true, nil, nil, 1, overrides, nil) now := time.Now() @@ -190,7 +187,7 @@ func TestIngesterStreaming(t *testing.T) { limits.QueryIngestersWithin = model.Duration(0) // Disable time filtering for this test overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, func(string) bool { + queryable := newDistributorQueryable(d, true, batch.NewChunkMergeIterator, func(string) bool { return partialDataEnabled }, 1, overrides, nil) querier, err := queryable.Querier(mint, maxt) @@ -345,13 +342,11 @@ func TestDistributorQuerier_Retry(t *testing.T) { res := []string{"foo"} for _, err := range tc.errors { d.On("LabelNamesStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(res, err).Once() - d.On("LabelNames", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(res, err).Once() } case "LabelValues": res := []string{"foo"} for _, err := range tc.errors { d.On("LabelValuesForLabelNameStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(res, err).Once() - d.On("LabelValuesForLabelName", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(res, err).Once() } } @@ -361,7 +356,7 @@ func TestDistributorQuerier_Retry(t *testing.T) { limits.QueryIngestersWithin = model.Duration(0) overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, func(string) bool { + queryable := newDistributorQueryable(d, true, batch.NewChunkMergeIterator, func(string) bool { return true }, ingesterQueryMaxAttempts, overrides, nil) querier, err := queryable.Querier(mint, maxt) @@ -419,7 +414,7 @@ func TestDistributorQuerier_Select_CancelledContext_NoRetry(t *testing.T) { ingesterQueryMaxAttempts := 1 limits := DefaultLimitsConfig() overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, func(string) bool { + queryable := newDistributorQueryable(d, true, batch.NewChunkMergeIterator, func(string) bool { return true }, ingesterQueryMaxAttempts, overrides, nil) querier, err := queryable.Querier(mint, maxt) @@ -453,7 +448,7 @@ func TestDistributorQuerier_Select_CancelledContext(t *testing.T) { ingesterQueryMaxAttempts := 2 limits := DefaultLimitsConfig() overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, func(string) bool { + queryable := newDistributorQueryable(d, true, batch.NewChunkMergeIterator, func(string) bool { return true }, ingesterQueryMaxAttempts, overrides, nil) querier, err := queryable.Querier(mint, maxt) @@ -478,7 +473,7 @@ func TestDistributorQuerier_Labels_CancelledContext(t *testing.T) { ingesterQueryMaxAttempts := 2 limits := DefaultLimitsConfig() overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, func(string) bool { + queryable := newDistributorQueryable(d, true, batch.NewChunkMergeIterator, func(string) bool { return true }, ingesterQueryMaxAttempts, overrides, nil) querier, err := queryable.Querier(mint, maxt) @@ -502,56 +497,49 @@ func TestDistributorQuerier_LabelNames(t *testing.T) { labelNames := []string{"foo", "job"} for _, labelNamesWithMatchers := range []bool{false, true} { - for _, streamingEnabled := range []bool{false, true} { - for _, partialDataEnabled := range []bool{false, true} { - streamingEnabled := streamingEnabled - labelNamesWithMatchers := labelNamesWithMatchers - t.Run("with matchers", func(t *testing.T) { - t.Parallel() - - metrics := []labels.Labels{ - labels.FromStrings("foo", "bar"), - labels.FromStrings("job", "baz"), - labels.FromStrings("job", "baz", "foo", "boom"), - } - d := &MockDistributor{} - - var partialDataErr error - if partialDataEnabled { - partialDataErr = partialdata.ErrPartialData - } - if labelNamesWithMatchers { - d.On("LabelNames", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). - Return(labelNames, partialDataErr) - d.On("LabelNamesStream", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). - Return(labelNames, partialDataErr) - } else { - d.On("MetricsForLabelMatchers", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). - Return(metrics, partialDataErr) - d.On("MetricsForLabelMatchersStream", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). - Return(metrics, partialDataErr) - } - - limits := DefaultLimitsConfig() - overrides := validation.NewOverrides(limits, nil) - - queryable := newDistributorQueryable(d, streamingEnabled, labelNamesWithMatchers, nil, func(string) bool { - return partialDataEnabled - }, 1, overrides, nil) - querier, err := queryable.Querier(mint, maxt) - require.NoError(t, err) - - ctx := context.Background() - names, warnings, err := querier.LabelNames(ctx, nil, someMatchers...) - require.NoError(t, err) - if partialDataEnabled { - assert.Contains(t, warnings, partialdata.ErrPartialData.Error()) - } else { - assert.Empty(t, warnings) - } - assert.Equal(t, labelNames, names) - }) - } + for _, partialDataEnabled := range []bool{false, true} { + labelNamesWithMatchers := labelNamesWithMatchers + t.Run("with matchers", func(t *testing.T) { + t.Parallel() + + metrics := []labels.Labels{ + labels.FromStrings("foo", "bar"), + labels.FromStrings("job", "baz"), + labels.FromStrings("job", "baz", "foo", "boom"), + } + d := &MockDistributor{} + + var partialDataErr error + if partialDataEnabled { + partialDataErr = partialdata.ErrPartialData + } + if labelNamesWithMatchers { + d.On("LabelNamesStream", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). + Return(labelNames, partialDataErr) + } else { + d.On("MetricsForLabelMatchersStream", mock.Anything, model.Time(mint), model.Time(maxt), mock.Anything, someMatchers). + Return(metrics, partialDataErr) + } + + limits := DefaultLimitsConfig() + overrides := validation.NewOverrides(limits, nil) + + queryable := newDistributorQueryable(d, labelNamesWithMatchers, nil, func(string) bool { + return partialDataEnabled + }, 1, overrides, nil) + querier, err := queryable.Querier(mint, maxt) + require.NoError(t, err) + + ctx := context.Background() + names, warnings, err := querier.LabelNames(ctx, nil, someMatchers...) + require.NoError(t, err) + if partialDataEnabled { + assert.Contains(t, warnings, partialdata.ErrPartialData.Error()) + } else { + assert.Empty(t, warnings) + } + assert.Equal(t, labelNames, names) + }) } } } @@ -625,7 +613,7 @@ func TestDistributorQuerier_QueryIngestersWithinBoundary(t *testing.T) { limits.QueryIngestersWithin = model.Duration(lookback) overrides := validation.NewOverrides(limits, nil) - queryable := newDistributorQueryable(distributor, false, true, nil, nil, 1, overrides, func() time.Time { return now }) + queryable := newDistributorQueryable(distributor, true, nil, nil, 1, overrides, func() time.Time { return now }) querier, err := queryable.Querier(testData.queryMinT, testData.queryMaxT) require.NoError(t, err) diff --git a/pkg/querier/querier.go b/pkg/querier/querier.go index bef94e47b17..70576d0b849 100644 --- a/pkg/querier/querier.go +++ b/pkg/querier/querier.go @@ -45,8 +45,6 @@ import ( type Config struct { MaxConcurrent int `yaml:"max_concurrent"` Timeout time.Duration `yaml:"timeout"` - IngesterStreaming bool `yaml:"ingester_streaming" doc:"hidden"` - IngesterMetadataStreaming bool `yaml:"ingester_metadata_streaming"` IngesterLabelNamesWithMatchers bool `yaml:"ingester_label_names_with_matchers"` MaxSamples int `yaml:"max_samples"` EnablePerStepStats bool `yaml:"per_step_stats_enabled"` @@ -133,7 +131,6 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) { cfg.StoreGatewayClient.RegisterFlagsWithPrefix("querier.store-gateway-client", f) f.IntVar(&cfg.MaxConcurrent, "querier.max-concurrent", 20, "The maximum number of concurrent queries.") f.DurationVar(&cfg.Timeout, "querier.timeout", 2*time.Minute, "The timeout for a query.") - f.BoolVar(&cfg.IngesterMetadataStreaming, "querier.ingester-metadata-streaming", true, "Deprecated (This feature will be always on after v1.18): Use streaming RPCs for metadata APIs from ingester.") f.BoolVar(&cfg.IngesterLabelNamesWithMatchers, "querier.ingester-label-names-with-matchers", false, "Use LabelNames ingester RPCs with match params.") f.IntVar(&cfg.MaxSamples, "querier.max-samples", 50e6, "Maximum number of samples a single query can load into memory.") f.BoolVar(&cfg.EnablePerStepStats, "querier.per-step-stats-enabled", false, "Enable returning samples stats per steps in query response.") @@ -266,7 +263,7 @@ func New(cfg Config, limits *validation.Overrides, distributor Distributor, stor ) } - distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, iteratorFunc, isPartialDataEnabled, cfg.IngesterQueryMaxAttempts, limits, nil) + distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, iteratorFunc, isPartialDataEnabled, cfg.IngesterQueryMaxAttempts, limits, nil) ns := make([]QueryableWithFilter, len(stores)) for ix, s := range stores { diff --git a/pkg/querier/querier_test.go b/pkg/querier/querier_test.go index 9c1e63a876d..53f11c8fa08 100644 --- a/pkg/querier/querier_test.go +++ b/pkg/querier/querier_test.go @@ -304,7 +304,7 @@ func TestShouldSortSeriesIfQueryingMultipleQueryables(t *testing.T) { limits := DefaultLimitsConfig() testOverrides := validation.NewOverrides(limits, nil) - distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) + distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) tCases := []struct { name string @@ -453,7 +453,7 @@ func TestLimits(t *testing.T) { limits := DefaultLimitsConfig() testOverrides := validation.NewOverrides(limits, nil) - distributorQueryableStreaming := newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) + distributorQueryableStreaming := newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) tCases := []struct { name string @@ -1089,184 +1089,177 @@ func TestQuerier_ValidateQueryTimeRange_MaxQueryLookback(t *testing.T) { Timeout: 1 * time.Minute, } queryEngine := promql.NewEngine(opts) - for _, ingesterStreaming := range []bool{true, false} { - expectedMethodForLabelMatchers := "MetricsForLabelMatchers" - expectedMethodForLabelNames := "LabelNames" - expectedMethodForLabelValues := "LabelValuesForLabelName" - if ingesterStreaming { - expectedMethodForLabelMatchers = "MetricsForLabelMatchersStream" - expectedMethodForLabelNames = "LabelNamesStream" - expectedMethodForLabelValues = "LabelValuesForLabelNameStream" - } - for testName, testData := range tests { - t.Run(testName, func(t *testing.T) { - ctx := user.InjectOrgID(context.Background(), "test") - - var cfg Config - flagext.DefaultValues(&cfg) - cfg.IngesterMetadataStreaming = ingesterStreaming - // Disable active query tracker to avoid mmap error. - cfg.ActiveQueryTrackerDir = "" - - limits := DefaultLimitsConfig() - limits.MaxQueryLookback = testData.maxQueryLookback - overrides := validation.NewOverrides(limits, nil) - - chunkStore := &emptyChunkStore{} - queryables := []QueryableWithFilter{UseAlwaysQueryable(NewMockStoreQueryable(chunkStore))} - - t.Run("query range", func(t *testing.T) { - if testData.query == "" { - return - } - distributor := &MockDistributor{} - distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&client.QueryStreamResponse{}, nil) + expectedMethodForLabelMatchers := "MetricsForLabelMatchersStream" + expectedMethodForLabelNames := "LabelNamesStream" + expectedMethodForLabelValues := "LabelValuesForLabelNameStream" - queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) + for testName, testData := range tests { + t.Run(testName, func(t *testing.T) { + ctx := user.InjectOrgID(context.Background(), "test") - query, err := queryEngine.NewRangeQuery(ctx, queryable, nil, testData.query, testData.queryStartTime, testData.queryEndTime, time.Minute) - require.NoError(t, err) + var cfg Config + flagext.DefaultValues(&cfg) + // Disable active query tracker to avoid mmap error. + cfg.ActiveQueryTrackerDir = "" - r := query.Exec(ctx) - require.Nil(t, r.Err) + limits := DefaultLimitsConfig() + limits.MaxQueryLookback = testData.maxQueryLookback + overrides := validation.NewOverrides(limits, nil) - _, err = r.Matrix() - require.Nil(t, err) - - if !testData.expectedSkipped { - // Assert on the time range of the actual executed query (5s delta). - delta := float64(5000) - require.Len(t, distributor.Calls, 1) - assert.InDelta(t, util.TimeToMillis(testData.expectedQueryStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) - assert.InDelta(t, util.TimeToMillis(testData.expectedQueryEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) - } else { - // Ensure no query has been executed (because skipped). - assert.Len(t, distributor.Calls, 0) - } - }) + chunkStore := &emptyChunkStore{} + queryables := []QueryableWithFilter{UseAlwaysQueryable(NewMockStoreQueryable(chunkStore))} - t.Run("series", func(t *testing.T) { - distributor := &MockDistributor{} - distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) - distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) + t.Run("query range", func(t *testing.T) { + if testData.query == "" { + return + } + distributor := &MockDistributor{} + distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&client.QueryStreamResponse{}, nil) - queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) - q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) - require.NoError(t, err) + queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) - // We apply the validation here again since when initializing querier we change the start/end time, - // but when querying series we don't validate again. So we should pass correct hints here. - start, end, err := validateQueryTimeRange(ctx, "test", util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime), overrides, 0) - // Skipped query will hit errEmptyTimeRange during validation. - if !testData.expectedSkipped { - require.NoError(t, err) - } + query, err := queryEngine.NewRangeQuery(ctx, queryable, nil, testData.query, testData.queryStartTime, testData.queryEndTime, time.Minute) + require.NoError(t, err) - hints := &storage.SelectHints{ - Start: start, - End: end, - Func: "series", - } - matcher := labels.MustNewMatcher(labels.MatchEqual, labels.MetricName, "test") - - set := q.Select(ctx, false, hints, matcher) - require.False(t, set.Next()) // Expected to be empty. - require.NoError(t, set.Err()) - - if !testData.expectedSkipped { - // Assert on the time range of the actual executed query (5s delta). - delta := float64(5000) - require.Len(t, distributor.Calls, 1) - assert.Equal(t, expectedMethodForLabelMatchers, distributor.Calls[0].Method) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) - } else { - // Ensure no query has been executed (because skipped). - assert.Len(t, distributor.Calls, 0) - } - }) + r := query.Exec(ctx) + require.Nil(t, r.Err) + + _, err = r.Matrix() + require.Nil(t, err) + + if !testData.expectedSkipped { + // Assert on the time range of the actual executed query (5s delta). + delta := float64(5000) + require.Len(t, distributor.Calls, 1) + assert.InDelta(t, util.TimeToMillis(testData.expectedQueryStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) + assert.InDelta(t, util.TimeToMillis(testData.expectedQueryEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) + } else { + // Ensure no query has been executed (because skipped). + assert.Len(t, distributor.Calls, 0) + } + }) - t.Run("label names", func(t *testing.T) { - distributor := &MockDistributor{} - distributor.On("LabelNames", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) - distributor.On("LabelNamesStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) + t.Run("series", func(t *testing.T) { + distributor := &MockDistributor{} + distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) + distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]labels.Labels{}, nil) - queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) - q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) - require.NoError(t, err) + queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) + q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) + require.NoError(t, err) - _, _, err = q.LabelNames(ctx, nil) + // We apply the validation here again since when initializing querier we change the start/end time, + // but when querying series we don't validate again. So we should pass correct hints here. + start, end, err := validateQueryTimeRange(ctx, "test", util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime), overrides, 0) + // Skipped query will hit errEmptyTimeRange during validation. + if !testData.expectedSkipped { require.NoError(t, err) + } - if !testData.expectedSkipped { - // Assert on the time range of the actual executed query (5s delta). - delta := float64(5000) - require.Len(t, distributor.Calls, 1) - assert.Equal(t, expectedMethodForLabelNames, distributor.Calls[0].Method) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) - } else { - // Ensure no query has been executed (because skipped). - assert.Len(t, distributor.Calls, 0) - } - }) + hints := &storage.SelectHints{ + Start: start, + End: end, + Func: "series", + } + matcher := labels.MustNewMatcher(labels.MatchEqual, labels.MetricName, "test") + + set := q.Select(ctx, false, hints, matcher) + require.False(t, set.Next()) // Expected to be empty. + require.NoError(t, set.Err()) + + if !testData.expectedSkipped { + // Assert on the time range of the actual executed query (5s delta). + delta := float64(5000) + require.Len(t, distributor.Calls, 1) + assert.Equal(t, expectedMethodForLabelMatchers, distributor.Calls[0].Method) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) + } else { + // Ensure no query has been executed (because skipped). + assert.Len(t, distributor.Calls, 0) + } + }) - t.Run("label names with matchers", func(t *testing.T) { - matchers := []*labels.Matcher{ - labels.MustNewMatcher(labels.MatchNotEqual, "route", "get_user"), - } - distributor := &MockDistributor{} - distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything, matchers).Return([]labels.Labels{}, nil) - distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, matchers).Return([]labels.Labels{}, nil) + t.Run("label names", func(t *testing.T) { + distributor := &MockDistributor{} + distributor.On("LabelNames", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) + distributor.On("LabelNamesStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) - queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) - q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) - require.NoError(t, err) + queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) + q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) + require.NoError(t, err) - _, _, err = q.LabelNames(ctx, nil, matchers...) - require.NoError(t, err) + _, _, err = q.LabelNames(ctx, nil) + require.NoError(t, err) - if !testData.expectedSkipped { - // Assert on the time range of the actual executed query (5s delta). - delta := float64(5000) - require.Len(t, distributor.Calls, 1) - assert.Equal(t, expectedMethodForLabelMatchers, distributor.Calls[0].Method) - args := distributor.Calls[0].Arguments - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(args.Get(1).(model.Time)), delta) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(args.Get(2).(model.Time)), delta) - assert.Equal(t, matchers, args.Get(4).([]*labels.Matcher)) - } else { - // Ensure no query has been executed (because skipped). - assert.Len(t, distributor.Calls, 0) - } - }) + if !testData.expectedSkipped { + // Assert on the time range of the actual executed query (5s delta). + delta := float64(5000) + require.Len(t, distributor.Calls, 1) + assert.Equal(t, expectedMethodForLabelNames, distributor.Calls[0].Method) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) + } else { + // Ensure no query has been executed (because skipped). + assert.Len(t, distributor.Calls, 0) + } + }) - t.Run("label values", func(t *testing.T) { - distributor := &MockDistributor{} - distributor.On("LabelValuesForLabelName", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) - distributor.On("LabelValuesForLabelNameStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) + t.Run("label names with matchers", func(t *testing.T) { + matchers := []*labels.Matcher{ + labels.MustNewMatcher(labels.MatchNotEqual, "route", "get_user"), + } + distributor := &MockDistributor{} + distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything, matchers).Return([]labels.Labels{}, nil) + distributor.On("MetricsForLabelMatchersStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, matchers).Return([]labels.Labels{}, nil) - queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) - q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) - require.NoError(t, err) + queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) + q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) + require.NoError(t, err) - _, _, err = q.LabelValues(ctx, labels.MetricName, nil) - require.NoError(t, err) + _, _, err = q.LabelNames(ctx, nil, matchers...) + require.NoError(t, err) - if !testData.expectedSkipped { - // Assert on the time range of the actual executed query (5s delta). - delta := float64(5000) - require.Len(t, distributor.Calls, 1) - assert.Equal(t, expectedMethodForLabelValues, distributor.Calls[0].Method) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) - assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) - } else { - // Ensure no query has been executed(because skipped). - assert.Len(t, distributor.Calls, 0) - } - }) + if !testData.expectedSkipped { + // Assert on the time range of the actual executed query (5s delta). + delta := float64(5000) + require.Len(t, distributor.Calls, 1) + assert.Equal(t, expectedMethodForLabelMatchers, distributor.Calls[0].Method) + args := distributor.Calls[0].Arguments + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(args.Get(1).(model.Time)), delta) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(args.Get(2).(model.Time)), delta) + assert.Equal(t, matchers, args.Get(4).([]*labels.Matcher)) + } else { + // Ensure no query has been executed (because skipped). + assert.Len(t, distributor.Calls, 0) + } }) - } + + t.Run("label values", func(t *testing.T) { + distributor := &MockDistributor{} + distributor.On("LabelValuesForLabelName", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) + distributor.On("LabelValuesForLabelNameStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) + + queryable, _, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger(), nil, nil) + q, err := queryable.Querier(util.TimeToMillis(testData.queryStartTime), util.TimeToMillis(testData.queryEndTime)) + require.NoError(t, err) + + _, _, err = q.LabelValues(ctx, labels.MetricName, nil) + require.NoError(t, err) + + if !testData.expectedSkipped { + // Assert on the time range of the actual executed query (5s delta). + delta := float64(5000) + require.Len(t, distributor.Calls, 1) + assert.Equal(t, expectedMethodForLabelValues, distributor.Calls[0].Method) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataStartTime), int64(distributor.Calls[0].Arguments.Get(1).(model.Time)), delta) + assert.InDelta(t, util.TimeToMillis(testData.expectedMetadataEndTime), int64(distributor.Calls[0].Arguments.Get(2).(model.Time)), delta) + } else { + // Ensure no query has been executed(because skipped). + assert.Len(t, distributor.Calls, 0) + } + }) + }) } } @@ -1892,11 +1885,11 @@ func TestQuerier_ProjectionHints(t *testing.T) { var distributorQueryable QueryableWithFilter if testData.queryIngesters { // Ingesters will be queried - distributorQueryable = newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) + distributorQueryable = newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil) } else { // Ingesters will not be queried (time range is too old) distributorQueryable = UseBeforeTimestampQueryable( - newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil), + newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, testOverrides, nil), start.Add(-1*time.Hour), ) } @@ -1961,7 +1954,7 @@ func TestQuerier_ResourceBasedLimiter(t *testing.T) { require.NoError(t, err) chunkStore := &errDistributor{} - distributorQueryable := newDistributorQueryable(chunkStore, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, overrides, nil) + distributorQueryable := newDistributorQueryable(chunkStore, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, overrides, nil) reg := prometheus.NewPedanticRegistry() queryable := NewQueryable(distributorQueryable, nil, cfg, overrides, resourceBasedLimiter, log.NewNopLogger(), reg) @@ -2007,7 +2000,7 @@ func TestQuerier_ResourceBasedLimiter_Nil(t *testing.T) { distributor.On("LabelValuesForLabelNameStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) distributor.On("LabelNamesStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]string{}, nil) - distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterMetadataStreaming, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, overrides, nil) + distributorQueryable := newDistributorQueryable(distributor, cfg.IngesterLabelNamesWithMatchers, batch.NewChunkMergeIterator, nil, 1, overrides, nil) // nil resourceBasedLimiter should not block queries. queryable := NewQueryable(distributorQueryable, nil, cfg, overrides, nil, log.NewNopLogger(), nil) diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index 462435e200e..82c4f45be61 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -4710,12 +4710,6 @@ "type": "number", "x-cli-flag": "frontend.fifocache.max-size-items" }, - "size": { - "default": 0, - "description": "Deprecated (use max-size-items or max-size-bytes instead): The number of entries to cache. ", - "type": "number", - "x-cli-flag": "frontend.fifocache.size" - }, "validity": { "default": "0s", "description": "The expiry duration for the cache.", @@ -6590,12 +6584,6 @@ "type": "boolean", "x-cli-flag": "querier.ingester-label-names-with-matchers" }, - "ingester_metadata_streaming": { - "default": true, - "description": "Deprecated (This feature will be always on after v1.18): Use streaming RPCs for metadata APIs from ingester.", - "type": "boolean", - "x-cli-flag": "querier.ingester-metadata-streaming" - }, "ingester_query_max_attempts": { "default": 1, "description": "The maximum number of times we attempt fetching data from ingesters for retryable errors (ex. partial data returned).",