From d087cd01cc150c62ed16669d1b8466954cf16a81 Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Mon, 22 Feb 2021 15:52:02 +0100 Subject: [PATCH 1/9] Add CompactorRetentionPeriod to per-tenant configurables. Signed-off-by: Steve Simpson --- docs/configuration/config-file-reference.md | 4 ++++ pkg/compactor/blocks_cleaner.go | 4 ++-- pkg/compactor/compactor.go | 12 +++++++++--- pkg/util/validation/limits.go | 10 ++++++++++ 4 files changed, 25 insertions(+), 5 deletions(-) diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 4bee82e6727..21c3629691a 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3752,6 +3752,10 @@ 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 data which is older than the specified retention period. 0 to disable. +# CLI flag: -compactor.retention-period +[compactor_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..0b4344f693c 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -35,7 +35,7 @@ type BlocksCleaner struct { services.Service cfg BlocksCleanerConfig - cfgProvider bucket.TenantConfigProvider + cfgProvider ConfigProvider logger log.Logger bucketClient objstore.Bucket usersScanner *cortex_tsdb.UsersScanner @@ -56,7 +56,7 @@ type BlocksCleaner struct { 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, diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 7141dd6ede4..9471b362cd0 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -155,13 +155,19 @@ func (cfg *Config) Validate() error { return nil } +// ConfigProvider defines the per-tenant config provider for the Compactor. +type ConfigProvider interface { + bucket.TenantConfigProvider + CompactorRetentionPeriod(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 +220,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 +246,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), diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 47f599559e2..5568da97062 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. + CompactorRetentionPeriod time.Duration `yaml:"compactor_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.CompactorRetentionPeriod, "compactor.retention-period", 0, "Delete data which is 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 } +// CompactorRetentionPeriod returns the retention period for a given user. +func (o *Overrides) CompactorRetentionPeriod(userID string) time.Duration { + return o.getOverridesForUser(userID).CompactorRetentionPeriod +} + // MetricRelabelConfigs returns the metric relabel configs for a given user. func (o *Overrides) MetricRelabelConfigs(userID string) []*relabel.Config { return o.getOverridesForUser(userID).MetricRelabelConfigs From a6fcca02a34f0a652c0d4ec8bc5a5c935f67efef Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Thu, 25 Feb 2021 12:04:46 +0100 Subject: [PATCH 2/9] Extend 'blocks_marked_for_deletion_total' metric with 'reason' label. Signed-off-by: Steve Simpson --- pkg/compactor/blocks_cleaner.go | 6 ++++++ pkg/compactor/compactor.go | 5 +++-- pkg/compactor/compactor_test.go | 15 ++++++++++----- 3 files changed, 19 insertions(+), 7 deletions(-) diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 0b4344f693c..96858712e52 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -50,6 +50,7 @@ type BlocksCleaner struct { runsLastSuccess prometheus.Gauge blocksCleanedTotal prometheus.Counter blocksFailedTotal prometheus.Counter + blocksMarkedForDeletion prometheus.Counter tenantBlocks *prometheus.GaugeVec tenantMarkedBlocks *prometheus.GaugeVec tenantPartialBlocks *prometheus.GaugeVec @@ -87,6 +88,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: "cortex_compactor_blocks_marked_for_deletion_total", + Help: "Total number of blocks marked for deletion in compactor.", + 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 diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 9471b362cd0..0c226277c1b 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -298,8 +298,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: "cortex_compactor_blocks_marked_for_deletion_total", + Help: "Total number of blocks marked for deletion in compactor.", + 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. From 65f66d7024879645e8fb74d1fc9a5eaac075589a Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Wed, 24 Feb 2021 08:55:26 +0100 Subject: [PATCH 3/9] Implement application of per-tenant retention period in Compactor. Signed-off-by: Steve Simpson --- pkg/compactor/blocks_cleaner.go | 66 +++++++ pkg/compactor/blocks_cleaner_test.go | 266 ++++++++++++++++++++++++++- 2 files changed, 328 insertions(+), 4 deletions(-) diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 96858712e52..f99e44e3db8 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" @@ -316,6 +317,16 @@ 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 { + if err := c.applyUserRetentionPeriod(ctx, idx, userID, userBucket, userLogger); err != nil { + level.Warn(userLogger).Log("msg", "failed to apply user retention period ", "err", err) + } + } + // 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) @@ -397,3 +408,58 @@ 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, userID string, userBucket *bucket.UserBucketClient, userLogger log.Logger) error { + retention := c.cfgProvider.CompactorRetentionPeriod(userID) + now := time.Now() + failed := 0 + + level.Info(userLogger).Log("msg", "applying retention", "now", now.String(), "retention", retention.String()) + blocks := listBlocksOutsideRetentionPeriod(idx, now, 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", "applying retention: marking block for deletion", "id", 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", "block", b.ID, "err", err) + failed++ + } + } + + if failed > 0 { + return errors.Errorf("failed to mark %d blocks for deletion", failed) + } + + return nil +} + +// listBlocksOutsideRetentionPeriod determines the blocks which have aged past +// the specified retention period, and are not already marked for deletion. +func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, now time.Time, retention time.Duration) (result bucketindex.Blocks) { + // The retention period of zero is a special value indicating to never delete. + if retention <= 0 { + return + } + + // 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{}) + for _, d := range idx.BlockDeletionMarks { + marked[d.ID] = struct{}{} + } + + for _, b := range idx.Blocks { + maxTime := time.Unix(b.MaxTime/1000, 0) + if now.After(maxTime.Add(retention)) { + _, isMarked := marked[b.ID] + if !isMarked { + result = append(result, b) + } + } + } + + return +} diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index 1d431b4081b..c730bc4da6f 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,231 @@ 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()) + + // Disabled (zero retention period) + result := listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 0) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + // Excessive retention period (wrapping epoch) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), time.Hour) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + // Normal operation - varying retention period. + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 4*time.Second) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 3*time.Second) + assert.ElementsMatch(t, []ulid.ULID{id1}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + assert.ElementsMatch(t, []ulid.ULID{id1, id2}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 1*time.Second) + 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(10, 0), 3*time.Second) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + assert.ElementsMatch(t, []ulid.ULID{id2}, result.GetULIDs()) + + idx.BlockDeletionMarks = bucketindex.BlockDeletionMarks{mark1, mark2} + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 3*time.Second) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) + + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 1*time.Second) + 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) + + assertBlocksExist := func(exists1, exists2, exists3, exists4 bool) { + exists, err := bucketClient.Exists(ctx, path.Join("user-1", block1.String(), metadata.MetaFilename)) + require.NoError(t, err) + assert.Equal(t, exists1, exists) + + exists, err = bucketClient.Exists(ctx, path.Join("user-1", block2.String(), metadata.MetaFilename)) + require.NoError(t, err) + assert.Equal(t, exists2, exists) + + exists, err = bucketClient.Exists(ctx, path.Join("user-2", block3.String(), metadata.MetaFilename)) + require.NoError(t, err) + assert.Equal(t, exists3, exists) + + exists, err = bucketClient.Exists(ctx, path.Join("user-2", block4.String(), metadata.MetaFilename)) + require.NoError(t, err) + assert.Equal(t, exists4, exists) + } + + require.NoError(t, cleaner.cleanUsers(ctx, true)) + assertBlocksExist(true, true, true, true) + + // Existing behaviour - retention period disabled. + + cfgProvider.UserRetentionPeriods["user-1"] = 0 + cfgProvider.UserRetentionPeriods["user-2"] = 0 + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlocksExist(true, true, true, 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)) + assertBlocksExist(true, true, true, 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)) + assertBlocksExist(true, true, true, true) + + // Marking the block again, before the deletion occurs, should not cause an error. + // This has the side-effect of including our mark in the metrics output. + + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlocksExist(true, true, true, 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", + )) + + // Reduce the deletion delay. Now the block will be deleted. + + cleaner.cfg.DeletionDelay = 0 + require.NoError(t, cleaner.cleanUsers(ctx, false)) + assertBlocksExist(false, true, true, 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)) + assertBlocksExist(false, true, false, 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 +647,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) CompactorRetentionPeriod(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 "" +} From 88ef36233910e588e9e6717e3a0118e9fc7cbbd4 Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Thu, 25 Feb 2021 13:49:12 +0100 Subject: [PATCH 4/9] Update Changelog. Signed-off-by: Steve Simpson --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e07249748f..410ec70e1d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ * `-alertmanager.alertmanager-client.tls-ca-path` * `-alertmanager.alertmanager-client.tls-server-name` * `-alertmanager.alertmanager-client.tls-insecure-skip-verify` +* [FEATURE] Compactor: implement per-tenant retention periods by performing block deletion in the Compactor. This is configured via `-compactor.retention-period`, and can be overriden 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. From 51db46d7f80031fe3e54ed339c7121a7a1fde57a Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Fri, 26 Feb 2021 09:09:29 +0100 Subject: [PATCH 5/9] Address review comments. Signed-off-by: Steve Simpson --- CHANGELOG.md | 3 +- docs/configuration/config-file-reference.md | 5 ++-- pkg/compactor/blocks_cleaner.go | 32 ++++++++------------- pkg/compactor/blocks_cleaner_test.go | 16 +++++------ pkg/compactor/compactor.go | 9 ++++-- pkg/util/validation/limits.go | 4 +-- 6 files changed, 34 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 410ec70e1d3..e4922d41ede 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. The existing total is labelled as `compaction`. Deletion marks due to 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,7 +27,7 @@ * `-alertmanager.alertmanager-client.tls-ca-path` * `-alertmanager.alertmanager-client.tls-server-name` * `-alertmanager.alertmanager-client.tls-insecure-skip-verify` -* [FEATURE] Compactor: implement per-tenant retention periods by performing block deletion in the Compactor. This is configured via `-compactor.retention-period`, and can be overriden on a per-tenant basis. #3879 +* [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 21c3629691a..74ce893fa22 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3752,9 +3752,10 @@ 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 data which is older than the specified retention period. 0 to disable. +# Delete blocks containing samples older than the specified retention period. 0 +# to disable. # CLI flag: -compactor.retention-period -[compactor_retention_period: | default = 0s] +[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 diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index f99e44e3db8..aaa53479154 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -90,8 +90,8 @@ func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, use Help: "Total number of blocks failed to be deleted.", }), blocksMarkedForDeletion: promauto.With(reg).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": "retention"}, }), @@ -322,9 +322,9 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b // The trade-off being that retention is not applied if the index has to be // built, but this is rare. if idx != nil { - if err := c.applyUserRetentionPeriod(ctx, idx, userID, userBucket, userLogger); err != nil { - level.Warn(userLogger).Log("msg", "failed to apply user retention period ", "err", err) - } + // We do not want to stop the remaining work in the cleaner if an + // error occurs here. Errors are logged in the function. + c.applyUserRetentionPeriod(ctx, idx, userID, userBucket, userLogger) } // Generate an updated in-memory version of the bucket index. @@ -410,29 +410,21 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map } // applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period. -func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, userID string, userBucket *bucket.UserBucketClient, userLogger log.Logger) error { +func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, userID string, userBucket *bucket.UserBucketClient, userLogger log.Logger) { retention := c.cfgProvider.CompactorRetentionPeriod(userID) now := time.Now() - failed := 0 - level.Info(userLogger).Log("msg", "applying retention", "now", now.String(), "retention", retention.String()) + level.Debug(userLogger).Log("msg", "applying retention", "retention", retention.String()) blocks := listBlocksOutsideRetentionPeriod(idx, now, 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", "applying retention: marking block for deletion", "id", b.ID, "maxTime", b.MaxTime) + 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", "block", b.ID, "err", err) - failed++ } } - - if failed > 0 { - return errors.Errorf("failed to mark %d blocks for deletion", failed) - } - - return nil } // listBlocksOutsideRetentionPeriod determines the blocks which have aged past @@ -446,16 +438,16 @@ func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, now time.Time, ret // 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{}) + marked := make(map[ulid.ULID]struct{}, len(idx.BlockDeletionMarks)) for _, d := range idx.BlockDeletionMarks { marked[d.ID] = struct{}{} } + threshold := now.Add(-retention) for _, b := range idx.Blocks { maxTime := time.Unix(b.MaxTime/1000, 0) - if now.After(maxTime.Add(retention)) { - _, isMarked := marked[b.ID] - if !isMarked { + if maxTime.Before(threshold) { + if _, isMarked := marked[b.ID]; !isMarked { result = append(result, b) } } diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index c730bc4da6f..2a2716c51ba 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -522,8 +522,8 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { // Existing behaviour - retention period disabled. - cfgProvider.UserRetentionPeriods["user-1"] = 0 - cfgProvider.UserRetentionPeriods["user-2"] = 0 + cfgProvider.userRetentionPeriods["user-1"] = 0 + cfgProvider.userRetentionPeriods["user-2"] = 0 require.NoError(t, cleaner.cleanUsers(ctx, false)) assertBlocksExist(true, true, true, true) @@ -548,7 +548,7 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { // Retention enabled only for a single user, but does nothing. - cfgProvider.UserRetentionPeriods["user-1"] = 9 * time.Hour + cfgProvider.userRetentionPeriods["user-1"] = 9 * time.Hour require.NoError(t, cleaner.cleanUsers(ctx, false)) assertBlocksExist(true, true, true, true) @@ -556,7 +556,7 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { // 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 + cfgProvider.userRetentionPeriods["user-1"] = 7 * time.Hour require.NoError(t, cleaner.cleanUsers(ctx, false)) assertBlocksExist(true, true, true, true) @@ -611,7 +611,7 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { // Retention enabled for other user; test deleting multiple blocks. - cfgProvider.UserRetentionPeriods["user-2"] = 5 * time.Hour + cfgProvider.userRetentionPeriods["user-2"] = 5 * time.Hour require.NoError(t, cleaner.cleanUsers(ctx, false)) assertBlocksExist(false, true, false, false) @@ -649,17 +649,17 @@ func (m *mockBucketFailure) Delete(ctx context.Context, name string) error { } type mockConfigProvider struct { - UserRetentionPeriods map[string]time.Duration + userRetentionPeriods map[string]time.Duration } func newMockConfigProvider() *mockConfigProvider { return &mockConfigProvider{ - UserRetentionPeriods: make(map[string]time.Duration), + userRetentionPeriods: make(map[string]time.Duration), } } func (m *mockConfigProvider) CompactorRetentionPeriod(user string) time.Duration { - if result, ok := m.UserRetentionPeriods[user]; ok { + if result, ok := m.userRetentionPeriods[user]; ok { return result } return 0 diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 0c226277c1b..a59d429cffd 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) @@ -298,8 +303,8 @@ 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{ diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 5568da97062..6590243b12e 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -87,7 +87,7 @@ type Limits struct { StoreGatewayTenantShardSize int `yaml:"store_gateway_tenant_shard_size" json:"store_gateway_tenant_shard_size"` // Compactor. - CompactorRetentionPeriod time.Duration `yaml:"compactor_retention_period"` + CompactorRetentionPeriod time.Duration `yaml:"compactor_blocks_retention_period"` // This config doesn't have a CLI flag registered here because they're registered in // their own original config struct. @@ -147,7 +147,7 @@ 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.CompactorRetentionPeriod, "compactor.retention-period", 0, "Delete data which is older than the specified retention period. 0 to disable.") + f.DurationVar(&l.CompactorRetentionPeriod, "compactor.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]") From edcfac530de86bf29f7ccb5dff1577d02ad47ddc Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Fri, 26 Feb 2021 10:04:39 +0100 Subject: [PATCH 6/9] Review comments (unit test). Signed-off-by: Steve Simpson --- pkg/compactor/blocks_cleaner_test.go | 247 ++++++++++++++------------- 1 file changed, 128 insertions(+), 119 deletions(-) diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index 2a2716c51ba..751a654513f 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -499,140 +499,149 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) { cleaner := NewBlocksCleaner(cfg, bucketClient, scanner, cfgProvider, logger, reg) - assertBlocksExist := func(exists1, exists2, exists3, exists4 bool) { - exists, err := bucketClient.Exists(ctx, path.Join("user-1", block1.String(), metadata.MetaFilename)) + 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, exists1, exists) - - exists, err = bucketClient.Exists(ctx, path.Join("user-1", block2.String(), metadata.MetaFilename)) - require.NoError(t, err) - assert.Equal(t, exists2, exists) - - exists, err = bucketClient.Exists(ctx, path.Join("user-2", block3.String(), metadata.MetaFilename)) - require.NoError(t, err) - assert.Equal(t, exists3, exists) - - exists, err = bucketClient.Exists(ctx, path.Join("user-2", block4.String(), metadata.MetaFilename)) - require.NoError(t, err) - assert.Equal(t, exists4, exists) + assert.Equal(t, expectExists, exists) } - require.NoError(t, cleaner.cleanUsers(ctx, true)) - assertBlocksExist(true, true, true, true) - // Existing behaviour - retention period disabled. - - cfgProvider.userRetentionPeriods["user-1"] = 0 - cfgProvider.userRetentionPeriods["user-2"] = 0 - - require.NoError(t, cleaner.cleanUsers(ctx, false)) - assertBlocksExist(true, true, true, 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", - )) + { + 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)) - assertBlocksExist(true, true, true, true) + { + 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)) - assertBlocksExist(true, true, true, true) + { + 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. - // This has the side-effect of including our mark in the metrics output. - - require.NoError(t, cleaner.cleanUsers(ctx, false)) - assertBlocksExist(true, true, true, 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", - )) + { + 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)) - assertBlocksExist(false, true, true, 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", - )) + { + 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)) - assertBlocksExist(false, true, false, 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", - )) + { + 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 { From b0a8267db6c2c8e53631fc93bd3e5964b2039811 Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Fri, 26 Feb 2021 10:36:37 +0100 Subject: [PATCH 7/9] Add missing json annotation. Signed-off-by: Steve Simpson --- pkg/util/validation/limits.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 6590243b12e..237a7f450a5 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -87,7 +87,7 @@ type Limits struct { StoreGatewayTenantShardSize int `yaml:"store_gateway_tenant_shard_size" json:"store_gateway_tenant_shard_size"` // Compactor. - CompactorRetentionPeriod time.Duration `yaml:"compactor_blocks_retention_period"` + CompactorRetentionPeriod 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. From da2eb99bd8da43146f277ee33da640df2668032e Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Fri, 26 Feb 2021 14:15:44 +0100 Subject: [PATCH 8/9] Review comments. Signed-off-by: Steve Simpson --- CHANGELOG.md | 2 +- docs/configuration/config-file-reference.md | 2 +- pkg/compactor/blocks_cleaner.go | 17 ++++++++--------- pkg/compactor/blocks_cleaner_test.go | 8 ++------ pkg/compactor/compactor.go | 2 +- pkg/util/validation/limits.go | 10 +++++----- 6 files changed, 18 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4922d41ede..33ff49674f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +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. The existing total is labelled as `compaction`. Deletion marks due to blocks passing the retention period are labelled as `retention`. #3879 +* [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` diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 74ce893fa22..6f4b34ad3cf 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3754,7 +3754,7 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # Delete blocks containing samples older than the specified retention period. 0 # to disable. -# CLI flag: -compactor.retention-period +# CLI flag: -compactor.blocks-retention-period [compactor_blocks_retention_period: | default = 0s] # S3 server-side encryption type. Required to enable server-side encryption diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index aaa53479154..967ae79ccc7 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -411,18 +411,22 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map // applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period. func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, userID string, userBucket *bucket.UserBucketClient, userLogger log.Logger) { - retention := c.cfgProvider.CompactorRetentionPeriod(userID) - now := time.Now() + retention := c.cfgProvider.CompactorBlocksRetentionPeriod(userID) + + // 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, now, retention) + blocks := listBlocksOutsideRetentionPeriod(idx, time.Now(), 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", "block", b.ID, "err", err) + level.Warn(userLogger).Log("msg", "failed to mark block for deletion", "block", b.ID, "err", err) } } } @@ -430,11 +434,6 @@ func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucke // listBlocksOutsideRetentionPeriod determines the blocks which have aged past // the specified retention period, and are not already marked for deletion. func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, now time.Time, retention time.Duration) (result bucketindex.Blocks) { - // The retention period of zero is a special value indicating to never delete. - if retention <= 0 { - return - } - // 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. diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index 751a654513f..e438a54dcfa 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -426,12 +426,8 @@ func TestBlocksCleaner_ListBlocksOutsideRetentionPeriod(t *testing.T) { assert.ElementsMatch(t, []ulid.ULID{id1, id2, id3}, idx.Blocks.GetULIDs()) - // Disabled (zero retention period) - result := listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 0) - assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) - // Excessive retention period (wrapping epoch) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), time.Hour) + result := listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), time.Hour) assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) // Normal operation - varying retention period. @@ -667,7 +663,7 @@ func newMockConfigProvider() *mockConfigProvider { } } -func (m *mockConfigProvider) CompactorRetentionPeriod(user string) time.Duration { +func (m *mockConfigProvider) CompactorBlocksRetentionPeriod(user string) time.Duration { if result, ok := m.userRetentionPeriods[user]; ok { return result } diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index a59d429cffd..efa4ee92f79 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -163,7 +163,7 @@ func (cfg *Config) Validate() error { // ConfigProvider defines the per-tenant config provider for the Compactor. type ConfigProvider interface { bucket.TenantConfigProvider - CompactorRetentionPeriod(user string) time.Duration + CompactorBlocksRetentionPeriod(user string) time.Duration } // Compactor is a multi-tenant TSDB blocks compactor based on Thanos. diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 237a7f450a5..be8a08fb56c 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -87,7 +87,7 @@ type Limits struct { StoreGatewayTenantShardSize int `yaml:"store_gateway_tenant_shard_size" json:"store_gateway_tenant_shard_size"` // Compactor. - CompactorRetentionPeriod time.Duration `yaml:"compactor_blocks_retention_period" json:"compactor_blocks_retention_period"` + 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. @@ -147,7 +147,7 @@ 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.CompactorRetentionPeriod, "compactor.retention-period", 0, "Delete blocks containing samples older than the specified retention period. 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]") @@ -420,9 +420,9 @@ func (o *Overrides) EvaluationDelay(userID string) time.Duration { return o.getOverridesForUser(userID).RulerEvaluationDelay } -// CompactorRetentionPeriod returns the retention period for a given user. -func (o *Overrides) CompactorRetentionPeriod(userID string) time.Duration { - return o.getOverridesForUser(userID).CompactorRetentionPeriod +// 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. From 05a00eaa37c7c06c33074cb8d9345db1e31ad56c Mon Sep 17 00:00:00 2001 From: Steve Simpson Date: Mon, 1 Mar 2021 14:55:20 +0100 Subject: [PATCH 9/9] Review comments. Signed-off-by: Steve Simpson --- pkg/compactor/blocks_cleaner.go | 12 +++++------- pkg/compactor/blocks_cleaner_test.go | 20 ++++++++++---------- 2 files changed, 15 insertions(+), 17 deletions(-) diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 967ae79ccc7..f6599e8190d 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -324,7 +324,8 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b 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. - c.applyUserRetentionPeriod(ctx, idx, userID, userBucket, userLogger) + retention := c.cfgProvider.CompactorBlocksRetentionPeriod(userID) + c.applyUserRetentionPeriod(ctx, idx, retention, userBucket, userLogger) } // Generate an updated in-memory version of the bucket index. @@ -410,16 +411,14 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map } // applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period. -func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, userID string, userBucket *bucket.UserBucketClient, userLogger log.Logger) { - retention := c.cfgProvider.CompactorBlocksRetentionPeriod(userID) - +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(), retention) + 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. @@ -433,7 +432,7 @@ func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucke // listBlocksOutsideRetentionPeriod determines the blocks which have aged past // the specified retention period, and are not already marked for deletion. -func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, now time.Time, retention time.Duration) (result bucketindex.Blocks) { +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. @@ -442,7 +441,6 @@ func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, now time.Time, ret marked[d.ID] = struct{}{} } - threshold := now.Add(-retention) for _, b := range idx.Blocks { maxTime := time.Unix(b.MaxTime/1000, 0) if maxTime.Before(threshold) { diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index e438a54dcfa..7576c6a6d80 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -427,20 +427,20 @@ func TestBlocksCleaner_ListBlocksOutsideRetentionPeriod(t *testing.T) { assert.ElementsMatch(t, []ulid.ULID{id1, id2, id3}, idx.Blocks.GetULIDs()) // Excessive retention period (wrapping epoch) - result := listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), time.Hour) + 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(10, 0), 4*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(6, 0)) assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 3*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) assert.ElementsMatch(t, []ulid.ULID{id1}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(8, 0)) assert.ElementsMatch(t, []ulid.ULID{id1, id2}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 1*time.Second) + 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. @@ -450,21 +450,21 @@ func TestBlocksCleaner_ListBlocksOutsideRetentionPeriod(t *testing.T) { idx.BlockDeletionMarks = bucketindex.BlockDeletionMarks{mark1} - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 3*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + 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(10, 0), 3*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(7, 0)) assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 2*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(8, 0)) assert.ElementsMatch(t, []ulid.ULID{}, result.GetULIDs()) - result = listBlocksOutsideRetentionPeriod(idx, time.Unix(10, 0), 1*time.Second) + result = listBlocksOutsideRetentionPeriod(idx, time.Unix(9, 0)) assert.ElementsMatch(t, []ulid.ULID{id3}, result.GetULIDs()) }