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/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..b8004a1bb46 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] @@ -4826,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] @@ -4842,6 +4833,15 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # 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. 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] + # 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..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", c.BlocksStorage.TSDB.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 8031b6be5c4..26d7f582f3a 100644 --- a/pkg/cortex/runtime_config.go +++ b/pkg/cortex/runtime_config.go @@ -80,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, l.cfg.BlocksStorage.TSDB.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 d877b90a12f..00b7064dc65 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,7 +3556,19 @@ func (i *Ingester) closeAndDeleteIdleUserTSDBs(ctx context.Context) error { return nil } - result := i.closeAndDeleteUserTSDBIfIdle(userID) + // 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. + 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, timeout) i.TSDBState.idleTsdbChecks.WithLabelValues(string(result)).Inc() } @@ -3577,14 +3591,41 @@ 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 } - if result := userDB.shouldCloseTSDB(i.cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout); !result.shouldClose() { + if result := userDB.shouldCloseTSDB(closeIdleTSDBTimeout); !result.shouldClose() { return result } @@ -3602,7 +3643,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..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,13 +5410,12 @@ 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) { 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()) @@ -5452,7 +5451,7 @@ func TestIngester_closingAndOpeningTsdbConcurrently(t *testing.T) { }() for range iterations { - i.closeAndDeleteUserTSDBIfIdle(userID) + i.closeAndDeleteUserTSDBIfIdle(userID, 0) } select { @@ -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()) @@ -5492,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) @@ -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,208 @@ func TestIngesterCompactAndCloseIdleTSDB(t *testing.T) { `), metricsToCheck...)) } +// 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 // 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 = 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) + + 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() + }) + return 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) +} + +func hasTSDB(i *Ingester, tenant string) bool { + i.stoppedMtx.Lock() + defer i.stoppedMtx.Unlock() + _, ok := i.TSDBState.dbs[tenant] + return ok +} + +// 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) { + closeIdleInterval := 100 * time.Millisecond + cfg := closeIdleTSDBTestConfig(t, closeIdleInterval) + + limits := defaultLimitsTestConfig() + limits.EnableNativeHistograms = true + limits.CloseIdleTSDBTimeout = 0 // Disabled + + i := startCloseIdleTSDBTestIngester(t, cfg, limits, nil) + + pushSingleSampleWithMetadata(t, i) + require.Equal(t, int64(1), i.TSDBState.seriesCount.Load()) + + // Once the head is compacted and shipped, only the disabled timeout keeps the TSDB open. + waitUntilTSDBCompactedAndShipped(t, i, userID) + + // 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) + + 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_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) + + 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) { db, err := i.getTSDB(userID) require.NoError(t, err) @@ -7996,7 +8198,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..95636dd681e 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -155,7 +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"` - CloseIdleTSDBTimeout time.Duration `yaml:"close_idle_tsdb_timeout"` + // 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"` @@ -205,7 +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.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.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/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..53e25252903 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"` @@ -327,7 +328,7 @@ 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.") @@ -335,6 +336,9 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { _ = 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.") + _ = 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. 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.") f.IntVar(&l.MaxQueryParallelism, "querier.max-query-parallelism", 14, "Maximum number of split queries will be scheduled in parallel by the frontend.") @@ -462,13 +466,14 @@ 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 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 +487,13 @@ func (l *Limits) ValidateQueryLimits(userID string, closeIdleTSDBTimeout time.Du userID, shuffleShardingLookback, queryStoreAfter) } + // 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 } @@ -1285,6 +1297,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..6f376085b83 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,6 +1087,27 @@ 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: 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, @@ -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) @@ -1237,16 +1259,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{ @@ -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) @@ -1278,3 +1302,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") + + 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..c21be9d8384 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. 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" + }, "compactor_blocks_retention_period": { "default": "0s", "description": "Delete blocks containing samples older than the specified retention period. 0 to disable.", @@ -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"