diff --git a/CHANGELOG.md b/CHANGELOG.md index 16148eeae63..d79855483eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ ## 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] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override will not crash the process: it makes every runtime config reload fail, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792 +* [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..05ac0dbf064 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` @@ -4830,12 +4825,6 @@ query_rejection: # them. [query_attributes: | default = []] -# Deprecated(use ruler.query-offset instead) and will be removed in v1.19.0: -# Duration to delay the evaluation of rules to ensure the underlying metrics -# have been pushed to Cortex. -# CLI flag: -ruler.evaluation-delay-duration -[ruler_evaluation_delay_duration: | default = 0s] - # The default tenant's shard size when the shuffle-sharding strategy is used by # ruler. When this setting is specified in the per-tenant overrides, a value of # 0 disables shuffle sharding for the tenant. If the value is < 1 the shard size @@ -5274,11 +5263,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/integration/ruler_test.go b/integration/ruler_test.go index be36e5bc7c9..04fcc963b0d 100644 --- a/integration/ruler_test.go +++ b/integration/ruler_test.go @@ -1126,8 +1126,6 @@ func TestRulerMetricsForInvalidQueries(t *testing.T) { // Evaluate rules often, so that we don't need to wait for metrics to show up. "-ruler.evaluation-interval": "2s", "-ruler.poll-interval": "2s", - // No delay - "-ruler.evaluation-delay-duration": "0", "-blocks-storage.tsdb.block-ranges-period": "1h", "-blocks-storage.bucket-store.sync-interval": "1s", @@ -1266,8 +1264,6 @@ func TestRulerMetricsWhenIngesterFails(t *testing.T) { // Evaluate rules often, so that we don't need to wait for metrics to show up. "-ruler.evaluation-interval": "2s", "-ruler.poll-interval": "2s", - // No delay - "-ruler.evaluation-delay-duration": "0", // We run single ingester only, no replication. "-distributor.replication-factor": "1", @@ -1370,8 +1366,6 @@ func TestRulerDisablesRuleGroups(t *testing.T) { // Evaluate rules often, so that we don't need to wait for metrics to show up. "-ruler.evaluation-interval": "2s", "-ruler.poll-interval": "2s", - // No delay - "-ruler.evaluation-delay-duration": "0", // We run single ingester only, no replication. "-distributor.replication-factor": "1", @@ -1628,8 +1622,6 @@ func TestRulerKeepFiring(t *testing.T) { // Evaluate rules often, so that we don't need to wait for metrics to show up. "-ruler.evaluation-interval": "2s", "-ruler.poll-interval": "2s", - // No delay - "-ruler.evaluation-delay-duration": "0", "-blocks-storage.tsdb.block-ranges-period": "1h", "-blocks-storage.bucket-store.sync-interval": "1s", @@ -1901,12 +1893,11 @@ func TestRulerXFunctionsWithThanosEngine(t *testing.T) { BlocksStorageFlags(), RulerFlags(), map[string]string{ - "-querier.thanos-engine": "true", - "-querier.enable-x-functions": "true", - "-ruler.evaluation-interval": "2s", - "-ruler.poll-interval": "2s", - "-ruler.evaluation-delay-duration": "0", - "-distributor.replication-factor": "1", + "-querier.thanos-engine": "true", + "-querier.enable-x-functions": "true", + "-ruler.evaluation-interval": "2s", + "-ruler.poll-interval": "2s", + "-distributor.replication-factor": "1", }, ) 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/exporter_test.go b/pkg/util/validation/exporter_test.go index 6067ed96067..bc27c97b2b9 100644 --- a/pkg/util/validation/exporter_test.go +++ b/pkg/util/validation/exporter_test.go @@ -112,7 +112,6 @@ func TestOverridesExporter_withConfig(t *testing.T) { cortex_overrides{limit_name="reject_old_samples",user="tenant-a"} 0 cortex_overrides{limit_name="reject_old_samples_max_age",user="tenant-a"} 1.2096e+06 cortex_overrides{limit_name="results_cache_ttl",user="tenant-a"} 0 - cortex_overrides{limit_name="ruler_evaluation_delay_duration",user="tenant-a"} 0 cortex_overrides{limit_name="ruler_max_rule_groups_per_tenant",user="tenant-a"} 0 cortex_overrides{limit_name="ruler_max_rules_per_rule_group",user="tenant-a"} 0 cortex_overrides{limit_name="ruler_query_offset",user="tenant-a"} 0 diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 019a5adc3ed..e665655950a 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -220,7 +220,6 @@ type Limits struct { QueryRejection QueryRejection `yaml:"query_rejection" json:"query_rejection" doc:"nocli|description=Configuration for query rejection."` // Ruler defaults and limits. - RulerEvaluationDelay model.Duration `yaml:"ruler_evaluation_delay_duration" json:"ruler_evaluation_delay_duration"` RulerTenantShardSize float64 `yaml:"ruler_tenant_shard_size" json:"ruler_tenant_shard_size"` RulerMaxRulesPerRuleGroup int `yaml:"ruler_max_rules_per_rule_group" json:"ruler_max_rules_per_rule_group"` RulerMaxRuleGroupsPerTenant int `yaml:"ruler_max_rule_groups_per_tenant" json:"ruler_max_rule_groups_per_tenant"` @@ -270,8 +269,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") @@ -354,7 +351,6 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.IntVar(&l.MaxOutstandingPerTenant, "frontend.max-outstanding-requests-per-tenant", 100, "Maximum number of outstanding requests per tenant per request queue (either query frontend or query scheduler); requests beyond this error with HTTP 429.") - f.Var(&l.RulerEvaluationDelay, "ruler.evaluation-delay-duration", "Deprecated(use ruler.query-offset instead) and will be removed in v1.19.0: Duration to delay the evaluation of rules to ensure the underlying metrics have been pushed to Cortex.") f.Float64Var(&l.RulerTenantShardSize, "ruler.tenant-shard-size", 0, "The default tenant's shard size when the shuffle-sharding strategy is used by ruler. When this setting is specified in the per-tenant overrides, a value of 0 disables shuffle sharding for the tenant. If the value is < 1 the shard size will be a percentage of the total rulers.") f.IntVar(&l.RulerMaxRulesPerRuleGroup, "ruler.max-rules-per-rule-group", 0, "Maximum number of rules per rule group per-tenant. 0 to disable.") f.IntVar(&l.RulerMaxRuleGroupsPerTenant, "ruler.max-rule-groups-per-tenant", 0, "Maximum number of rule groups per-tenant. 0 to disable.") @@ -1076,13 +1072,7 @@ func (o *Overrides) RulerMaxRuleGroupsPerTenant(userID string) int { // RulerQueryOffset returns the rule query offset for a given user. func (o *Overrides) RulerQueryOffset(userID string) time.Duration { - ruleOffset := time.Duration(o.GetOverridesForUser(userID).RulerQueryOffset) - evaluationDelay := time.Duration(o.GetOverridesForUser(userID).RulerEvaluationDelay) - if evaluationDelay > ruleOffset { - level.Warn(util_log.Logger).Log("msg", "ruler.query-offset was overridden by highest value in [Deprecated]ruler.evaluation-delay-duration", "ruler.query-offset", ruleOffset, "ruler.evaluation-delay-duration", evaluationDelay) - return evaluationDelay - } - return ruleOffset + return time.Duration(o.GetOverridesForUser(userID).RulerQueryOffset) } // RulesPartialData returns whether rule may be evaluated with data from a single zone, if other zones are not available. diff --git a/pkg/util/validation/limits_test.go b/pkg/util/validation/limits_test.go index 6c5813e80ba..31c832dcc73 100644 --- a/pkg/util/validation/limits_test.go +++ b/pkg/util/validation/limits_test.go @@ -943,23 +943,6 @@ func TestCompileQueryPriorityRegex(t *testing.T) { require.Nil(t, l.QueryPriority.Priorities[0].QueryAttributes[0].CompiledRegex) } -func TestEvaluationDelayHigherThanRulerQueryOffset(t *testing.T) { - tenant := "tenant" - evaluationDelay := time.Duration(10) - tenantLimits := map[string]*Limits{ - tenant: { - RulerQueryOffset: 5, - RulerEvaluationDelay: model.Duration(evaluationDelay), - }, - } - - defaults := Limits{} - ov := NewOverrides(defaults, newMockTenantLimits(tenantLimits)) - - rulerQueryOffset := ov.RulerQueryOffset(tenant) - assert.Equal(t, evaluationDelay, rulerQueryOffset) -} - func TestLimitsPerLabelSetsForSeries(t *testing.T) { for _, tc := range []struct { name string diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index aed2998e063..37563e827e1 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.", @@ -6129,13 +6123,6 @@ "description": "Go text/template for alert generator URLs. Available variables: .ExternalURL (resolved external URL) and .Expression (PromQL expression). Built-in functions like urlquery are available. A jsonEscape function is also provided for embedding expressions inside JSON-encoded URL parameters. If empty, uses default Prometheus /graph format.", "type": "string" }, - "ruler_evaluation_delay_duration": { - "default": "0s", - "description": "Deprecated(use ruler.query-offset instead) and will be removed in v1.19.0: Duration to delay the evaluation of rules to ensure the underlying metrics have been pushed to Cortex.", - "type": "string", - "x-cli-flag": "ruler.evaluation-delay-duration", - "x-format": "duration" - }, "ruler_external_labels": { "additionalProperties": true, "default": [], @@ -6590,12 +6577,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).",