diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e07249748f..33ff49674f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * [CHANGE] Query-frontend: removed `-querier.split-queries-by-day` (deprecated in Cortex 0.4.0). You should use `-querier.split-queries-by-interval` instead. #3813 * [CHANGE] Store-gateway: the chunks pool controlled by `-blocks-storage.bucket-store.max-chunk-pool-bytes` is now shared across all tenants. #3830 * [CHANGE] Ingester: return error code 400 instead of 429 when per-user/per-tenant series/metadata limits are reached. #3833 +* [CHANGE] Compactor: add `reason` label to `cortex_compactor_blocks_marked_for_deletion_total` metric. Source blocks marked for deletion by compactor are labelled as `compaction`, while blocks passing the retention period are labelled as `retention`. #3879 * [FEATURE] Experimental Ruler Storage: Add a separate set of configuration options to configure the ruler storage backend under the `-ruler-storage.` flag prefix. All blocks storage bucket clients and the config service are currently supported. Clients using this implementation will only be enabled if the existing `-ruler.storage` flags are left unset. #3805 #3864 * [FEATURE] Adds support to S3 server-side encryption using KMS. The S3 server-side encryption config can be overridden on a per-tenant basis for the blocks storage and ruler. Deprecated `-.s3.sse-encryption`, you should use the following CLI flags that have been added. #3651 #3810 #3811 #3870 - `-.s3.sse.type` @@ -26,6 +27,7 @@ * `-alertmanager.alertmanager-client.tls-ca-path` * `-alertmanager.alertmanager-client.tls-server-name` * `-alertmanager.alertmanager-client.tls-insecure-skip-verify` +* [FEATURE] Compactor: added blocks storage per-tenant retention support. This is configured via `-compactor.retention-period`, and can be overridden on a per-tenant basis. #3879 * [ENHANCEMENT] Ruler: Add TLS and explicit basis authentication configuration options for the HTTP client the ruler uses to communicate with the alertmanager. #3752 * `-ruler.alertmanager-client.basic-auth-username`: Configure the basic authentication username used by the client. Takes precedent over a URL configured username. * `-ruler.alertmanager-client.basic-auth-password`: Configure the basic authentication password used by the client. Takes precedent over a URL configured password. diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 4bee82e6727..6f4b34ad3cf 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3752,6 +3752,11 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # CLI flag: -store-gateway.tenant-shard-size [store_gateway_tenant_shard_size: | default = 0] +# Delete blocks containing samples older than the specified retention period. 0 +# to disable. +# CLI flag: -compactor.blocks-retention-period +[compactor_blocks_retention_period: | default = 0s] + # S3 server-side encryption type. Required to enable server-side encryption # overrides for a specific tenant. If not set, the default S3 client settings # are used. diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 9354d6e6aa1..f6599e8190d 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -2,6 +2,7 @@ package compactor import ( "context" + "fmt" "time" "github.com/go-kit/kit/log" @@ -35,7 +36,7 @@ type BlocksCleaner struct { services.Service cfg BlocksCleanerConfig - cfgProvider bucket.TenantConfigProvider + cfgProvider ConfigProvider logger log.Logger bucketClient objstore.Bucket usersScanner *cortex_tsdb.UsersScanner @@ -50,13 +51,14 @@ type BlocksCleaner struct { runsLastSuccess prometheus.Gauge blocksCleanedTotal prometheus.Counter blocksFailedTotal prometheus.Counter + blocksMarkedForDeletion prometheus.Counter tenantBlocks *prometheus.GaugeVec tenantMarkedBlocks *prometheus.GaugeVec tenantPartialBlocks *prometheus.GaugeVec tenantBucketIndexLastUpdate *prometheus.GaugeVec } -func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider bucket.TenantConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner { +func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider ConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner { c := &BlocksCleaner{ cfg: cfg, bucketClient: bucketClient, @@ -87,6 +89,11 @@ func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, use Name: "cortex_compactor_block_cleanup_failures_total", Help: "Total number of blocks failed to be deleted.", }), + blocksMarkedForDeletion: promauto.With(reg).NewCounter(prometheus.CounterOpts{ + Name: blocksMarkedForDeletionName, + Help: blocksMarkedForDeletionHelp, + ConstLabels: prometheus.Labels{"reason": "retention"}, + }), // The following metrics don't have the "cortex_compactor" prefix because not strictly related to // the compactor. They're just tracked by the compactor because it's the most logical place where these @@ -310,6 +317,17 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b return err } + // Mark blocks for future deletion based on the retention period for the user. + // Note doing this before UpdateIndex, so it reads in the deletion marks. + // The trade-off being that retention is not applied if the index has to be + // built, but this is rare. + if idx != nil { + // We do not want to stop the remaining work in the cleaner if an + // error occurs here. Errors are logged in the function. + retention := c.cfgProvider.CompactorBlocksRetentionPeriod(userID) + c.applyUserRetentionPeriod(ctx, idx, retention, userBucket, userLogger) + } + // Generate an updated in-memory version of the bucket index. w := bucketindex.NewUpdater(c.bucketClient, userID, c.cfgProvider, c.logger) idx, partials, err := w.UpdateIndex(ctx, idx) @@ -391,3 +409,46 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map level.Info(userLogger).Log("msg", "deleted partial block marked for deletion", "block", blockID) } } + +// applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period. +func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, retention time.Duration, userBucket *bucket.UserBucketClient, userLogger log.Logger) { + // The retention period of zero is a special value indicating to never delete. + if retention <= 0 { + return + } + + level.Debug(userLogger).Log("msg", "applying retention", "retention", retention.String()) + blocks := listBlocksOutsideRetentionPeriod(idx, time.Now().Add(-retention)) + + // Attempt to mark all blocks. It is not critical if a marking fails, as + // the cleaner will retry applying the retention in its next cycle. + for _, b := range blocks { + level.Info(userLogger).Log("msg", "applied retention: marking block for deletion", "block", b.ID, "maxTime", b.MaxTime) + if err := block.MarkForDeletion(ctx, userLogger, userBucket, b.ID, fmt.Sprintf("block exceeding retention of %v", retention), c.blocksMarkedForDeletion); err != nil { + level.Warn(userLogger).Log("msg", "failed to mark block for deletion", "block", b.ID, "err", err) + } + } +} + +// listBlocksOutsideRetentionPeriod determines the blocks which have aged past +// the specified retention period, and are not already marked for deletion. +func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, threshold time.Time) (result bucketindex.Blocks) { + // Whilst re-marking a block is not harmful, it is wasteful and generates + // a warning log message. Use the block deletion marks already in-memory + // to prevent marking blocks already marked for deletion. + marked := make(map[ulid.ULID]struct{}, len(idx.BlockDeletionMarks)) + for _, d := range idx.BlockDeletionMarks { + marked[d.ID] = struct{}{} + } + + for _, b := range idx.Blocks { + maxTime := time.Unix(b.MaxTime/1000, 0) + if maxTime.Before(threshold) { + if _, isMarked := marked[b.ID]; !isMarked { + result = append(result, b) + } + } + } + + return +} diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index 1d431b4081b..7576c6a6d80 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -115,8 +115,9 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions reg := prometheus.NewPedanticRegistry() logger := log.NewNopLogger() scanner := tsdb.NewUsersScanner(bucketClient, tsdb.AllUsers, logger) + cfgProvider := newMockConfigProvider() - cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, nil, logger, reg) + cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, reg) require.NoError(t, services.StartAndAwaitRunning(ctx, cleaner)) defer services.StopAndAwaitTerminated(ctx, cleaner) //nolint:errcheck @@ -248,8 +249,9 @@ func TestBlocksCleaner_ShouldContinueOnBlockDeletionFailure(t *testing.T) { logger := log.NewNopLogger() scanner := tsdb.NewUsersScanner(bucketClient, tsdb.AllUsers, logger) + cfgProvider := newMockConfigProvider() - cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, nil, logger, nil) + cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, nil) require.NoError(t, services.StartAndAwaitRunning(ctx, cleaner)) defer services.StopAndAwaitTerminated(ctx, cleaner) //nolint:errcheck @@ -307,8 +309,9 @@ func TestBlocksCleaner_ShouldRebuildBucketIndexOnCorruptedOne(t *testing.T) { logger := log.NewNopLogger() scanner := tsdb.NewUsersScanner(bucketClient, tsdb.AllUsers, logger) + cfgProvider := newMockConfigProvider() - cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, nil, logger, nil) + cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, nil) require.NoError(t, services.StartAndAwaitRunning(ctx, cleaner)) defer services.StopAndAwaitTerminated(ctx, cleaner) //nolint:errcheck @@ -357,8 +360,9 @@ func TestBlocksCleaner_ShouldRemoveMetricsForTenantsNotBelongingAnymoreToTheShar logger := log.NewNopLogger() reg := prometheus.NewPedanticRegistry() scanner := tsdb.NewUsersScanner(bucketClient, tsdb.AllUsers, logger) + cfgProvider := newMockConfigProvider() - cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, nil, logger, reg) + cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, reg) require.NoError(t, cleaner.cleanUsers(ctx, true)) assert.NoError(t, prom_testutil.GatherAndCompare(reg, strings.NewReader(` @@ -406,6 +410,236 @@ func TestBlocksCleaner_ShouldRemoveMetricsForTenantsNotBelongingAnymoreToTheShar )) } +func TestBlocksCleaner_ListBlocksOutsideRetentionPeriod(t *testing.T) { + bucketClient, _ := cortex_testutil.PrepareFilesystemBucket(t) + bucketClient = bucketindex.BucketWithGlobalMarkers(bucketClient) + ctx := context.Background() + logger := log.NewNopLogger() + + id1 := createTSDBBlock(t, bucketClient, "user-1", 5000, 6000, nil) + id2 := createTSDBBlock(t, bucketClient, "user-1", 6000, 7000, nil) + id3 := createTSDBBlock(t, bucketClient, "user-1", 7000, 8000, nil) + + w := bucketindex.NewUpdater(bucketClient, "user-1", nil, logger) + idx, _, err := w.UpdateIndex(ctx, nil) + require.NoError(t, err) + + assert.ElementsMatch(t, []ulid.ULID{id1, id2, id3}, idx.Blocks.GetULIDs()) + + // Excessive retention period (wrapping epoch) + result := listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0).Add(-time.Hour)) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + // Normal operation - varying retention period. + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(6, 0)) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) + assert.ElementsMatch(t, []ulid.ULID{id1}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(8, 0)) + assert.ElementsMatch(t, []ulid.ULID{id1, id2}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(9, 0)) + assert.ElementsMatch(t, []ulid.ULID{id1, id2, id3}, result.GetULIDs()) + + // Avoiding redundant marking - blocks already marked for deletion. + + mark1 := &bucketindex.BlockDeletionMark{ID: id1} + mark2 := &bucketindex.BlockDeletionMark{ID: id2} + + idx.BlockDeletionMarks = bucketindex.BlockDeletionMarks{mark1} + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(8, 0)) + assert.ElementsMatch(t, []ulid.ULID{id2}, result.GetULIDs()) + + idx.BlockDeletionMarks = bucketindex.BlockDeletionMarks{mark1, mark2} + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(8, 0)) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(9, 0)) + assert.ElementsMatch(t, []ulid.ULID{id3}, result.GetULIDs()) +} + +func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { + bucketClient, _ := cortex_testutil.PrepareFilesystemBucket(t) + bucketClient = bucketindex.BucketWithGlobalMarkers(bucketClient) + + ts := func(hours int) int64 { + return time.Now().Add(time.Duration(hours)*time.Hour).Unix() * 1000 + } + + block1 := createTSDBBlock(t, bucketClient, "user-1", ts(-10), ts(-8), nil) + block2 := createTSDBBlock(t, bucketClient, "user-1", ts(-8), ts(-6), nil) + block3 := createTSDBBlock(t, bucketClient, "user-2", ts(-10), ts(-8), nil) + block4 := createTSDBBlock(t, bucketClient, "user-2", ts(-8), ts(-6), nil) + + cfg := BlocksCleanerConfig{ + DeletionDelay: time.Hour, + CleanupInterval: time.Minute, + CleanupConcurrency: 1, + } + + ctx := context.Background() + logger := log.NewNopLogger() + reg := prometheus.NewPedanticRegistry() + scanner := tsdb.NewUsersScanner(bucketClient, tsdb.AllUsers, logger) + cfgProvider := newMockConfigProvider() + + cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, reg) + + assertBlockExists := func(user string, block ulid.ULID, expectExists bool) { + exists, err := bucketClient.Exists(ctx, path.Join(user, block.String(), metadata.MetaFilename)) + require.NoError(t, err) + assert.Equal(t, expectExists, exists) + } + + // Existing behaviour - retention period disabled. + { + cfgProvider.userRetentionPeriods["user-1"] = 0 + cfgProvider.userRetentionPeriods["user-2"] = 0 + + require.NoError(t, cleaner.cleanUsers(ctx, true)) + assertBlockExists("user-1", block1, true) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, true) + assertBlockExists("user-2", block4, true) + + assert.NoError(t, prom_testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP cortex_bucket_blocks_count Total number of blocks in the bucket. Includes blocks marked for deletion, but not partial blocks. + # TYPE cortex_bucket_blocks_count gauge + cortex_bucket_blocks_count{user="user-1"} 2 + cortex_bucket_blocks_count{user="user-2"} 2 + # HELP cortex_bucket_blocks_marked_for_deletion_count Total number of blocks marked for deletion in the bucket. + # TYPE cortex_bucket_blocks_marked_for_deletion_count gauge + cortex_bucket_blocks_marked_for_deletion_count{user="user-1"} 0 + cortex_bucket_blocks_marked_for_deletion_count{user="user-2"} 0 + # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. + # TYPE cortex_compactor_blocks_marked_for_deletion_total counter + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 + `), + "cortex_bucket_blocks_count", + "cortex_bucket_blocks_marked_for_deletion_count", + "cortex_compactor_blocks_marked_for_deletion_total", + )) + } + + // Retention enabled only for a single user, but does nothing. + { + cfgProvider.userRetentionPeriods["user-1"] = 9 * time.Hour + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlockExists("user-1", block1, true) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, true) + assertBlockExists("user-2", block4, true) + } + + // Retention enabled only for a single user, marking a single block. + // Note the block won't be deleted yet due to deletion delay. + { + cfgProvider.userRetentionPeriods["user-1"] = 7 * time.Hour + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlockExists("user-1", block1, true) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, true) + assertBlockExists("user-2", block4, true) + + assert.NoError(t, prom_testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP cortex_bucket_blocks_count Total number of blocks in the bucket. Includes blocks marked for deletion, but not partial blocks. + # TYPE cortex_bucket_blocks_count gauge + cortex_bucket_blocks_count{user="user-1"} 2 + cortex_bucket_blocks_count{user="user-2"} 2 + # HELP cortex_bucket_blocks_marked_for_deletion_count Total number of blocks marked for deletion in the bucket. + # TYPE cortex_bucket_blocks_marked_for_deletion_count gauge + cortex_bucket_blocks_marked_for_deletion_count{user="user-1"} 1 + cortex_bucket_blocks_marked_for_deletion_count{user="user-2"} 0 + # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. + # TYPE cortex_compactor_blocks_marked_for_deletion_total counter + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 1 + `), + "cortex_bucket_blocks_count", + "cortex_bucket_blocks_marked_for_deletion_count", + "cortex_compactor_blocks_marked_for_deletion_total", + )) + } + + // Marking the block again, before the deletion occurs, should not cause an error. + { + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlockExists("user-1", block1, true) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, true) + assertBlockExists("user-2", block4, true) + } + + // Reduce the deletion delay. Now the block will be deleted. + { + cleaner.cfg.DeletionDelay = 0 + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlockExists("user-1", block1, false) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, true) + assertBlockExists("user-2", block4, true) + + assert.NoError(t, prom_testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP cortex_bucket_blocks_count Total number of blocks in the bucket. Includes blocks marked for deletion, but not partial blocks. + # TYPE cortex_bucket_blocks_count gauge + cortex_bucket_blocks_count{user="user-1"} 1 + cortex_bucket_blocks_count{user="user-2"} 2 + # HELP cortex_bucket_blocks_marked_for_deletion_count Total number of blocks marked for deletion in the bucket. + # TYPE cortex_bucket_blocks_marked_for_deletion_count gauge + cortex_bucket_blocks_marked_for_deletion_count{user="user-1"} 0 + cortex_bucket_blocks_marked_for_deletion_count{user="user-2"} 0 + # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. + # TYPE cortex_compactor_blocks_marked_for_deletion_total counter + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 1 + `), + "cortex_bucket_blocks_count", + "cortex_bucket_blocks_marked_for_deletion_count", + "cortex_compactor_blocks_marked_for_deletion_total", + )) + } + + // Retention enabled for other user; test deleting multiple blocks. + { + cfgProvider.userRetentionPeriods["user-2"] = 5 * time.Hour + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlockExists("user-1", block1, false) + assertBlockExists("user-1", block2, true) + assertBlockExists("user-2", block3, false) + assertBlockExists("user-2", block4, false) + + assert.NoError(t, prom_testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP cortex_bucket_blocks_count Total number of blocks in the bucket. Includes blocks marked for deletion, but not partial blocks. + # TYPE cortex_bucket_blocks_count gauge + cortex_bucket_blocks_count{user="user-1"} 1 + cortex_bucket_blocks_count{user="user-2"} 0 + # HELP cortex_bucket_blocks_marked_for_deletion_count Total number of blocks marked for deletion in the bucket. + # TYPE cortex_bucket_blocks_marked_for_deletion_count gauge + cortex_bucket_blocks_marked_for_deletion_count{user="user-1"} 0 + cortex_bucket_blocks_marked_for_deletion_count{user="user-2"} 0 + # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. + # TYPE cortex_compactor_blocks_marked_for_deletion_total counter + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 3 + `), + "cortex_bucket_blocks_count", + "cortex_bucket_blocks_marked_for_deletion_count", + "cortex_compactor_blocks_marked_for_deletion_total", + )) + } +} + type mockBucketFailure struct { objstore.Bucket @@ -418,3 +652,32 @@ func (m *mockBucketFailure) Delete(ctx context.Context, name string) error { } return m.Bucket.Delete(ctx, name) } + +type mockConfigProvider struct { + userRetentionPeriods map[string]time.Duration +} + +func newMockConfigProvider() *mockConfigProvider { + return &mockConfigProvider{ + userRetentionPeriods: make(map[string]time.Duration), + } +} + +func (m *mockConfigProvider) CompactorBlocksRetentionPeriod(user string) time.Duration { + if result, ok := m.userRetentionPeriods[user]; ok { + return result + } + return 0 +} + +func (m *mockConfigProvider) S3SSEType(user string) string { + return "" +} + +func (m *mockConfigProvider) S3SSEKMSKeyID(userID string) string { + return "" +} + +func (m *mockConfigProvider) S3SSEKMSEncryptionContext(userID string) string { + return "" +} diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 7141dd6ede4..efa4ee92f79 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -34,6 +34,11 @@ import ( "github.com/cortexproject/cortex/pkg/util/services" ) +const ( + blocksMarkedForDeletionName = "cortex_compactor_blocks_marked_for_deletion_total" + blocksMarkedForDeletionHelp = "Total number of blocks marked for deletion in compactor." +) + var ( errInvalidBlockRanges = "compactor block range periods should be divisible by the previous one, but %s is not divisible by %s" RingOp = ring.NewOp([]ring.IngesterState{ring.ACTIVE}, nil) @@ -155,13 +160,19 @@ func (cfg *Config) Validate() error { return nil } +// ConfigProvider defines the per-tenant config provider for the Compactor. +type ConfigProvider interface { + bucket.TenantConfigProvider + CompactorBlocksRetentionPeriod(user string) time.Duration +} + // Compactor is a multi-tenant TSDB blocks compactor based on Thanos. type Compactor struct { services.Service compactorCfg Config storageCfg cortex_tsdb.BlocksStorageConfig - cfgProvider bucket.TenantConfigProvider + cfgProvider ConfigProvider logger log.Logger parentLogger log.Logger registerer prometheus.Registerer @@ -214,7 +225,7 @@ type Compactor struct { } // NewCompactor makes a new Compactor. -func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, cfgProvider bucket.TenantConfigProvider, logger log.Logger, registerer prometheus.Registerer) (*Compactor, error) { +func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, cfgProvider ConfigProvider, logger log.Logger, registerer prometheus.Registerer) (*Compactor, error) { bucketClientFactory := func(ctx context.Context) (objstore.Bucket, error) { return bucket.NewClient(ctx, storageCfg.Bucket, "compactor", logger, registerer) } @@ -240,7 +251,7 @@ func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfi func newCompactor( compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, - cfgProvider bucket.TenantConfigProvider, + cfgProvider ConfigProvider, logger log.Logger, registerer prometheus.Registerer, bucketClientFactory func(ctx context.Context) (objstore.Bucket, error), @@ -292,8 +303,9 @@ func newCompactor( Help: "Number of tenants failed processing during the current compaction run. Reset to 0 when compactor is idle.", }), blocksMarkedForDeletion: promauto.With(registerer).NewCounter(prometheus.CounterOpts{ - Name: "cortex_compactor_blocks_marked_for_deletion_total", - Help: "Total number of blocks marked for deletion in compactor.", + Name: blocksMarkedForDeletionName, + Help: blocksMarkedForDeletionHelp, + ConstLabels: prometheus.Labels{"reason": "compaction"}, }), garbageCollectedBlocks: promauto.With(registerer).NewCounter(prometheus.CounterOpts{ Name: "cortex_compactor_garbage_collected_blocks_total", diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go index 1c7b3255d17..d69ee54bc71 100644 --- a/pkg/compactor/compactor_test.go +++ b/pkg/compactor/compactor_test.go @@ -226,7 +226,8 @@ func TestCompactor_ShouldDoNothingOnNoUserBlocks(t *testing.T) { # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. # TYPE cortex_compactor_blocks_marked_for_deletion_total counter - cortex_compactor_blocks_marked_for_deletion_total 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="compaction"} 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 # TYPE cortex_compactor_block_cleanup_started_total counter # HELP cortex_compactor_block_cleanup_started_total Total number of blocks cleanup runs started. @@ -371,7 +372,8 @@ func TestCompactor_ShouldRetryCompactionOnFailureWhileDiscoveringUsersFromBucket # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. # TYPE cortex_compactor_blocks_marked_for_deletion_total counter - cortex_compactor_blocks_marked_for_deletion_total 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="compaction"} 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 # TYPE cortex_compactor_block_cleanup_started_total counter # HELP cortex_compactor_block_cleanup_started_total Total number of blocks cleanup runs started. @@ -504,7 +506,8 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) { # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. # TYPE cortex_compactor_blocks_marked_for_deletion_total counter - cortex_compactor_blocks_marked_for_deletion_total 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="compaction"} 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 # TYPE cortex_compactor_block_cleanup_started_total counter # HELP cortex_compactor_block_cleanup_started_total Total number of blocks cleanup runs started. @@ -625,7 +628,8 @@ func TestCompactor_ShouldNotCompactBlocksMarkedForDeletion(t *testing.T) { # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. # TYPE cortex_compactor_blocks_marked_for_deletion_total counter - cortex_compactor_blocks_marked_for_deletion_total 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="compaction"} 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 # TYPE cortex_compactor_block_cleanup_started_total counter # HELP cortex_compactor_block_cleanup_started_total Total number of blocks cleanup runs started. @@ -729,7 +733,8 @@ func TestCompactor_ShouldNotCompactBlocksForUsersMarkedForDeletion(t *testing.T) # HELP cortex_compactor_blocks_marked_for_deletion_total Total number of blocks marked for deletion in compactor. # TYPE cortex_compactor_blocks_marked_for_deletion_total counter - cortex_compactor_blocks_marked_for_deletion_total 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="compaction"} 0 + cortex_compactor_blocks_marked_for_deletion_total{reason="retention"} 0 # TYPE cortex_compactor_block_cleanup_started_total counter # HELP cortex_compactor_block_cleanup_started_total Total number of blocks cleanup runs started. diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 47f599559e2..be8a08fb56c 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -86,6 +86,9 @@ type Limits struct { // Store-gateway. StoreGatewayTenantShardSize int `yaml:"store_gateway_tenant_shard_size" json:"store_gateway_tenant_shard_size"` + // Compactor. + CompactorBlocksRetentionPeriod time.Duration `yaml:"compactor_blocks_retention_period" json:"compactor_blocks_retention_period"` + // This config doesn't have a CLI flag registered here because they're registered in // their own original config struct. S3SSEType string `yaml:"s3_sse_type" json:"s3_sse_type" doc:"nocli|description=S3 server-side encryption type. Required to enable server-side encryption overrides for a specific tenant. If not set, the default S3 client settings are used."` @@ -144,6 +147,8 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.IntVar(&l.RulerMaxRulesPerRuleGroup, "ruler.max-rules-per-rule-group", 0, "Maximum number of rules per rule group per-tenant. 0 to disable.") f.IntVar(&l.RulerMaxRuleGroupsPerTenant, "ruler.max-rule-groups-per-tenant", 0, "Maximum number of rule groups per-tenant. 0 to disable.") + f.DurationVar(&l.CompactorBlocksRetentionPeriod, "compactor.blocks-retention-period", 0, "Delete blocks containing samples older than the specified retention period. 0 to disable.") + f.StringVar(&l.PerTenantOverrideConfig, "limits.per-user-override-config", "", "File name of per-user overrides. [deprecated, use -runtime-config.file instead]") f.DurationVar(&l.PerTenantOverridePeriod, "limits.per-user-override-period", 10*time.Second, "Period with which to reload the overrides. [deprecated, use -runtime-config.reload-period instead]") @@ -415,6 +420,11 @@ func (o *Overrides) EvaluationDelay(userID string) time.Duration { return o.getOverridesForUser(userID).RulerEvaluationDelay } +// CompactorBlocksRetentionPeriod returns the retention period for a given user. +func (o *Overrides) CompactorBlocksRetentionPeriod(userID string) time.Duration { + return o.getOverridesForUser(userID).CompactorBlocksRetentionPeriod +} + // MetricRelabelConfigs returns the metric relabel configs for a given user. func (o *Overrides) MetricRelabelConfigs(userID string) []*relabel.Config { return o.getOverridesForUser(userID).MetricRelabelConfigs