From 6241eff84220f164575e8d975515ca138f573a28 Mon Sep 17 00:00:00 2001 From: Shvejan Mutheboyina Date: Wed, 26 Aug 2026 07:22:22 +0000 Subject: [PATCH 1/2] per tenant close_idle_tsdb_timeout Signed-off-by: Shvejan Mutheboyina --- docs/blocks-storage/querier.md | 9 -- docs/blocks-storage/store-gateway.md | 9 -- docs/configuration/config-file-reference.md | 19 ++-- pkg/cortex/cortex.go | 2 +- pkg/cortex/runtime_config.go | 3 +- pkg/ingester/ingester.go | 21 ++++- pkg/ingester/ingester_test.go | 93 ++++++++++++++++++-- pkg/storage/tsdb/config.go | 2 - pkg/util/validation/exporter_test.go | 1 + pkg/util/validation/limits.go | 19 +++- pkg/util/validation/limits_test.go | 96 ++++++++++++++++++--- schemas/cortex-config-schema.json | 16 ++-- 12 files changed, 227 insertions(+), 63 deletions(-) diff --git a/docs/blocks-storage/querier.md b/docs/blocks-storage/querier.md index 15bea9a237d..6ccc5555263 100644 --- a/docs/blocks-storage/querier.md +++ b/docs/blocks-storage/querier.md @@ -2260,15 +2260,6 @@ blocks_storage: # CLI flag: -blocks-storage.tsdb.flush-blocks-on-shutdown [flush_blocks_on_shutdown: | default = false] - # If TSDB has not received any data for this duration, and all blocks from - # TSDB have been shipped, TSDB is closed and deleted from local disk. If set - # to positive value, this value must be greater than - # -limits.query-ingesters-within flag to make sure that TSDB is not closed - # prematurely, which could cause partial query results. 0 or negative value - # disables closing of idle TSDB. - # CLI flag: -blocks-storage.tsdb.close-idle-tsdb-timeout - [close_idle_tsdb_timeout: | default = 0s] - # The size of the in-memory queue used before flushing chunks to the disk. # CLI flag: -blocks-storage.tsdb.head-chunks-write-queue-size [head_chunks_write_queue_size: | default = 0] diff --git a/docs/blocks-storage/store-gateway.md b/docs/blocks-storage/store-gateway.md index 0dc4d398023..ee95f9c7e11 100644 --- a/docs/blocks-storage/store-gateway.md +++ b/docs/blocks-storage/store-gateway.md @@ -2313,15 +2313,6 @@ blocks_storage: # CLI flag: -blocks-storage.tsdb.flush-blocks-on-shutdown [flush_blocks_on_shutdown: | default = false] - # If TSDB has not received any data for this duration, and all blocks from - # TSDB have been shipped, TSDB is closed and deleted from local disk. If set - # to positive value, this value must be greater than - # -limits.query-ingesters-within flag to make sure that TSDB is not closed - # prematurely, which could cause partial query results. 0 or negative value - # disables closing of idle TSDB. - # CLI flag: -blocks-storage.tsdb.close-idle-tsdb-timeout - [close_idle_tsdb_timeout: | default = 0s] - # The size of the in-memory queue used before flushing chunks to the disk. # CLI flag: -blocks-storage.tsdb.head-chunks-write-queue-size [head_chunks_write_queue_size: | default = 0] diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 5bdae1e6923..98d989b95cd 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -2974,15 +2974,6 @@ tsdb: # CLI flag: -blocks-storage.tsdb.flush-blocks-on-shutdown [flush_blocks_on_shutdown: | default = false] - # If TSDB has not received any data for this duration, and all blocks from - # TSDB have been shipped, TSDB is closed and deleted from local disk. If set - # to positive value, this value must be greater than - # -limits.query-ingesters-within flag to make sure that TSDB is not closed - # prematurely, which could cause partial query results. 0 or negative value - # disables closing of idle TSDB. - # CLI flag: -blocks-storage.tsdb.close-idle-tsdb-timeout - [close_idle_tsdb_timeout: | default = 0s] - # The size of the in-memory queue used before flushing chunks to the disk. # CLI flag: -blocks-storage.tsdb.head-chunks-write-queue-size [head_chunks_write_queue_size: | default = 0] @@ -4838,10 +4829,18 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # Lookback period for shuffle sharding of ingesters. This is a per-tenant limit # that can be overridden in the runtime configuration. Should be greater than or -# equal to query-ingesters-within. +# equal to query-store-after and query-ingesters-within. # CLI flag: -limits.shuffle-sharding-ingesters-lookback-period [shuffle_sharding_ingesters_lookback_period: | default = 0s] +# If TSDB has not received any data for this duration, and all blocks from TSDB +# have been shipped, TSDB is closed and deleted from local disk. This is a +# per-tenant limit that can be overridden in the runtime configuration. Should +# be greater than or equal to -limits.query-ingesters-within to prevent +# premature TSDB closure. 0 to disable. +# CLI flag: -limits.close-idle-tsdb-timeout +[close_idle_tsdb_timeout: | default = 0s] + # The maximum number of rows that can be fetched when querying parquet storage. # Each row maps to a series in a parquet file. This limit applies before # materializing chunks. 0 to disable. diff --git a/pkg/cortex/cortex.go b/pkg/cortex/cortex.go index 25c5049ba02..f3c4eaa8479 100644 --- a/pkg/cortex/cortex.go +++ b/pkg/cortex/cortex.go @@ -315,7 +315,7 @@ func (c *Config) Validate(log log.Logger) error { if err := c.LimitsConfig.Validate(c.NameValidationScheme, c.Distributor.ShardByAllLabels, c.Ingester.ActiveSeriesMetricsEnabled, c.Distributor.HATrackerConfig.UpdateTimeout, c.Distributor.HATrackerConfig.UpdateTimeoutJitterMax); err != nil { return errors.Wrap(err, "invalid limits config") } - if err := c.LimitsConfig.ValidateQueryLimits("default", c.BlocksStorage.TSDB.CloseIdleTSDBTimeout); err != nil { + if err := c.LimitsConfig.ValidateQueryLimits("default", time.Duration(c.LimitsConfig.CloseIdleTSDBTimeout)); err != nil { return errors.Wrap(err, "invalid query routing config") } if err := c.ResourceMonitor.Validate(); err != nil { diff --git a/pkg/cortex/runtime_config.go b/pkg/cortex/runtime_config.go index 8031b6be5c4..f5d849f7423 100644 --- a/pkg/cortex/runtime_config.go +++ b/pkg/cortex/runtime_config.go @@ -5,6 +5,7 @@ import ( "io" "net/http" "strings" + "time" "gopkg.in/yaml.v2" @@ -80,7 +81,7 @@ func (l runtimeConfigLoader) load(r io.Reader) (any, error) { if err := ul.Validate(l.cfg.NameValidationScheme, l.cfg.Distributor.ShardByAllLabels, l.cfg.Ingester.ActiveSeriesMetricsEnabled, l.cfg.Distributor.HATrackerConfig.UpdateTimeout, l.cfg.Distributor.HATrackerConfig.UpdateTimeoutJitterMax); err != nil { return nil, err } - if err := ul.ValidateQueryLimits(userID, l.cfg.BlocksStorage.TSDB.CloseIdleTSDBTimeout); err != nil { + if err := ul.ValidateQueryLimits(userID, time.Duration(ul.CloseIdleTSDBTimeout)); err != nil { return nil, err } } diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index d877b90a12f..93f2e707bbd 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -1023,7 +1023,9 @@ func (i *Ingester) starting(ctx context.Context) error { servs = append(servs, shippingService) } - if i.cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout > 0 { + // The close-idle TSDB timeout is now a per-tenant limit, so the timer always runs; + // closeAndDeleteIdleUserTSDBs decides per tenant whether any TSDB is actually closed. + { interval := i.cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval if interval == 0 { interval = cortex_tsdb.DefaultCloseIdleTSDBInterval @@ -3554,6 +3556,17 @@ func (i *Ingester) closeAndDeleteIdleUserTSDBs(ctx context.Context) error { return nil } + // A non-positive close-idle timeout disables idle-based closing for this tenant, + // so we skip the idle check entirely (and don't record an idleTsdbChecks result, + // since no check is performed). A tenant marked for deletion is the one exception: + // its local TSDB must still be closed and deleted regardless of the timeout. + if i.limits.CloseIdleTSDBTimeout(userID) <= 0 { + userDB, err := i.getTSDB(userID) + if err != nil || userDB == nil || !userDB.deletionMarkFound.Load() { + continue + } + } + result := i.closeAndDeleteUserTSDBIfIdle(userID) i.TSDBState.idleTsdbChecks.WithLabelValues(string(result)).Inc() @@ -3584,7 +3597,9 @@ func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckRes return tsdbShippingDisabled } - if result := userDB.shouldCloseTSDB(i.cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout); !result.shouldClose() { + closeIdleTSDBTimeout := i.limits.CloseIdleTSDBTimeout(userID) + + if result := userDB.shouldCloseTSDB(closeIdleTSDBTimeout); !result.shouldClose() { return result } @@ -3602,7 +3617,7 @@ func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckRes // Verify again, things may have changed during the checks and pushes. tenantDeleted := false - if result := userDB.shouldCloseTSDB(i.cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout); !result.shouldClose() { + if result := userDB.shouldCloseTSDB(closeIdleTSDBTimeout); !result.shouldClose() { // This will also change TSDB state back to active (via defer above). return result } else if result == tsdbTenantMarkedForDeletion { diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index d9a5299e0aa..0456719ed1a 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -5416,7 +5416,6 @@ func TestIngester_closeAndDeleteUserTSDBIfIdle_shouldNotCloseTSDBIfShippingIsInP func TestIngester_closingAndOpeningTsdbConcurrently(t *testing.T) { ctx := context.Background() cfg := defaultIngesterTestConfig(t) - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 0 // Will not run the loop, but will allow us to close any TSDB fast. // Create ingester i, err := prepareIngesterWithBlocksStorage(t, cfg, prometheus.NewRegistry()) @@ -5469,7 +5468,6 @@ func TestIngester_idleCloseEmptyTSDB(t *testing.T) { cfg := defaultIngesterTestConfig(t) cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Minute cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Minute - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 0 // Will not run the loop, but will allow us to close any TSDB fast. // Create ingester i, err := prepareIngesterWithBlocksStorage(t, cfg, prometheus.NewRegistry()) @@ -5517,7 +5515,6 @@ func TestIngester_ReadNotFailWhenTSDBIsBeingDeleted(t *testing.T) { t.Run(name, func(t *testing.T) { ctx := context.Background() cfg := defaultIngesterTestConfig(t) - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 0 // Will not run the loop, but will allow us to close any TSDB fast. cfg.BlocksStorageConfig.TSDB.KeepUserTSDBOpenOnShutdown = true // Create ingester @@ -6179,13 +6176,16 @@ func TestIngesterCompactAndCloseIdleTSDB(t *testing.T) { cfg.BlocksStorageConfig.TSDB.ShipConcurrency = 1 cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Second cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout = 1 * time.Second - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 1 * time.Second cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = 100 * time.Millisecond + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = model.Duration(1 * time.Second) + r := prometheus.NewRegistry() // Create ingester - i, err := prepareIngesterWithBlocksStorage(t, cfg, r) + i, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, nil, "", r) require.NoError(t, err) require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) @@ -6290,6 +6290,88 @@ func TestIngesterCompactAndCloseIdleTSDB(t *testing.T) { `), metricsToCheck...)) } +func TestIngesterCloseIdleTSDB_PerTenantTimeout(t *testing.T) { + cfg := defaultIngesterTestConfig(t) + cfg.LifecyclerConfig.JoinAfter = 0 + cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.ShipConcurrency = 1 + cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = 100 * time.Millisecond + + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + // Per-tenant limit enables close-idle for this tenant + limits.CloseIdleTSDBTimeout = model.Duration(1 * time.Second) + + tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &limits}) + + i, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, tenantLimits, "", prometheus.NewRegistry()) + require.NoError(t, err) + + require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) + t.Cleanup(func() { + require.NoError(t, services.StopAndAwaitTerminated(context.Background(), i)) + }) + + test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { + return i.lifecycler.GetState() + }) + + pushSingleSampleWithMetadata(t, i) + + require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) + + // Wait until TSDB has been closed via the per-tenant timeout. + test.Poll(t, 10*time.Second, 0, func() any { + i.stoppedMtx.Lock() + defer i.stoppedMtx.Unlock() + return len(i.TSDBState.dbs) + }) + + require.Greater(t, testutil.ToFloat64(i.TSDBState.idleTsdbChecks.WithLabelValues(string(tsdbIdleClosed))), float64(0)) +} + +func TestIngesterCloseIdleTSDB_DisabledWhenZero(t *testing.T) { + cfg := defaultIngesterTestConfig(t) + cfg.LifecyclerConfig.JoinAfter = 0 + cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.ShipConcurrency = 1 + cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = 100 * time.Millisecond + + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = 0 // Disabled + + tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &limits}) + + i, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, tenantLimits, "", prometheus.NewRegistry()) + require.NoError(t, err) + + require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) + t.Cleanup(func() { + require.NoError(t, services.StopAndAwaitTerminated(context.Background(), i)) + }) + + test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { + return i.lifecycler.GetState() + }) + + pushSingleSampleWithMetadata(t, i) + + require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) + + // Wait a bit and confirm TSDB is NOT closed (timeout is disabled). + time.Sleep(500 * time.Millisecond) + + i.stoppedMtx.Lock() + numDBs := len(i.TSDBState.dbs) + i.stoppedMtx.Unlock() + require.Equal(t, 1, numDBs) +} + func verifyCompactedHead(t *testing.T, i *Ingester, expected bool) { db, err := i.getTSDB(userID) require.NoError(t, err) @@ -7996,7 +8078,6 @@ func TestIngester_UpdateLabelSetMetrics(t *testing.T) { func TestIngesterPanicHandling(t *testing.T) { ctx := context.Background() cfg := defaultIngesterTestConfig(t) - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 0 // Will not run the loop, but will allow us to close any TSDB fast. cfg.BlocksStorageConfig.TSDB.KeepUserTSDBOpenOnShutdown = true // Create ingester diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index db51ebc5765..1c391de583a 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -155,7 +155,6 @@ type TSDBConfig struct { WALCompressionType string `yaml:"wal_compression_type"` WALSegmentSizeBytes int `yaml:"wal_segment_size_bytes"` FlushBlocksOnShutdown bool `yaml:"flush_blocks_on_shutdown"` - CloseIdleTSDBTimeout time.Duration `yaml:"close_idle_tsdb_timeout"` // The size of the in-memory queue used before flushing chunks to the disk. HeadChunksWriteQueueSize int `yaml:"head_chunks_write_queue_size"` @@ -205,7 +204,6 @@ func (cfg *TSDBConfig) RegisterFlags(f *flag.FlagSet) { f.StringVar(&cfg.WALCompressionType, "blocks-storage.tsdb.wal-compression-type", "", "TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable compression)") f.IntVar(&cfg.WALSegmentSizeBytes, "blocks-storage.tsdb.wal-segment-size-bytes", wlog.DefaultSegmentSize, "TSDB WAL segments files max size (bytes).") f.BoolVar(&cfg.FlushBlocksOnShutdown, "blocks-storage.tsdb.flush-blocks-on-shutdown", false, "True to flush blocks to storage on shutdown. If false, incomplete blocks will be reused after restart.") - f.DurationVar(&cfg.CloseIdleTSDBTimeout, "blocks-storage.tsdb.close-idle-tsdb-timeout", 0, "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. If set to positive value, this value must be greater than -limits.query-ingesters-within flag to make sure that TSDB is not closed prematurely, which could cause partial query results. 0 or negative value disables closing of idle TSDB.") f.IntVar(&cfg.HeadChunksWriteQueueSize, "blocks-storage.tsdb.head-chunks-write-queue-size", chunks.DefaultWriteQueueSize, "The size of the in-memory queue used before flushing chunks to the disk.") f.IntVar(&cfg.MaxExemplars, "blocks-storage.tsdb.max-exemplars", 0, "Deprecated (use the per-tenant max_exemplars limit instead) and will be removed in v1.24.0: the global fallback for the maximum number of exemplars stored in TSDB, used only when the per-tenant max_exemplars limit is 0. 0 or less means exemplars are disabled.") 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.") diff --git a/pkg/util/validation/exporter_test.go b/pkg/util/validation/exporter_test.go index 4559786a723..76c44e37997 100644 --- a/pkg/util/validation/exporter_test.go +++ b/pkg/util/validation/exporter_test.go @@ -48,6 +48,7 @@ func TestOverridesExporter_withConfig(t *testing.T) { cortex_overrides{limit_name="alertmanager_max_templates_count",user="tenant-a"} 0 cortex_overrides{limit_name="alertmanager_notification_rate_limit",user="tenant-a"} 0 cortex_overrides{limit_name="alertmanager_receivers_firewall_block_private_addresses",user="tenant-a"} 0 + cortex_overrides{limit_name="close_idle_tsdb_timeout",user="tenant-a"} 0 cortex_overrides{limit_name="compactor_blocks_retention_period",user="tenant-a"} 0 cortex_overrides{limit_name="compactor_partition_index_size_bytes",user="tenant-a"} 6.8719476736e+10 cortex_overrides{limit_name="compactor_partition_series_count",user="tenant-a"} 0 diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index cf66b88cd86..688dc1a84d5 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -206,6 +206,7 @@ type Limits struct { // "now - queryStoreAfter" so that most recent blocks are not queried. QueryStoreAfter model.Duration `yaml:"query_store_after" json:"query_store_after"` ShuffleShardingIngestersLookbackPeriod model.Duration `yaml:"shuffle_sharding_ingesters_lookback_period" json:"shuffle_sharding_ingesters_lookback_period"` + CloseIdleTSDBTimeout model.Duration `yaml:"close_idle_tsdb_timeout" json:"close_idle_tsdb_timeout"` // Parquet Queryable enforced limits. ParquetMaxFetchedRowCount int `yaml:"parquet_max_fetched_row_count" json:"parquet_max_fetched_row_count"` @@ -333,7 +334,10 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.Var(&l.QueryStoreAfter, "limits.query-store-after", "Minimum age of data before querying the long-term storage. Queries for data younger than this will only query ingesters. This is a per-tenant limit that can be overridden in the runtime configuration.") _ = l.ShuffleShardingIngestersLookbackPeriod.Set("0") - f.Var(&l.ShuffleShardingIngestersLookbackPeriod, "limits.shuffle-sharding-ingesters-lookback-period", "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-ingesters-within.") + f.Var(&l.ShuffleShardingIngestersLookbackPeriod, "limits.shuffle-sharding-ingesters-lookback-period", "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-store-after and query-ingesters-within.") + + _ = l.CloseIdleTSDBTimeout.Set("0") + f.Var(&l.CloseIdleTSDBTimeout, "limits.close-idle-tsdb-timeout", "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to -limits.query-ingesters-within to prevent premature TSDB closure. 0 to disable.") f.Var(&l.MaxQueryLength, "store.max-query-length", "Limit the query time range (end - start time of range query parameter and max - min of data fetched time range). This limit is enforced in the query-frontend and ruler (on the received query). 0 to disable.") f.Var(&l.MaxQueryLookback, "querier.max-query-lookback", "Limit how long back data (series and metadata) can be queried, up until duration ago. This limit is enforced in the query-frontend, querier and ruler. If the requested time range is outside the allowed range, the request will not fail but will be manipulated to only query data within the allowed time range. 0 to disable.") @@ -467,8 +471,8 @@ func (l *Limits) ValidateQueryLimits(userID string, closeIdleTSDBTimeout time.Du queryStoreAfter := time.Duration(l.QueryStoreAfter) shuffleShardingLookback := time.Duration(l.ShuffleShardingIngestersLookbackPeriod) - if queryIngestersWithin > 0 && closeIdleTSDBTimeout > 0 && queryIngestersWithin >= closeIdleTSDBTimeout { - return fmt.Errorf("tenant %s: query_ingesters_within (%s) must be less than close_idle_tsdb_timeout (%s)", + if queryIngestersWithin > 0 && closeIdleTSDBTimeout > 0 && queryIngestersWithin > closeIdleTSDBTimeout { + return fmt.Errorf("tenant %s: query_ingesters_within (%s) must be less than or equal to close_idle_tsdb_timeout (%s)", userID, queryIngestersWithin, closeIdleTSDBTimeout) } @@ -482,6 +486,11 @@ func (l *Limits) ValidateQueryLimits(userID string, closeIdleTSDBTimeout time.Du userID, shuffleShardingLookback, queryStoreAfter) } + if queryIngestersWithin > 0 && shuffleShardingLookback > 0 && shuffleShardingLookback < queryIngestersWithin { + return fmt.Errorf("tenant %s: shuffle_sharding_ingesters_lookback_period (%s) is less than query_ingesters_within (%s)", + userID, shuffleShardingLookback, queryIngestersWithin) + } + return nil } @@ -1285,6 +1294,10 @@ func (o *Overrides) ShuffleShardingIngestersLookbackPeriod(userID string) time.D return time.Duration(o.GetOverridesForUser(userID).ShuffleShardingIngestersLookbackPeriod) } +func (o *Overrides) CloseIdleTSDBTimeout(userID string) time.Duration { + return time.Duration(o.GetOverridesForUser(userID).CloseIdleTSDBTimeout) +} + // GetOverridesForUser returns the per-tenant limits with overrides. func (o *Overrides) GetOverridesForUser(userID string) *Limits { if o.tenantLimits != nil { diff --git a/pkg/util/validation/limits_test.go b/pkg/util/validation/limits_test.go index 31c832dcc73..a4a1c90822a 100644 --- a/pkg/util/validation/limits_test.go +++ b/pkg/util/validation/limits_test.go @@ -1052,19 +1052,19 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { closeIdleTSDBTimeout: 0, expectedError: "", }, - "invalid: queryIngestersWithin >= closeIdleTSDBTimeout": { + "valid: queryIngestersWithin == closeIdleTSDBTimeout": { queryIngestersWithin: 25 * time.Hour, queryStoreAfter: 24 * time.Hour, shuffleShardingIngestersLookbackPeriod: 25 * time.Hour, closeIdleTSDBTimeout: 25 * time.Hour, - expectedError: "query_ingesters_within (25h0m0s) must be less than close_idle_tsdb_timeout (25h0m0s)", + expectedError: "", }, "invalid: queryIngestersWithin > closeIdleTSDBTimeout": { queryIngestersWithin: 26 * time.Hour, queryStoreAfter: 24 * time.Hour, shuffleShardingIngestersLookbackPeriod: 26 * time.Hour, closeIdleTSDBTimeout: 25 * time.Hour, - expectedError: "query_ingesters_within (26h0m0s) must be less than close_idle_tsdb_timeout (25h0m0s)", + expectedError: "query_ingesters_within (26h0m0s) must be less than or equal to close_idle_tsdb_timeout (25h0m0s)", }, "invalid: queryStoreAfter >= queryIngestersWithin": { queryIngestersWithin: 24 * time.Hour, @@ -1087,12 +1087,12 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "shuffle_sharding_ingesters_lookback_period (20h0m0s) is less than query_store_after (24h0m0s)", }, - "valid: shuffleShardingLookback between queryStoreAfter and queryIngestersWithin": { + "invalid: shuffleShardingLookback < queryIngestersWithin": { queryIngestersWithin: 25 * time.Hour, queryStoreAfter: 20 * time.Hour, shuffleShardingIngestersLookbackPeriod: 22 * time.Hour, closeIdleTSDBTimeout: 26 * time.Hour, - expectedError: "", + expectedError: "shuffle_sharding_ingesters_lookback_period (22h0m0s) is less than query_ingesters_within (25h0m0s)", }, "boundary: queryIngestersWithin exactly 1ms less than closeIdleTSDBTimeout": { queryIngestersWithin: 25*time.Hour - time.Millisecond, @@ -1108,10 +1108,10 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "", }, - "boundary: shuffleShardingLookback exactly equal to queryStoreAfter": { + "boundary: shuffleShardingLookback exactly equal to queryIngestersWithin": { queryIngestersWithin: 25 * time.Hour, queryStoreAfter: 24 * time.Hour, - shuffleShardingIngestersLookbackPeriod: 24 * time.Hour, + shuffleShardingIngestersLookbackPeriod: 25 * time.Hour, closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "", }, @@ -1237,16 +1237,16 @@ func TestQueryLimits_TenantOverridesValidation(t *testing.T) { tenantID: "invalid-tenant", expectedError: "query_store_after (25h0m0s) must be less than query_ingesters_within (24h0m0s)", }, - "invalid tenant: queryIngestersWithin >= closeIdleTSDBTimeout": { + "invalid tenant: queryIngestersWithin > closeIdleTSDBTimeout": { tenantLimits: map[string]*Limits{ "invalid-tenant": { - QueryIngestersWithin: model.Duration(26 * time.Hour), + QueryIngestersWithin: model.Duration(27 * time.Hour), QueryStoreAfter: model.Duration(24 * time.Hour), - ShuffleShardingIngestersLookbackPeriod: model.Duration(26 * time.Hour), + ShuffleShardingIngestersLookbackPeriod: model.Duration(27 * time.Hour), }, }, tenantID: "invalid-tenant", - expectedError: "query_ingesters_within (26h0m0s) must be less than close_idle_tsdb_timeout (26h0m0s)", + expectedError: "query_ingesters_within (27h0m0s) must be less than or equal to close_idle_tsdb_timeout (26h0m0s)", }, "invalid tenant: shuffleShardingLookback < queryStoreAfter": { tenantLimits: map[string]*Limits{ @@ -1278,3 +1278,77 @@ func TestQueryLimits_TenantOverridesValidation(t *testing.T) { }) } } + +func TestQueryLimits_PerTenantCloseIdleTSDBTimeout(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + tenantCloseIdleTSDBTimeout time.Duration + queryIngestersWithin time.Duration + expectedError string + }{ + "tenant override allows larger query_ingesters_within": { + tenantCloseIdleTSDBTimeout: 30 * time.Hour, + queryIngestersWithin: 27 * time.Hour, + expectedError: "", + }, + "tenant override allows query_ingesters_within equal to close_idle_tsdb_timeout": { + tenantCloseIdleTSDBTimeout: 28 * time.Hour, + queryIngestersWithin: 28 * time.Hour, + expectedError: "", + }, + "tenant override is validated against its own close_idle_tsdb_timeout": { + tenantCloseIdleTSDBTimeout: 28 * time.Hour, + queryIngestersWithin: 29 * time.Hour, + expectedError: "query_ingesters_within (29h0m0s) must be less than or equal to close_idle_tsdb_timeout (28h0m0s)", + }, + "zero close_idle_tsdb_timeout skips validation": { + tenantCloseIdleTSDBTimeout: 0, + queryIngestersWithin: 100 * time.Hour, + expectedError: "", + }, + } + + for testName, testData := range tests { + t.Run(testName, func(t *testing.T) { + limits := Limits{ + QueryIngestersWithin: model.Duration(testData.queryIngestersWithin), + CloseIdleTSDBTimeout: model.Duration(testData.tenantCloseIdleTSDBTimeout), + } + + err := limits.ValidateQueryLimits("test-tenant", time.Duration(limits.CloseIdleTSDBTimeout)) + + if testData.expectedError == "" { + assert.NoError(t, err, "expected no error but got: %v", err) + } else { + assert.Error(t, err, "expected error but got none") + if err != nil { + assert.Contains(t, err.Error(), testData.expectedError, "error message mismatch") + } + } + }) + } +} + +func TestCloseIdleTSDBTimeout_OverridesAccessor(t *testing.T) { + t.Parallel() + + tenantLimits := map[string]*Limits{ + "tenant-with-override": { + CloseIdleTSDBTimeout: model.Duration(30 * time.Hour), + }, + "tenant-with-zero": { + CloseIdleTSDBTimeout: 0, + }, + } + + defaults := Limits{ + CloseIdleTSDBTimeout: 0, + } + + ov := NewOverrides(defaults, newMockTenantLimits(tenantLimits)) + + assert.Equal(t, 30*time.Hour, ov.CloseIdleTSDBTimeout("tenant-with-override")) + assert.Equal(t, time.Duration(0), ov.CloseIdleTSDBTimeout("tenant-with-zero")) + assert.Equal(t, time.Duration(0), ov.CloseIdleTSDBTimeout("unknown-tenant")) +} diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index 776aa5ee08d..fff70a52ee2 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -3488,13 +3488,6 @@ "type": "array", "x-cli-flag": "blocks-storage.tsdb.block-ranges-period" }, - "close_idle_tsdb_timeout": { - "default": "0s", - "description": "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. If set to positive value, this value must be greater than -limits.query-ingesters-within flag to make sure that TSDB is not closed prematurely, which could cause partial query results. 0 or negative value disables closing of idle TSDB.", - "type": "string", - "x-cli-flag": "blocks-storage.tsdb.close-idle-tsdb-timeout", - "x-format": "duration" - }, "dir": { "default": "tsdb", "description": "Local directory to store TSDBs in the ingesters.", @@ -5683,6 +5676,13 @@ "type": "boolean", "x-cli-flag": "alertmanager.receivers-firewall-block-private-addresses" }, + "close_idle_tsdb_timeout": { + "default": "0s", + "description": "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to -limits.query-ingesters-within to prevent premature TSDB closure. 0 to disable.", + "type": "string", + "x-cli-flag": "limits.close-idle-tsdb-timeout", + "x-format": "duration" + }, "compactor_blocks_retention_period": { "default": "0s", "description": "Delete blocks containing samples older than the specified retention period. 0 to disable.", @@ -6253,7 +6253,7 @@ }, "shuffle_sharding_ingesters_lookback_period": { "default": "0s", - "description": "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-ingesters-within.", + "description": "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-store-after and query-ingesters-within.", "type": "string", "x-cli-flag": "limits.shuffle-sharding-ingesters-lookback-period", "x-format": "duration" From e59ddeaa32cd3486a611b5daf25161e165ea7384 Mon Sep 17 00:00:00 2001 From: Shvejan Date: Sun, 15 Nov 2026 04:06:41 +0000 Subject: [PATCH 2/2] addressing comments Signed-off-by: Shvejan --- CHANGELOG.md | 1 + docs/configuration/config-file-reference.md | 11 +- pkg/cortex/cortex.go | 3 +- pkg/cortex/cortex_test.go | 19 ++ pkg/cortex/runtime_config.go | 3 +- pkg/distributor/distributor_test.go | 4 +- pkg/ingester/ingester.go | 42 +++- pkg/ingester/ingester_test.go | 214 +++++++++++++++----- pkg/storage/tsdb/config.go | 4 + pkg/util/validation/limits.go | 17 +- pkg/util/validation/limits_test.go | 40 +++- schemas/cortex-config-schema.json | 6 +- 12 files changed, 281 insertions(+), 83 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 75220a91e14..d2020610e06 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## master / unreleased +* [CHANGE] Ingester: Make `close_idle_tsdb_timeout` a per-tenant limit (`-limits.close-idle-tsdb-timeout`). `-blocks-storage.tsdb.close-idle-tsdb-timeout` is deprecated and will be removed in v1.24.0. A positive `close_idle_tsdb_timeout` must now be at least `query_store_after` when `query_ingesters_within` is 0. #7813 * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873 * [FEATURE] Storage: Add support for Alibaba Cloud OSS as a backend for blocks, ruler, and alertmanager storage. Configured via `-.oss.*` flags with `backend: oss`. #7907 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 98d989b95cd..b8004a1bb46 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -4817,7 +4817,7 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # Maximum lookback duration for querying data from ingesters. Queries for data # older than this will only query the long-term storage. This is a per-tenant # limit that can be overridden in the runtime configuration. Must be less than -# -blocks-storage.tsdb.close-idle-tsdb-timeout when both are positive. +# or equal to -limits.close-idle-tsdb-timeout when both are positive. # CLI flag: -limits.query-ingesters-within [query_ingesters_within: | default = 0s] @@ -4829,15 +4829,16 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # Lookback period for shuffle sharding of ingesters. This is a per-tenant limit # that can be overridden in the runtime configuration. Should be greater than or -# equal to query-store-after and query-ingesters-within. +# equal to query-ingesters-within. # CLI flag: -limits.shuffle-sharding-ingesters-lookback-period [shuffle_sharding_ingesters_lookback_period: | default = 0s] # If TSDB has not received any data for this duration, and all blocks from TSDB # have been shipped, TSDB is closed and deleted from local disk. This is a -# per-tenant limit that can be overridden in the runtime configuration. Should -# be greater than or equal to -limits.query-ingesters-within to prevent -# premature TSDB closure. 0 to disable. +# per-tenant limit that can be overridden in the runtime configuration. When +# positive, it must be greater than or equal to -limits.query-ingesters-within +# and -limits.query-store-after, so that a TSDB is not closed while queriers +# still read its data only from ingesters. 0 to disable. # CLI flag: -limits.close-idle-tsdb-timeout [close_idle_tsdb_timeout: | default = 0s] diff --git a/pkg/cortex/cortex.go b/pkg/cortex/cortex.go index f3c4eaa8479..2c942889f41 100644 --- a/pkg/cortex/cortex.go +++ b/pkg/cortex/cortex.go @@ -315,7 +315,7 @@ func (c *Config) Validate(log log.Logger) error { if err := c.LimitsConfig.Validate(c.NameValidationScheme, c.Distributor.ShardByAllLabels, c.Ingester.ActiveSeriesMetricsEnabled, c.Distributor.HATrackerConfig.UpdateTimeout, c.Distributor.HATrackerConfig.UpdateTimeoutJitterMax); err != nil { return errors.Wrap(err, "invalid limits config") } - if err := c.LimitsConfig.ValidateQueryLimits("default", time.Duration(c.LimitsConfig.CloseIdleTSDBTimeout)); err != nil { + if err := c.LimitsConfig.ValidateQueryLimits("default"); err != nil { return errors.Wrap(err, "invalid query routing config") } if err := c.ResourceMonitor.Validate(); err != nil { @@ -390,6 +390,7 @@ func (c *Config) deprecatedLimits() []deprecatedLimit { {"querier.query_store_after", "limits.query_store_after", "querier.query-store-after", "limits.query-store-after", c.Querier.DeprecatedQueryStoreAfter, &c.LimitsConfig.QueryStoreAfter, defaults.QueryStoreAfter}, {"querier.shuffle_sharding_ingesters_lookback_period", "limits.shuffle_sharding_ingesters_lookback_period", "querier.shuffle-sharding-ingesters-lookback-period", "limits.shuffle-sharding-ingesters-lookback-period", c.Querier.DeprecatedShuffleShardingIngestersLookbackPeriod, &c.LimitsConfig.ShuffleShardingIngestersLookbackPeriod, defaults.ShuffleShardingIngestersLookbackPeriod}, {"distributor.ha_tracker.ha_tracker_failover_timeout", "limits.ha_tracker_failover_timeout", "", "distributor.ha-tracker.failover-timeout", c.Distributor.HATrackerConfig.DeprecatedFailoverTimeout, &c.LimitsConfig.HATrackerFailoverTimeout, defaults.HATrackerFailoverTimeout}, + {"blocks_storage.tsdb.close_idle_tsdb_timeout", "limits.close_idle_tsdb_timeout", "blocks-storage.tsdb.close-idle-tsdb-timeout", "limits.close-idle-tsdb-timeout", c.BlocksStorage.TSDB.DeprecatedCloseIdleTSDBTimeout, &c.LimitsConfig.CloseIdleTSDBTimeout, defaults.CloseIdleTSDBTimeout}, } } diff --git a/pkg/cortex/cortex_test.go b/pkg/cortex/cortex_test.go index 457fa59b636..489a5d50fde 100644 --- a/pkg/cortex/cortex_test.go +++ b/pkg/cortex/cortex_test.go @@ -339,6 +339,25 @@ func TestConfigValidation_DeprecatedLimits(t *testing.T) { assert.Equal(t, model.Duration(45*time.Second), cfg.LimitsConfig.HATrackerFailoverTimeout) }, }, + { + name: "old blocks storage YAML field sets the default close-idle TSDB timeout limit", + yaml: "blocks_storage:\n tsdb:\n close_idle_tsdb_timeout: 14h\n", + check: func(t *testing.T, cfg *Config) { + assert.Equal(t, model.Duration(14*time.Hour), cfg.LimitsConfig.CloseIdleTSDBTimeout) + }, + }, + { + name: "old blocks storage CLI flag sets the default close-idle TSDB timeout limit", + args: []string{"-blocks-storage.tsdb.close-idle-tsdb-timeout=14h"}, + check: func(t *testing.T, cfg *Config) { + assert.Equal(t, model.Duration(14*time.Hour), cfg.LimitsConfig.CloseIdleTSDBTimeout) + }, + }, + { + name: "old and new close-idle TSDB timeout set to different values", + yaml: "blocks_storage:\n tsdb:\n close_idle_tsdb_timeout: 14h\nlimits:\n close_idle_tsdb_timeout: 15h\n", + expectedError: "are both set: remove blocks_storage.tsdb.close_idle_tsdb_timeout", + }, { name: "old and new set to the same value", yaml: "querier:\n query_store_after: 12h\nlimits:\n query_store_after: 12h\n", diff --git a/pkg/cortex/runtime_config.go b/pkg/cortex/runtime_config.go index f5d849f7423..26d7f582f3a 100644 --- a/pkg/cortex/runtime_config.go +++ b/pkg/cortex/runtime_config.go @@ -5,7 +5,6 @@ import ( "io" "net/http" "strings" - "time" "gopkg.in/yaml.v2" @@ -81,7 +80,7 @@ func (l runtimeConfigLoader) load(r io.Reader) (any, error) { if err := ul.Validate(l.cfg.NameValidationScheme, l.cfg.Distributor.ShardByAllLabels, l.cfg.Ingester.ActiveSeriesMetricsEnabled, l.cfg.Distributor.HATrackerConfig.UpdateTimeout, l.cfg.Distributor.HATrackerConfig.UpdateTimeoutJitterMax); err != nil { return nil, err } - if err := ul.ValidateQueryLimits(userID, time.Duration(ul.CloseIdleTSDBTimeout)); err != nil { + if err := ul.ValidateQueryLimits(userID); err != nil { return nil, err } } diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go index 79c95bc71a1..e8d0fa8c608 100644 --- a/pkg/distributor/distributor_test.go +++ b/pkg/distributor/distributor_test.go @@ -4939,9 +4939,9 @@ func TestDistributor_ShuffleShardingIngestersLookbackPeriod_Validation(t *testin flagext.DefaultValues(limits) limits.QueryStoreAfter = model.Duration(testData.queryStoreAfter) limits.ShuffleShardingIngestersLookbackPeriod = model.Duration(testData.shuffleShardingIngestersLookbackPeriod) + limits.CloseIdleTSDBTimeout = model.Duration(13 * time.Hour) - // ValidateQueryLimits requires userID and closeIdleTSDBTimeout - err := limits.ValidateQueryLimits("test-user", 13*time.Hour) + err := limits.ValidateQueryLimits("test-user") if testData.shouldBeValid { assert.NoError(t, err, testData.description) diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 93f2e707bbd..00b7064dc65 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -3556,18 +3556,19 @@ func (i *Ingester) closeAndDeleteIdleUserTSDBs(ctx context.Context) error { return nil } - // A non-positive close-idle timeout disables idle-based closing for this tenant, - // so we skip the idle check entirely (and don't record an idleTsdbChecks result, - // since no check is performed). A tenant marked for deletion is the one exception: + // A non-positive close-idle timeout, or an invalid one, disables idle-based closing for + // this tenant, so we skip the idle check entirely (and don't record an idleTsdbChecks + // result, since no check is performed). A tenant marked for deletion is the one exception: // its local TSDB must still be closed and deleted regardless of the timeout. - if i.limits.CloseIdleTSDBTimeout(userID) <= 0 { + timeout := i.closeIdleTSDBTimeout(userID) + if timeout <= 0 { userDB, err := i.getTSDB(userID) if err != nil || userDB == nil || !userDB.deletionMarkFound.Load() { continue } } - result := i.closeAndDeleteUserTSDBIfIdle(userID) + result := i.closeAndDeleteUserTSDBIfIdle(userID, timeout) i.TSDBState.idleTsdbChecks.WithLabelValues(string(result)).Inc() } @@ -3590,15 +3591,40 @@ func (i *Ingester) expirePostingsCache(ctx context.Context) error { return nil } -func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckResult { +// closeIdleTSDBTimeout returns the close-idle TSDB timeout for the tenant, or 0 (disabled) when it is +// shorter than the period in which queriers read the tenant's data only from ingesters. The runtime +// config is not validated on ingesters, so a per-tenant override could otherwise close a TSDB that +// queriers still read from. +func (i *Ingester) closeIdleTSDBTimeout(userID string) time.Duration { + timeout := i.limits.CloseIdleTSDBTimeout(userID) + if timeout <= 0 { + return timeout + } + + // With query_ingesters_within set, queriers read from ingesters up to that age; with it + // disabled, they read data younger than query_store_after only from ingesters. + minTimeout, minTimeoutName := i.limits.QueryIngestersWithin(userID), "query_ingesters_within" + if minTimeout <= 0 { + minTimeout, minTimeoutName = i.limits.QueryStoreAfter(userID), "query_store_after" + } + + if timeout < minTimeout { + level.Error(i.logger).Log("msg", "invalid close_idle_tsdb_timeout, closing idle TSDB is disabled for the tenant", "user", userID, "err", fmt.Sprintf("close_idle_tsdb_timeout (%s) must be greater than or equal to %s (%s)", timeout, minTimeoutName, minTimeout)) + return 0 + } + return timeout +} + +// closeAndDeleteUserTSDBIfIdle closes and deletes the tenant's TSDB if it has been idle for longer than +// closeIdleTSDBTimeout, or if the tenant is marked for deletion. Callers must not pass a non-positive +// timeout for a tenant whose TSDB should stay open: a TSDB counts as idle immediately with such a timeout. +func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string, closeIdleTSDBTimeout time.Duration) tsdbCloseCheckResult { userDB, err := i.getTSDB(userID) if err != nil || userDB == nil || userDB.shipper == nil { // We will not delete local data when not using shipping to storage. return tsdbShippingDisabled } - closeIdleTSDBTimeout := i.limits.CloseIdleTSDBTimeout(userID) - if result := userDB.shouldCloseTSDB(closeIdleTSDBTimeout); !result.shouldClose() { return result } diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index 0456719ed1a..3f532d33596 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -656,7 +656,7 @@ func TestIngesterPerLabelsetLimitExceeded(t *testing.T) { // Force set tenant to be deleted. db, _ := ing.getTSDB(userID) db.deletionMarkFound.Store(true) - require.Equal(t, tsdbTenantMarkedForDeletion, ing.closeAndDeleteUserTSDBIfIdle(userID)) + require.Equal(t, tsdbTenantMarkedForDeletion, ing.closeAndDeleteUserTSDBIfIdle(userID, 0)) // LabelSet metrics cleaned up. require.NoError(t, testutil.GatherAndCompare(registry, bytes.NewBufferString(``), "cortex_ingester_usage_per_labelset", "cortex_ingester_limits_per_labelset")) @@ -5283,7 +5283,7 @@ func TestIngester_dontShipBlocksWhenTenantDeletionMarkerIsPresent(t *testing.T) numObjectsAfterMarkingTenantForDeletion := len(bucket.Objects()) require.Equal(t, numObjects, numObjectsAfterMarkingTenantForDeletion) - require.Equal(t, tsdbTenantMarkedForDeletion, i.closeAndDeleteUserTSDBIfIdle(userID)) + require.Equal(t, tsdbTenantMarkedForDeletion, i.closeAndDeleteUserTSDBIfIdle(userID, 0)) } func TestIngester_seriesCountIsCorrectAfterClosingTSDBForDeletedTenant(t *testing.T) { @@ -5324,7 +5324,7 @@ func TestIngester_seriesCountIsCorrectAfterClosingTSDBForDeletedTenant(t *testin // If we try to close TSDB now, it should succeed, even though TSDB is not idle and empty. require.Equal(t, uint64(1), db.Head().NumSeries()) - require.Equal(t, tsdbTenantMarkedForDeletion, i.closeAndDeleteUserTSDBIfIdle(userID)) + require.Equal(t, tsdbTenantMarkedForDeletion, i.closeAndDeleteUserTSDBIfIdle(userID, 0)) // Closing should decrease series count. require.Equal(t, int64(0), i.TSDBState.seriesCount.Load()) @@ -5410,7 +5410,7 @@ func TestIngester_closeAndDeleteUserTSDBIfIdle_shouldNotCloseTSDBIfShippingIsInP defer db.stateMtx.RUnlock() return db.state }) - assert.Equal(t, tsdbNotActive, i.closeAndDeleteUserTSDBIfIdle(userID)) + assert.Equal(t, tsdbNotActive, i.closeAndDeleteUserTSDBIfIdle(userID, 0)) } func TestIngester_closingAndOpeningTsdbConcurrently(t *testing.T) { @@ -5451,7 +5451,7 @@ func TestIngester_closingAndOpeningTsdbConcurrently(t *testing.T) { }() for range iterations { - i.closeAndDeleteUserTSDBIfIdle(userID) + i.closeAndDeleteUserTSDBIfIdle(userID, 0) } select { @@ -5490,7 +5490,7 @@ func TestIngester_idleCloseEmptyTSDB(t *testing.T) { i.shipBlocks(context.Background(), nil) // Make sure we can close completely empty TSDB without problems. - require.Equal(t, tsdbIdleClosed, i.closeAndDeleteUserTSDBIfIdle(userID)) + require.Equal(t, tsdbIdleClosed, i.closeAndDeleteUserTSDBIfIdle(userID, 0)) // Verify that it was closed. db, err = i.getTSDB(userID) @@ -6290,22 +6290,20 @@ func TestIngesterCompactAndCloseIdleTSDB(t *testing.T) { `), metricsToCheck...)) } -func TestIngesterCloseIdleTSDB_PerTenantTimeout(t *testing.T) { +// closeIdleTSDBTestConfig returns an ingester config that compacts, ships and checks idle TSDBs +// every closeIdleInterval, so that a test can observe the close-idle loop within seconds. +func closeIdleTSDBTestConfig(t *testing.T, closeIdleInterval time.Duration) Config { cfg := defaultIngesterTestConfig(t) cfg.LifecyclerConfig.JoinAfter = 0 - cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Second + cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Second // Required to enable shipping. cfg.BlocksStorageConfig.TSDB.ShipConcurrency = 1 cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Second cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout = 1 * time.Second - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = 100 * time.Millisecond - - limits := defaultLimitsTestConfig() - limits.EnableNativeHistograms = true - // Per-tenant limit enables close-idle for this tenant - limits.CloseIdleTSDBTimeout = model.Duration(1 * time.Second) - - tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &limits}) + cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = closeIdleInterval + return cfg +} +func startCloseIdleTSDBTestIngester(t *testing.T, cfg Config, limits validation.Limits, tenantLimits validation.TenantLimits) *Ingester { i, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, tenantLimits, "", prometheus.NewRegistry()) require.NoError(t, err) @@ -6317,59 +6315,181 @@ func TestIngesterCloseIdleTSDB_PerTenantTimeout(t *testing.T) { test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { return i.lifecycler.GetState() }) + return i +} - pushSingleSampleWithMetadata(t, i) +func pushSingleSampleForUser(t *testing.T, i *Ingester, tenant string) { + ctx := user.InjectOrgID(context.Background(), tenant) + req, _ := mockWriteRequest(t, labels.FromStrings("__name__", "test"), 0, util.TimeToMillis(time.Now())) + _, err := i.Push(ctx, req) + require.NoError(t, err) +} - require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) +func hasTSDB(i *Ingester, tenant string) bool { + i.stoppedMtx.Lock() + defer i.stoppedMtx.Unlock() + _, ok := i.TSDBState.dbs[tenant] + return ok +} - // Wait until TSDB has been closed via the per-tenant timeout. - test.Poll(t, 10*time.Second, 0, func() any { - i.stoppedMtx.Lock() - defer i.stoppedMtx.Unlock() - return len(i.TSDBState.dbs) +// waitUntilTSDBCompactedAndShipped waits until the only thing keeping the tenant's TSDB open is +// the close-idle timeout: its head has been compacted and every block has been shipped. +func waitUntilTSDBCompactedAndShipped(t *testing.T, i *Ingester, tenant string) { + test.Poll(t, 10*time.Second, true, func() any { + db, err := i.getTSDB(tenant) + if err != nil || db == nil { + return false + } + return db.Head().NumSeries() == 0 && db.getOldestUnshippedBlockTime() == 0 }) +} + +func TestIngesterCloseIdleTSDB_PerTenantTimeout(t *testing.T) { + const otherUserID = "other-user" + closeIdleInterval := 100 * time.Millisecond + cfg := closeIdleTSDBTestConfig(t, closeIdleInterval) + + // The default leaves close-idle disabled; only userID overrides it. + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = 0 + + userLimits := limits + userLimits.CloseIdleTSDBTimeout = model.Duration(1 * time.Second) + tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &userLimits}) + + i := startCloseIdleTSDBTestIngester(t, cfg, limits, tenantLimits) + pushSingleSampleForUser(t, i, userID) + pushSingleSampleForUser(t, i, otherUserID) + require.Equal(t, int64(2), i.TSDBState.seriesCount.Load()) + + // The TSDB of the tenant with the override is closed. + test.Poll(t, 10*time.Second, false, func() any { + return hasTSDB(i, userID) + }) require.Greater(t, testutil.ToFloat64(i.TSDBState.idleTsdbChecks.WithLabelValues(string(tsdbIdleClosed))), float64(0)) + + // The TSDB of the tenant on the default stays open, even once it could be closed. + waitUntilTSDBCompactedAndShipped(t, i, otherUserID) + time.Sleep(5 * closeIdleInterval) + require.True(t, hasTSDB(i, otherUserID)) } func TestIngesterCloseIdleTSDB_DisabledWhenZero(t *testing.T) { - cfg := defaultIngesterTestConfig(t) - cfg.LifecyclerConfig.JoinAfter = 0 - cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Second - cfg.BlocksStorageConfig.TSDB.ShipConcurrency = 1 - cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Second - cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout = 1 * time.Second - cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBInterval = 100 * time.Millisecond + closeIdleInterval := 100 * time.Millisecond + cfg := closeIdleTSDBTestConfig(t, closeIdleInterval) limits := defaultLimitsTestConfig() limits.EnableNativeHistograms = true limits.CloseIdleTSDBTimeout = 0 // Disabled - tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &limits}) + i := startCloseIdleTSDBTestIngester(t, cfg, limits, nil) - i, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, tenantLimits, "", prometheus.NewRegistry()) - require.NoError(t, err) + pushSingleSampleWithMetadata(t, i) + require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) - require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) - t.Cleanup(func() { - require.NoError(t, services.StopAndAwaitTerminated(context.Background(), i)) - }) + // Once the head is compacted and shipped, only the disabled timeout keeps the TSDB open. + waitUntilTSDBCompactedAndShipped(t, i, userID) - test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { - return i.lifecycler.GetState() - }) + // Let the close-idle loop run several more times: the TSDB must not be closed. + time.Sleep(5 * closeIdleInterval) + require.True(t, hasTSDB(i, userID)) + require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.idleTsdbChecks.WithLabelValues(string(tsdbIdleClosed)))) +} + +func TestIngesterCloseIdleTSDB_DisabledWhenTimeoutInvalid(t *testing.T) { + closeIdleInterval := 100 * time.Millisecond + cfg := closeIdleTSDBTestConfig(t, closeIdleInterval) + + // An override that queriers would reject: the ingester must ignore it and not close the TSDB, + // because queriers read the last hour of data only from ingesters. + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = model.Duration(1 * time.Second) + limits.QueryIngestersWithin = model.Duration(1 * time.Hour) + + i := startCloseIdleTSDBTestIngester(t, cfg, limits, nil) pushSingleSampleWithMetadata(t, i) + waitUntilTSDBCompactedAndShipped(t, i, userID) - require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) + time.Sleep(5 * closeIdleInterval) + require.True(t, hasTSDB(i, userID)) + require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.idleTsdbChecks.WithLabelValues(string(tsdbIdleClosed)))) +} - // Wait a bit and confirm TSDB is NOT closed (timeout is disabled). - time.Sleep(500 * time.Millisecond) +func TestIngesterCloseIdleTSDB_TenantMarkedForDeletionClosedWhenDisabled(t *testing.T) { + // A long interval keeps the timer out of the way; the test runs the loop itself. + cfg := closeIdleTSDBTestConfig(t, 1*time.Hour) - i.stoppedMtx.Lock() - numDBs := len(i.TSDBState.dbs) - i.stoppedMtx.Unlock() - require.Equal(t, 1, numDBs) + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = 0 // Disabled + + i := startCloseIdleTSDBTestIngester(t, cfg, limits, nil) + + pushSingleSampleWithMetadata(t, i) + db, err := i.getTSDB(userID) + require.NoError(t, err) + require.NotNil(t, db) + db.deletionMarkFound.Store(true) + + require.NoError(t, i.closeAndDeleteIdleUserTSDBs(context.Background())) + + require.False(t, hasTSDB(i, userID)) + require.Equal(t, float64(1), testutil.ToFloat64(i.TSDBState.idleTsdbChecks.WithLabelValues(string(tsdbTenantMarkedForDeletion)))) +} + +func TestIngester_closeIdleTSDBTimeout(t *testing.T) { + for name, tc := range map[string]struct { + closeIdleTSDBTimeout time.Duration + queryIngestersWithin time.Duration + queryStoreAfter time.Duration + expected time.Duration + }{ + "disabled": { + closeIdleTSDBTimeout: 0, + queryIngestersWithin: 1 * time.Hour, + expected: 0, + }, + "greater than query_ingesters_within": { + closeIdleTSDBTimeout: 2 * time.Hour, + queryIngestersWithin: 1 * time.Hour, + expected: 2 * time.Hour, + }, + "equal to query_ingesters_within": { + closeIdleTSDBTimeout: 1 * time.Hour, + queryIngestersWithin: 1 * time.Hour, + expected: 1 * time.Hour, + }, + "less than query_ingesters_within is disabled": { + closeIdleTSDBTimeout: 1 * time.Minute, + queryIngestersWithin: 1 * time.Hour, + queryStoreAfter: 30 * time.Minute, + expected: 0, + }, + "less than query_store_after with query_ingesters_within disabled is disabled": { + closeIdleTSDBTimeout: 1 * time.Hour, + queryStoreAfter: 12 * time.Hour, + expected: 0, + }, + "query_ingesters_within and query_store_after disabled": { + closeIdleTSDBTimeout: 1 * time.Minute, + expected: 1 * time.Minute, + }, + } { + t.Run(name, func(t *testing.T) { + limits := defaultLimitsTestConfig() + limits.CloseIdleTSDBTimeout = model.Duration(tc.closeIdleTSDBTimeout) + limits.QueryIngestersWithin = model.Duration(tc.queryIngestersWithin) + limits.QueryStoreAfter = model.Duration(tc.queryStoreAfter) + overrides := validation.NewOverrides(limits, nil) + + i := &Ingester{limits: overrides, logger: log.NewNopLogger()} + require.Equal(t, tc.expected, i.closeIdleTSDBTimeout(userID)) + }) + } } func verifyCompactedHead(t *testing.T, i *Ingester, expected bool) { diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index 1c391de583a..95636dd681e 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -155,6 +155,9 @@ type TSDBConfig struct { WALCompressionType string `yaml:"wal_compression_type"` WALSegmentSizeBytes int `yaml:"wal_segment_size_bytes"` FlushBlocksOnShutdown bool `yaml:"flush_blocks_on_shutdown"` + // DeprecatedCloseIdleTSDBTimeout is kept so that configs from before close_idle_tsdb_timeout + // became a per-tenant limit keep working. Use limits.close_idle_tsdb_timeout instead. + DeprecatedCloseIdleTSDBTimeout time.Duration `yaml:"close_idle_tsdb_timeout,omitempty" doc:"hidden"` // The size of the in-memory queue used before flushing chunks to the disk. HeadChunksWriteQueueSize int `yaml:"head_chunks_write_queue_size"` @@ -204,6 +207,7 @@ func (cfg *TSDBConfig) RegisterFlags(f *flag.FlagSet) { f.StringVar(&cfg.WALCompressionType, "blocks-storage.tsdb.wal-compression-type", "", "TSDB WAL type. Supported values are: 'snappy', 'zstd' and '' (disable compression)") f.IntVar(&cfg.WALSegmentSizeBytes, "blocks-storage.tsdb.wal-segment-size-bytes", wlog.DefaultSegmentSize, "TSDB WAL segments files max size (bytes).") f.BoolVar(&cfg.FlushBlocksOnShutdown, "blocks-storage.tsdb.flush-blocks-on-shutdown", false, "True to flush blocks to storage on shutdown. If false, incomplete blocks will be reused after restart.") + f.DurationVar(&cfg.DeprecatedCloseIdleTSDBTimeout, "blocks-storage.tsdb.close-idle-tsdb-timeout", 0, "Deprecated (use -limits.close-idle-tsdb-timeout instead) and will be removed in v1.24.0.") f.IntVar(&cfg.HeadChunksWriteQueueSize, "blocks-storage.tsdb.head-chunks-write-queue-size", chunks.DefaultWriteQueueSize, "The size of the in-memory queue used before flushing chunks to the disk.") f.IntVar(&cfg.MaxExemplars, "blocks-storage.tsdb.max-exemplars", 0, "Deprecated (use the per-tenant max_exemplars limit instead) and will be removed in v1.24.0: the global fallback for the maximum number of exemplars stored in TSDB, used only when the per-tenant max_exemplars limit is 0. 0 or less means exemplars are disabled.") 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.") diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 688dc1a84d5..53e25252903 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -328,16 +328,16 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.IntVar(&l.MaxFetchedDataBytesPerQuery, "querier.max-fetched-data-bytes-per-query", 0, "The maximum combined size of all data that a query can fetch from each ingester and storage. This limit is enforced in the querier and ruler for `query`, `query_range` and `series` APIs. 0 to disable.") _ = l.QueryIngestersWithin.Set("0") - f.Var(&l.QueryIngestersWithin, "limits.query-ingesters-within", "Maximum lookback duration for querying data from ingesters. Queries for data older than this will only query the long-term storage. This is a per-tenant limit that can be overridden in the runtime configuration. Must be less than -blocks-storage.tsdb.close-idle-tsdb-timeout when both are positive.") + f.Var(&l.QueryIngestersWithin, "limits.query-ingesters-within", "Maximum lookback duration for querying data from ingesters. Queries for data older than this will only query the long-term storage. This is a per-tenant limit that can be overridden in the runtime configuration. Must be less than or equal to -limits.close-idle-tsdb-timeout when both are positive.") _ = l.QueryStoreAfter.Set("0") f.Var(&l.QueryStoreAfter, "limits.query-store-after", "Minimum age of data before querying the long-term storage. Queries for data younger than this will only query ingesters. This is a per-tenant limit that can be overridden in the runtime configuration.") _ = l.ShuffleShardingIngestersLookbackPeriod.Set("0") - f.Var(&l.ShuffleShardingIngestersLookbackPeriod, "limits.shuffle-sharding-ingesters-lookback-period", "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-store-after and query-ingesters-within.") + f.Var(&l.ShuffleShardingIngestersLookbackPeriod, "limits.shuffle-sharding-ingesters-lookback-period", "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-ingesters-within.") _ = l.CloseIdleTSDBTimeout.Set("0") - f.Var(&l.CloseIdleTSDBTimeout, "limits.close-idle-tsdb-timeout", "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to -limits.query-ingesters-within to prevent premature TSDB closure. 0 to disable.") + f.Var(&l.CloseIdleTSDBTimeout, "limits.close-idle-tsdb-timeout", "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. When positive, it must be greater than or equal to -limits.query-ingesters-within and -limits.query-store-after, so that a TSDB is not closed while queriers still read its data only from ingesters. 0 to disable.") f.Var(&l.MaxQueryLength, "store.max-query-length", "Limit the query time range (end - start time of range query parameter and max - min of data fetched time range). This limit is enforced in the query-frontend and ruler (on the received query). 0 to disable.") f.Var(&l.MaxQueryLookback, "querier.max-query-lookback", "Limit how long back data (series and metadata) can be queried, up until duration ago. This limit is enforced in the query-frontend, querier and ruler. If the requested time range is outside the allowed range, the request will not fail but will be manipulated to only query data within the allowed time range. 0 to disable.") @@ -466,10 +466,11 @@ func (l *Limits) Validate(nameValidationScheme model.ValidationScheme, shardByAl return nil } -func (l *Limits) ValidateQueryLimits(userID string, closeIdleTSDBTimeout time.Duration) error { +func (l *Limits) ValidateQueryLimits(userID string) error { queryIngestersWithin := time.Duration(l.QueryIngestersWithin) queryStoreAfter := time.Duration(l.QueryStoreAfter) shuffleShardingLookback := time.Duration(l.ShuffleShardingIngestersLookbackPeriod) + closeIdleTSDBTimeout := time.Duration(l.CloseIdleTSDBTimeout) if queryIngestersWithin > 0 && closeIdleTSDBTimeout > 0 && queryIngestersWithin > closeIdleTSDBTimeout { return fmt.Errorf("tenant %s: query_ingesters_within (%s) must be less than or equal to close_idle_tsdb_timeout (%s)", @@ -486,9 +487,11 @@ func (l *Limits) ValidateQueryLimits(userID string, closeIdleTSDBTimeout time.Du userID, shuffleShardingLookback, queryStoreAfter) } - if queryIngestersWithin > 0 && shuffleShardingLookback > 0 && shuffleShardingLookback < queryIngestersWithin { - return fmt.Errorf("tenant %s: shuffle_sharding_ingesters_lookback_period (%s) is less than query_ingesters_within (%s)", - userID, shuffleShardingLookback, queryIngestersWithin) + // With query_ingesters_within at 0 the checks above don't bound close_idle_tsdb_timeout, but + // queriers still read data younger than query_store_after only from ingesters. + if closeIdleTSDBTimeout > 0 && queryStoreAfter > 0 && closeIdleTSDBTimeout < queryStoreAfter { + return fmt.Errorf("tenant %s: close_idle_tsdb_timeout (%s) must be greater than or equal to query_store_after (%s)", + userID, closeIdleTSDBTimeout, queryStoreAfter) } return nil diff --git a/pkg/util/validation/limits_test.go b/pkg/util/validation/limits_test.go index a4a1c90822a..6f376085b83 100644 --- a/pkg/util/validation/limits_test.go +++ b/pkg/util/validation/limits_test.go @@ -1087,12 +1087,33 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "shuffle_sharding_ingesters_lookback_period (20h0m0s) is less than query_store_after (24h0m0s)", }, - "invalid: shuffleShardingLookback < queryIngestersWithin": { + "invalid: closeIdleTSDBTimeout < queryStoreAfter with queryIngestersWithin disabled": { + queryIngestersWithin: 0, + queryStoreAfter: 12 * time.Hour, + shuffleShardingIngestersLookbackPeriod: 0, + closeIdleTSDBTimeout: 1 * time.Hour, + expectedError: "close_idle_tsdb_timeout (1h0m0s) must be greater than or equal to query_store_after (12h0m0s)", + }, + "boundary: closeIdleTSDBTimeout == queryStoreAfter with queryIngestersWithin disabled": { + queryIngestersWithin: 0, + queryStoreAfter: 12 * time.Hour, + shuffleShardingIngestersLookbackPeriod: 0, + closeIdleTSDBTimeout: 12 * time.Hour, + expectedError: "", + }, + "valid: closeIdleTSDBTimeout with queryIngestersWithin and queryStoreAfter disabled": { + queryIngestersWithin: 0, + queryStoreAfter: 0, + shuffleShardingIngestersLookbackPeriod: 0, + closeIdleTSDBTimeout: 1 * time.Hour, + expectedError: "", + }, + "valid: shuffleShardingLookback between queryStoreAfter and queryIngestersWithin": { queryIngestersWithin: 25 * time.Hour, queryStoreAfter: 20 * time.Hour, shuffleShardingIngestersLookbackPeriod: 22 * time.Hour, closeIdleTSDBTimeout: 26 * time.Hour, - expectedError: "shuffle_sharding_ingesters_lookback_period (22h0m0s) is less than query_ingesters_within (25h0m0s)", + expectedError: "", }, "boundary: queryIngestersWithin exactly 1ms less than closeIdleTSDBTimeout": { queryIngestersWithin: 25*time.Hour - time.Millisecond, @@ -1108,10 +1129,10 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "", }, - "boundary: shuffleShardingLookback exactly equal to queryIngestersWithin": { + "boundary: shuffleShardingLookback exactly equal to queryStoreAfter": { queryIngestersWithin: 25 * time.Hour, queryStoreAfter: 24 * time.Hour, - shuffleShardingIngestersLookbackPeriod: 25 * time.Hour, + shuffleShardingIngestersLookbackPeriod: 24 * time.Hour, closeIdleTSDBTimeout: 26 * time.Hour, expectedError: "", }, @@ -1137,9 +1158,10 @@ func TestLimits_ValidateQueryLimits(t *testing.T) { QueryIngestersWithin: model.Duration(testData.queryIngestersWithin), QueryStoreAfter: model.Duration(testData.queryStoreAfter), ShuffleShardingIngestersLookbackPeriod: model.Duration(testData.shuffleShardingIngestersLookbackPeriod), + CloseIdleTSDBTimeout: model.Duration(testData.closeIdleTSDBTimeout), } - err := limits.ValidateQueryLimits("test-tenant", testData.closeIdleTSDBTimeout) + err := limits.ValidateQueryLimits("test-tenant") if testData.expectedError == "" { assert.NoError(t, err, "expected no error but got: %v", err) @@ -1263,8 +1285,10 @@ func TestQueryLimits_TenantOverridesValidation(t *testing.T) { for testName, testData := range tests { t.Run(testName, func(t *testing.T) { - limits := testData.tenantLimits[testData.tenantID] - err := limits.ValidateQueryLimits(testData.tenantID, closeIdleTSDBTimeout) + // The tenant inherits close_idle_tsdb_timeout from the defaults. + limits := *testData.tenantLimits[testData.tenantID] + limits.CloseIdleTSDBTimeout = model.Duration(closeIdleTSDBTimeout) + err := limits.ValidateQueryLimits(testData.tenantID) if testData.expectedError == "" { assert.NoError(t, err, "expected no error but got: %v", err) @@ -1316,7 +1340,7 @@ func TestQueryLimits_PerTenantCloseIdleTSDBTimeout(t *testing.T) { CloseIdleTSDBTimeout: model.Duration(testData.tenantCloseIdleTSDBTimeout), } - err := limits.ValidateQueryLimits("test-tenant", time.Duration(limits.CloseIdleTSDBTimeout)) + err := limits.ValidateQueryLimits("test-tenant") if testData.expectedError == "" { assert.NoError(t, err, "expected no error but got: %v", err) diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index fff70a52ee2..c21be9d8384 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -5678,7 +5678,7 @@ }, "close_idle_tsdb_timeout": { "default": "0s", - "description": "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to -limits.query-ingesters-within to prevent premature TSDB closure. 0 to disable.", + "description": "If TSDB has not received any data for this duration, and all blocks from TSDB have been shipped, TSDB is closed and deleted from local disk. This is a per-tenant limit that can be overridden in the runtime configuration. When positive, it must be greater than or equal to -limits.query-ingesters-within and -limits.query-store-after, so that a TSDB is not closed while queriers still read its data only from ingesters. 0 to disable.", "type": "string", "x-cli-flag": "limits.close-idle-tsdb-timeout", "x-format": "duration" @@ -6106,7 +6106,7 @@ }, "query_ingesters_within": { "default": "0s", - "description": "Maximum lookback duration for querying data from ingesters. Queries for data older than this will only query the long-term storage. This is a per-tenant limit that can be overridden in the runtime configuration. Must be less than -blocks-storage.tsdb.close-idle-tsdb-timeout when both are positive.", + "description": "Maximum lookback duration for querying data from ingesters. Queries for data older than this will only query the long-term storage. This is a per-tenant limit that can be overridden in the runtime configuration. Must be less than or equal to -limits.close-idle-tsdb-timeout when both are positive.", "type": "string", "x-cli-flag": "limits.query-ingesters-within", "x-format": "duration" @@ -6253,7 +6253,7 @@ }, "shuffle_sharding_ingesters_lookback_period": { "default": "0s", - "description": "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-store-after and query-ingesters-within.", + "description": "Lookback period for shuffle sharding of ingesters. This is a per-tenant limit that can be overridden in the runtime configuration. Should be greater than or equal to query-ingesters-within.", "type": "string", "x-cli-flag": "limits.shuffle-sharding-ingesters-lookback-period", "x-format": "duration"