diff --git a/CHANGELOG.md b/CHANGELOG.md index 16148eeae63..676c5d1d5a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## master / unreleased * [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) + - `-querier.batch-iterators` (deprecated in 1.17.0) + - `-querier.query-store-for-labels-enabled` (deprecated in 1.18.0) + - `-querier.max-outstanding-requests-per-tenant` (deprecated in 1.18.0; use `-frontend.max-outstanding-requests-per-tenant`) + - `-query-scheduler.max-outstanding-requests-per-tenant` (deprecated in 1.18.0; use `-frontend.max-outstanding-requests-per-tenant`) + - `-blocks-storage.tsdb.wal-compression-enabled` (deprecated in 1.19.0; use `-blocks-storage.tsdb.wal-compression-type`) + - `-ingester.max-series-per-query` (a chunks-storage limit, ignored since blocks storage; use `-querier.max-fetched-series-per-query`) * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446 * [CHANGE] HA Tracker: Move `-distributor.ha-tracker.failover-timeout` from a global config to a per-tenant runtime config. The flag name and default value (30s) remain the same. #7481 diff --git a/docs/blocks-storage/migrate-from-chunks-to-blocks.md b/docs/blocks-storage/migrate-from-chunks-to-blocks.md index 848f22d2ba8..b08e09def7d 100644 --- a/docs/blocks-storage/migrate-from-chunks-to-blocks.md +++ b/docs/blocks-storage/migrate-from-chunks-to-blocks.md @@ -112,7 +112,6 @@ Querier (and ruler) can be reconfigured to use `blocks` as "primary" store to se - `-store.engine=blocks` - `-querier.second-store-engine=chunks` - `-querier.use-second-store-before-time=` -- `-querier.ingester-streaming=true` #### `-querier.use-second-store-before-time` @@ -135,7 +134,6 @@ During the rollback, queriers and rulers need to use the same configuration chan - `-store.engine=chunks` - `-querier.second-store-engine=blocks` - `-querier.use-second-store-before-time` should not be set -- `-querier.ingester-streaming=false` Once the rollback is complete, some configuration changes need to stay in place, because some data has already been stored to blocks: @@ -241,16 +239,12 @@ Most important thing is generating resources with blocks configuration, and expo // Querier and ruler configuration used during migration, and after. query_config_during_migration:: { - // Disable streaming, as it is broken when querying both chunks and blocks ingesters at the same time. - 'querier.ingester-streaming': 'false', - // query-store-after is required during migration, since new ingesters running on blocks will not load any chunks from chunks-WAL. // All such chunks are however flushed to the store. 'querier.query-store-after': '0', }, query_config_after_migration:: { - 'querier.ingester-streaming': 'true', 'querier.query-ingesters-within': '13h', // TSDB ingesters have data for up to 4d. 'querier.query-store-after': '12h', // Can be enabled once blocks ingesters are running for 12h. diff --git a/docs/blocks-storage/querier.md b/docs/blocks-storage/querier.md index 91c39e2d741..2466ecf7847 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 b7922a94376..27c24b55c7a 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` @@ -5274,11 +5269,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/frontend/v1/frontend.go b/pkg/frontend/v1/frontend.go index f58cde326af..6b185034c83 100644 --- a/pkg/frontend/v1/frontend.go +++ b/pkg/frontend/v1/frontend.go @@ -19,9 +19,7 @@ import ( "github.com/cortexproject/cortex/pkg/frontend/v1/frontendv1pb" "github.com/cortexproject/cortex/pkg/querier/stats" "github.com/cortexproject/cortex/pkg/scheduler/queue" - "github.com/cortexproject/cortex/pkg/util/flagext" "github.com/cortexproject/cortex/pkg/util/httpgrpcutil" - util_log "github.com/cortexproject/cortex/pkg/util/log" "github.com/cortexproject/cortex/pkg/util/services" "github.com/cortexproject/cortex/pkg/util/users" "github.com/cortexproject/cortex/pkg/util/validation" @@ -38,9 +36,6 @@ type Config struct { // RegisterFlags adds the flags required to config this to the given FlagSet. func (cfg *Config) RegisterFlags(f *flag.FlagSet) { - //lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods - flagext.DeprecatedFlag(f, "querier.max-outstanding-requests-per-tenant", "Deprecated: Use frontend.max-outstanding-requests-per-tenant instead.", util_log.Logger) - f.DurationVar(&cfg.QuerierForgetDelay, "query-frontend.querier-forget-delay", 0, "If a querier disconnects without sending notification about graceful shutdown, the query-frontend will keep the querier in the tenant's shard until the forget delay has passed. This feature is useful to reduce the blast radius when shuffle-sharding is enabled.") } 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 020b1793b92..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"` @@ -130,19 +128,9 @@ var ( func (cfg *Config) RegisterFlags(f *flag.FlagSet) { cfg.ThanosEngine.RegisterFlagsWithPrefix("querier.", f) - //lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods - flagext.DeprecatedFlag(f, "querier.ingester-streaming", "Deprecated: Use streaming RPCs to query ingester. QueryStream is always enabled and the flag is not effective anymore.", util_log.Logger) - //lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods - flagext.DeprecatedFlag(f, "querier.iterators", "Deprecated: Use iterators to execute query. This flag is no longer functional; Batch iterator is always enabled instead.", util_log.Logger) - //lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods - flagext.DeprecatedFlag(f, "querier.batch-iterators", "Deprecated: Use batch iterators to execute query. This flag is no longer functional; Batch iterator is always enabled now.", util_log.Logger) - //lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods - flagext.DeprecatedFlag(f, "querier.query-store-for-labels-enabled", "Deprecated: Querying long-term store is always enabled.", util_log.Logger) - 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.") @@ -275,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 f062eeaac21..8990039ec8f 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/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index d84cdb5f8df..8b75c0d1494 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -30,10 +30,8 @@ import ( "github.com/cortexproject/cortex/pkg/scheduler/fragment_table" "github.com/cortexproject/cortex/pkg/scheduler/queue" "github.com/cortexproject/cortex/pkg/scheduler/schedulerpb" - "github.com/cortexproject/cortex/pkg/util/flagext" "github.com/cortexproject/cortex/pkg/util/grpcclient" "github.com/cortexproject/cortex/pkg/util/httpgrpcutil" - util_log "github.com/cortexproject/cortex/pkg/util/log" "github.com/cortexproject/cortex/pkg/util/services" "github.com/cortexproject/cortex/pkg/util/users" "github.com/cortexproject/cortex/pkg/util/validation" @@ -109,7 +107,6 @@ type Config struct { } func (cfg *Config) RegisterFlags(f *flag.FlagSet) { - flagext.DeprecatedFlag(f, "query-scheduler.max-outstanding-requests-per-tenant", "Deprecated: Use frontend.max-outstanding-requests-per-tenant instead.", util_log.Logger) f.DurationVar(&cfg.QuerierForgetDelay, "query-scheduler.querier-forget-delay", 0, "If a querier disconnects without sending notification about graceful shutdown, the query-scheduler will keep the querier in the tenant's shard until the forget delay has passed. This feature is useful to reduce the blast radius when shuffle-sharding is enabled.") cfg.GRPCClientConfig.RegisterFlagsWithPrefix("query-scheduler.grpc-client-config", "", f) } diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index f3817fe32c7..385b4d73fd4 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -16,8 +16,6 @@ import ( "github.com/thanos-io/thanos/pkg/store" "github.com/cortexproject/cortex/pkg/storage/bucket" - "github.com/cortexproject/cortex/pkg/util/flagext" - util_log "github.com/cortexproject/cortex/pkg/util/log" "github.com/cortexproject/cortex/pkg/util/parquetutil" "github.com/cortexproject/cortex/pkg/util/users" ) @@ -213,8 +211,6 @@ func (cfg *TSDBConfig) RegisterFlags(f *flag.FlagSet) { f.BoolVar(&cfg.MemorySnapshotOnShutdown, "blocks-storage.tsdb.memory-snapshot-on-shutdown", false, "True to enable snapshotting of in-memory TSDB data on disk when shutting down.") f.Int64Var(&cfg.OutOfOrderCapMax, "blocks-storage.tsdb.out-of-order-cap-max", tsdb.DefaultOutOfOrderCapMax, "[EXPERIMENTAL] Configures the maximum number of samples per chunk that can be out-of-order.") - flagext.DeprecatedFlag(f, "blocks-storage.tsdb.wal-compression-enabled", "Deprecated (use blocks-storage.tsdb.wal-compression-type instead): True to enable TSDB WAL compression.", util_log.Logger) - cfg.PostingsCache.RegisterFlagsWithPrefix("blocks-storage.", f) } diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 019a5adc3ed..c8246201131 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -270,8 +270,6 @@ type Limits struct { // RegisterFlags adds the flags required to config this to the given FlagSet func (l *Limits) RegisterFlags(f *flag.FlagSet) { - flagext.DeprecatedFlag(f, "ingester.max-series-per-query", "Deprecated: The maximum number of series for which a query can fetch samples from each ingester. This limit is enforced only in the ingesters (when querying samples not flushed to the storage yet) and it's a per-instance limit. This limit is ignored when running the Cortex blocks storage. When running Cortex with blocks storage use -querier.max-fetched-series-per-query limit instead.", util_log.Logger) - f.IntVar(&l.IngestionTenantShardSize, "distributor.ingestion-tenant-shard-size", 0, "The default tenant's shard size when the shuffle-sharding strategy is used. Must be set both on ingesters and distributors. When this setting is specified in the per-tenant overrides, a value of 0 disables shuffle sharding for the tenant.") f.Float64Var(&l.IngestionRate, "distributor.ingestion-rate-limit", 25000, "Per-user ingestion rate limit in samples per second.") f.Float64Var(&l.NativeHistogramIngestionRate, "distributor.native-histogram-ingestion-rate-limit", float64(rate.Inf), "Per-user native histogram ingestion rate limit in samples per second. Disabled by default") diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index aed2998e063..0c4edbb4953 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).",