diff --git a/cmd/mimir/config-descriptor.json b/cmd/mimir/config-descriptor.json index 9c417cf5544..09e4a76616d 100644 --- a/cmd/mimir/config-descriptor.json +++ b/cmd/mimir/config-descriptor.json @@ -13349,6 +13349,17 @@ "fieldType": "list of durations", "fieldCategory": "advanced" }, + { + "kind": "field", + "name": "skip_elapsed_intermediate_block_ranges", + "required": false, + "desc": "When enabled, the compactor merges blocks directly into a larger compaction range rather than compacting into an intermediate range when the time period covered has elapsed.", + "fieldValue": null, + "fieldDefaultValue": false, + "fieldFlag": "compactor.skip-elapsed-intermediate-block-ranges", + "fieldType": "boolean", + "fieldCategory": "experimental" + }, { "kind": "field", "name": "block_sync_concurrency", diff --git a/cmd/mimir/help-all.txt.tmpl b/cmd/mimir/help-all.txt.tmpl index 93489a59abb..c4004506cc5 100644 --- a/cmd/mimir/help-all.txt.tmpl +++ b/cmd/mimir/help-all.txt.tmpl @@ -1401,6 +1401,8 @@ Usage of ./cmd/mimir/mimir: [experimental] Maximum backoff time for compaction executor retries when sending scheduler status updates. (default 32s) -compactor.scheduler-client.update-min-backoff duration [experimental] Minimum backoff time for compaction executor retries when sending scheduler status updates. (default 1s) + -compactor.skip-elapsed-intermediate-block-ranges + [experimental] When enabled, the compactor merges blocks directly into a larger compaction range rather than compacting into an intermediate range when the time period covered has elapsed. -compactor.split-and-merge-shards int The number of shards to use when splitting blocks. 0 to disable splitting. Values greater than 1 are rounded up to the next power of two. -compactor.split-groups int diff --git a/docs/sources/mimir/configure/about-versioning.md b/docs/sources/mimir/configure/about-versioning.md index b3d45ebe092..74d73fd7652 100644 --- a/docs/sources/mimir/configure/about-versioning.md +++ b/docs/sources/mimir/configure/about-versioning.md @@ -68,6 +68,8 @@ The following features are currently experimental: - `-compactor.max-lookback` - Limit how many block indexes are validated for health concurrently during a compaction job. - `-compactor.block-health-validation-concurrency` + - Merge blocks directly into a larger compaction range rather than compacting into an intermediate range when the time period covered has elapsed. + - `-compactor.skip-elapsed-intermediate-block-ranges` - Compactor scheduler - Coordinate compactors through a shared job queue and expose additional metrics about pending and active compaction work. - `-compactor-scheduler.*` diff --git a/docs/sources/mimir/configure/configuration-parameters/index.md b/docs/sources/mimir/configure/configuration-parameters/index.md index a708c8c6f30..23b4784a173 100644 --- a/docs/sources/mimir/configure/configuration-parameters/index.md +++ b/docs/sources/mimir/configure/configuration-parameters/index.md @@ -6685,6 +6685,12 @@ The `compactor` block configures the compactor component. # CLI flag: -compactor.block-ranges [block_ranges: | default = 2h0m0s,12h0m0s,24h0m0s] +# (experimental) When enabled, the compactor merges blocks directly into a +# larger compaction range rather than compacting into an intermediate range when +# the time period covered has elapsed. +# CLI flag: -compactor.skip-elapsed-intermediate-block-ranges +[skip_elapsed_intermediate_block_ranges: | default = false] + # (advanced) Number of goroutines to use when downloading blocks for compaction # and uploading resulting blocks. # CLI flag: -compactor.block-sync-concurrency diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 41cd85e3cb6..f698b65c5dd 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -43,15 +43,16 @@ const ( ) type BlocksCleanerConfig struct { - DeletionDelay time.Duration - CleanupInterval time.Duration - CleanupConcurrency int - TenantCleanupDelay time.Duration // Delay before removing tenant deletion mark and "debug". - DeleteBlocksConcurrency int - GetDeletionMarkersConcurrency int - UpdateBlocksConcurrency int - CompactionBlockRanges mimir_tsdb.DurationList // Used for estimating compaction jobs. - EstimateCompactionJobs bool + DeletionDelay time.Duration + CleanupInterval time.Duration + CleanupConcurrency int + TenantCleanupDelay time.Duration // Delay before removing tenant deletion mark and "debug". + DeleteBlocksConcurrency int + GetDeletionMarkersConcurrency int + UpdateBlocksConcurrency int + CompactionBlockRanges mimir_tsdb.DurationList // Used for estimating compaction jobs. + SkipElapsedIntermediateBlockRanges bool // Used for estimating compaction jobs. + EstimateCompactionJobs bool } type BlocksCleaner struct { @@ -525,7 +526,7 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, userLogger c.tenantBucketIndexLastUpdate.WithLabelValues(userID).Set(float64(idx.UpdatedAt)) if c.cfg.EstimateCompactionJobs { - jobs, err := estimateCompactionJobsFromBucketIndex(ctx, userID, userBucket, idx, c.cfg.CompactionBlockRanges, c.cfgProvider) + jobs, err := estimateCompactionJobsFromBucketIndex(ctx, userID, userBucket, idx, c.cfg.CompactionBlockRanges, c.cfg.SkipElapsedIntermediateBlockRanges, c.cfgProvider) if err != nil { // When compactor is shutting down, we get context cancellation. There's no reason to report that as error. if !errors.Is(err, context.Canceled) { @@ -755,7 +756,7 @@ func (c *BlocksCleaner) stalePartialBlockLastModifiedTime(ctx context.Context, b return lastModified, err } -func estimateCompactionJobsFromBucketIndex(ctx context.Context, userID string, userBucket objstore.InstrumentedBucket, idx *bucketindex.Index, compactionBlockRanges mimir_tsdb.DurationList, cfgProvider ConfigProvider) ([]*Job, error) { +func estimateCompactionJobsFromBucketIndex(ctx context.Context, userID string, userBucket objstore.InstrumentedBucket, idx *bucketindex.Index, compactionBlockRanges mimir_tsdb.DurationList, skipElapsedIntermediateBlockRanges bool, cfgProvider ConfigProvider) ([]*Job, error) { metas := ConvertBucketIndexToMetasForCompactionJobPlanning(idx) // We need to pass this metric to MetadataFilters, but we don't need to report this value from BlocksCleaner. @@ -772,7 +773,7 @@ func estimateCompactionJobsFromBucketIndex(ctx context.Context, userID string, u } } - grouper := NewSplitAndMergeGrouper(userID, compactionBlockRanges.ToMilliseconds(), cfgProvider, log.NewNopLogger()) + grouper := NewSplitAndMergeGrouper(userID, compactionBlockRanges.ToMilliseconds(), skipElapsedIntermediateBlockRanges, cfgProvider, log.NewNopLogger()) jobs, err := grouper.Groups(metas) return jobs, err } diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index b34e5ce2a18..5fceb1c7687 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -1354,7 +1354,7 @@ func TestComputeCompactionJobs(t *testing.T) { index := &bucketindex.Index{Blocks: c.blocks} cfgProvider := newMockConfigProvider() cfgProvider.splitAndMergeShards[user] = 3 - jobs, err := estimateCompactionJobsFromBucketIndex(context.Background(), user, userBucket, index, cfg.CompactionBlockRanges, cfgProvider) + jobs, err := estimateCompactionJobsFromBucketIndex(context.Background(), user, userBucket, index, cfg.CompactionBlockRanges, false, cfgProvider) require.NoError(t, err) split, merge := computeSplitAndMergeJobs(jobs) require.Equal(t, c.expectedSplits, split) @@ -1432,7 +1432,7 @@ func TestComputeCompactionJobsWithOOOShards(t *testing.T) { cfgProvider := newMockConfigProvider() cfgProvider.splitAndMergeShards[user] = c.mergeShards cfgProvider.oooSplitAndMergeShards[user] = c.oooMergeShards - jobs, err := estimateCompactionJobsFromBucketIndex(context.Background(), user, userBucket, index, cfg.CompactionBlockRanges, cfgProvider) + jobs, err := estimateCompactionJobsFromBucketIndex(context.Background(), user, userBucket, index, cfg.CompactionBlockRanges, false, cfgProvider) require.NoError(t, err) var inOrderJob, oooJob *Job diff --git a/pkg/compactor/bucket_compactor_e2e_test.go b/pkg/compactor/bucket_compactor_e2e_test.go index 035e41b3b47..2ccbd766d2d 100644 --- a/pkg/compactor/bucket_compactor_e2e_test.go +++ b/pkg/compactor/bucket_compactor_e2e_test.go @@ -164,7 +164,7 @@ func TestSyncer_GarbageCollect_e2e(t *testing.T) { require.NoError(t, sy.GarbageCollect(ctx)) // Only the level 3 block, the last source block in both resolutions should be left. - grouper := NewSplitAndMergeGrouper("user-1", []int64{2 * time.Hour.Milliseconds()}, newMockConfigProvider(), log.NewNopLogger()) + grouper := NewSplitAndMergeGrouper("user-1", []int64{2 * time.Hour.Milliseconds()}, false, newMockConfigProvider(), log.NewNopLogger()) groups, err := grouper.Groups(sy.Metas()) require.NoError(t, err) @@ -240,7 +240,7 @@ func TestGroupCompactE2E(t *testing.T) { require.NoError(t, err) planner := NewSplitAndMergePlanner([]int64{1000, 3000}) - grouper := NewSplitAndMergeGrouper("user-1", []int64{1000, 3000}, newMockConfigProvider(), logger) + grouper := NewSplitAndMergeGrouper("user-1", []int64{1000, 3000}, false, newMockConfigProvider(), logger) metrics := NewBucketCompactorMetrics(blocksMarkedForDeletion, prometheus.NewPedanticRegistry()) cfg := indexheader.Config{VerifyOnLoad: true} bComp, err := NewBucketCompactor( diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 95e02a051a3..6cb875b8abe 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -96,22 +96,23 @@ type BlocksCompactorFactory func( // Config holds the MultitenantCompactor config. type Config struct { - BlockRanges mimir_tsdb.DurationList `yaml:"block_ranges" category:"advanced"` - BlockSyncConcurrency int `yaml:"block_sync_concurrency" category:"advanced"` - BlockHealthValidationConcurrency int `yaml:"block_health_validation_concurrency" category:"experimental"` - MetaSyncConcurrency int `yaml:"meta_sync_concurrency" category:"advanced"` - DataDir string `yaml:"data_dir"` - CompactionInterval time.Duration `yaml:"compaction_interval" category:"advanced"` - CompactionRetries int `yaml:"compaction_retries" category:"advanced"` - CompactionConcurrency int `yaml:"compaction_concurrency" category:"advanced"` - CompactionWaitPeriod time.Duration `yaml:"first_level_compaction_wait_period"` - CompactionOOOWaitPeriod time.Duration `yaml:"first_level_compaction_ooo_wait_period"` - CompactionSkipFutureMaxTime bool `yaml:"first_level_compaction_skip_future_max_time"` - CleanupInterval time.Duration `yaml:"cleanup_interval" category:"advanced"` - CleanupConcurrency int `yaml:"cleanup_concurrency" category:"advanced"` - DeletionDelay time.Duration `yaml:"deletion_delay" category:"advanced"` - TenantCleanupDelay time.Duration `yaml:"tenant_cleanup_delay" category:"advanced"` - MaxCompactionTime time.Duration `yaml:"max_compaction_time" category:"advanced"` + BlockRanges mimir_tsdb.DurationList `yaml:"block_ranges" category:"advanced"` + SkipElapsedIntermediateBlockRanges bool `yaml:"skip_elapsed_intermediate_block_ranges" category:"experimental"` + BlockSyncConcurrency int `yaml:"block_sync_concurrency" category:"advanced"` + BlockHealthValidationConcurrency int `yaml:"block_health_validation_concurrency" category:"experimental"` + MetaSyncConcurrency int `yaml:"meta_sync_concurrency" category:"advanced"` + DataDir string `yaml:"data_dir"` + CompactionInterval time.Duration `yaml:"compaction_interval" category:"advanced"` + CompactionRetries int `yaml:"compaction_retries" category:"advanced"` + CompactionConcurrency int `yaml:"compaction_concurrency" category:"advanced"` + CompactionWaitPeriod time.Duration `yaml:"first_level_compaction_wait_period"` + CompactionOOOWaitPeriod time.Duration `yaml:"first_level_compaction_ooo_wait_period"` + CompactionSkipFutureMaxTime bool `yaml:"first_level_compaction_skip_future_max_time"` + CleanupInterval time.Duration `yaml:"cleanup_interval" category:"advanced"` + CleanupConcurrency int `yaml:"cleanup_concurrency" category:"advanced"` + DeletionDelay time.Duration `yaml:"deletion_delay" category:"advanced"` + TenantCleanupDelay time.Duration `yaml:"tenant_cleanup_delay" category:"advanced"` + MaxCompactionTime time.Duration `yaml:"max_compaction_time" category:"advanced"` // Compactor concurrency options MaxOpeningBlocksConcurrency int `yaml:"max_opening_blocks_concurrency" category:"advanced"` // Number of goroutines opening blocks before compaction. @@ -163,6 +164,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet, logger log.Logger) { cfg.retryMaxBackoff = time.Minute f.Var(&cfg.BlockRanges, "compactor.block-ranges", "List of compaction time ranges.") + f.BoolVar(&cfg.SkipElapsedIntermediateBlockRanges, "compactor.skip-elapsed-intermediate-block-ranges", false, "When enabled, the compactor merges blocks directly into a larger compaction range rather than compacting into an intermediate range when the time period covered has elapsed.") f.IntVar(&cfg.BlockSyncConcurrency, "compactor.block-sync-concurrency", 8, "Number of goroutines to use when downloading blocks for compaction and uploading resulting blocks.") f.IntVar(&cfg.BlockHealthValidationConcurrency, "compactor.block-health-validation-concurrency", 0, "Number of blocks whose health can be validated concurrently during a compaction job. A nonpositive value means no limit.") f.IntVar(&cfg.MetaSyncConcurrency, "compactor.meta-sync-concurrency", 20, "Number of goroutines to use when syncing block meta files from the long term storage.") @@ -599,15 +601,16 @@ func (c *MultitenantCompactor) starting(ctx context.Context) error { // Create the blocks cleaner (service). c.blocksCleaner = NewBlocksCleaner(BlocksCleanerConfig{ - DeletionDelay: c.compactorCfg.DeletionDelay, - CleanupInterval: util.DurationWithJitter(c.compactorCfg.CleanupInterval, 0.1), - CleanupConcurrency: c.compactorCfg.CleanupConcurrency, - TenantCleanupDelay: c.compactorCfg.TenantCleanupDelay, - DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency, - GetDeletionMarkersConcurrency: defaultGetDeletionMarkersConcurrency, - UpdateBlocksConcurrency: c.compactorCfg.UpdateBlocksConcurrency, - CompactionBlockRanges: c.compactorCfg.BlockRanges, - EstimateCompactionJobs: !c.compactorCfg.SchedulerClientConfig.Enabled, + DeletionDelay: c.compactorCfg.DeletionDelay, + CleanupInterval: util.DurationWithJitter(c.compactorCfg.CleanupInterval, 0.1), + CleanupConcurrency: c.compactorCfg.CleanupConcurrency, + TenantCleanupDelay: c.compactorCfg.TenantCleanupDelay, + DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency, + GetDeletionMarkersConcurrency: defaultGetDeletionMarkersConcurrency, + UpdateBlocksConcurrency: c.compactorCfg.UpdateBlocksConcurrency, + CompactionBlockRanges: c.compactorCfg.BlockRanges, + SkipElapsedIntermediateBlockRanges: c.compactorCfg.SkipElapsedIntermediateBlockRanges, + EstimateCompactionJobs: !c.compactorCfg.SchedulerClientConfig.Enabled, }, c.bucketClient, c.shardingStrategy.blocksCleanerOwnsUser, c.cfgProvider, c.parentLogger, c.registerer) // Start blocks cleaner asynchronously, don't wait until initial cleanup is finished. diff --git a/pkg/compactor/planned_jobs_http.go b/pkg/compactor/planned_jobs_http.go index 4da79d2a208..f6f5e02a99a 100644 --- a/pkg/compactor/planned_jobs_http.go +++ b/pkg/compactor/planned_jobs_http.go @@ -133,7 +133,7 @@ func (c *MultitenantCompactor) PlannedJobsHandler(w http.ResponseWriter, req *ht oooMergeShards: oooMergeShards, splitGroups: splitGroups, } - jobs, err := estimateCompactionJobsFromBucketIndex(req.Context(), tenantID, bucket.NewUserBucketClient(tenantID, c.bucketClient, c.cfgProvider), idx, c.compactorCfg.BlockRanges, cfgOverride) + jobs, err := estimateCompactionJobsFromBucketIndex(req.Context(), tenantID, bucket.NewUserBucketClient(tenantID, c.bucketClient, c.cfgProvider), idx, c.compactorCfg.BlockRanges, c.compactorCfg.SkipElapsedIntermediateBlockRanges, cfgOverride) if err != nil { level.Error(c.logger).Log("msg", "failed to compute compaction jobs from bucket index for tenant while listing compaction jobs", "user", tenantID, "err", err) util.WriteTextResponse(w, "Failed to compute compaction jobs from bucket index") diff --git a/pkg/compactor/split_merge_compactor.go b/pkg/compactor/split_merge_compactor.go index 351284497c3..f7f41811115 100644 --- a/pkg/compactor/split_merge_compactor.go +++ b/pkg/compactor/split_merge_compactor.go @@ -16,6 +16,7 @@ func splitAndMergeGrouperFactory(_ context.Context, cfg Config, cfgProvider Conf return NewSplitAndMergeGrouper( userID, cfg.BlockRanges.ToMilliseconds(), + cfg.SkipElapsedIntermediateBlockRanges, cfgProvider, logger) } diff --git a/pkg/compactor/split_merge_grouper.go b/pkg/compactor/split_merge_grouper.go index 4cd51a2b61f..e206cb264a7 100644 --- a/pkg/compactor/split_merge_grouper.go +++ b/pkg/compactor/split_merge_grouper.go @@ -21,24 +21,27 @@ import ( ) type SplitAndMergeGrouper struct { - userID string - ranges []int64 - cfgProvider ConfigProvider - logger log.Logger + userID string + ranges []int64 + skipElapsedIntermediateRanges bool + cfgProvider ConfigProvider + logger log.Logger } // NewSplitAndMergeGrouper makes a new SplitAndMergeGrouper. The provided ranges must be sorted. func NewSplitAndMergeGrouper( userID string, ranges []int64, + skipElapsedIntermediateRanges bool, cfgProvider ConfigProvider, logger log.Logger, ) *SplitAndMergeGrouper { return &SplitAndMergeGrouper{ - userID: userID, - ranges: ranges, - cfgProvider: cfgProvider, - logger: logger, + userID: userID, + ranges: ranges, + skipElapsedIntermediateRanges: skipElapsedIntermediateRanges, + cfgProvider: cfgProvider, + logger: logger, } } @@ -48,7 +51,7 @@ func (g *SplitAndMergeGrouper) Groups(blocks map[ulid.ULID]*block.Meta) (res []* flatBlocks = append(flatBlocks, b) } - for _, job := range planCompaction(g.userID, flatBlocks, g.ranges, g.cfgProvider) { + for _, job := range planCompaction(g.userID, flatBlocks, g.ranges, g.skipElapsedIntermediateRanges, g.cfgProvider) { jobShardCount := effectiveShardCount(job.blocks, g.userID, g.cfgProvider) // Sanity check: if splitting is disabled, we don't expect any job for the split stage. @@ -112,7 +115,7 @@ func effectiveShardCount(blocks []*block.Meta, userID string, cfgProvider Config // planCompaction analyzes the input blocks and returns a list of compaction jobs that can be // run concurrently. Each returned job may belong either to this compactor instance or another one // in the cluster, so the caller should check if they belong to their instance before running them. -func planCompaction(userID string, blocks []*block.Meta, ranges []int64, cfgProvider ConfigProvider) (jobs []*job) { +func planCompaction(userID string, blocks []*block.Meta, ranges []int64, skipElapsedIntermediateRanges bool, cfgProvider ConfigProvider) (jobs []*job) { if len(blocks) == 0 || len(ranges) == 0 { return nil } @@ -126,6 +129,8 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, cfgProv } splitGroups := uint32(cfgProvider.CompactorSplitGroups(userID)) + highestMaxTime := getMaxTime(blocks) + now := time.Now().UnixMilli() for _, mainBlocks := range mainGroups { // Sort blocks by min time. @@ -141,36 +146,42 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, cfgProv // We can plan a job only if it doesn't conflict with other jobs already planned. // Since we run the planning for each compaction range in increasing order, we guarantee // that a job for the current time range is planned only if there's no other job for the - // same shard ID and an overlapping smaller time range. - for _, j := range jobs { - if job.conflicts(j) { + // same shard ID and an overlapping smaller time range. The only exception is a job which + // can replace the single job it conflicts with to skip an elapsed intermediate range. + replaceIdx := -1 + + for idx, j := range jobs { + if !job.conflicts(j) { + continue + } + + // Give up on this job unless it can replace an elapsed intermediate range + if !skipElapsedIntermediateRanges || replaceIdx >= 0 || !canSkipElapsedIntermediateRange(job, j, ranges[0], highestMaxTime, now) { continue nextJob } + + replaceIdx = idx } - jobs = append(jobs, job) + // Replacing the job in place is fine, because jobs get sorted before being returned. + if replaceIdx >= 0 { + jobs[replaceIdx] = job + } else { + jobs = append(jobs, job) + } } } } // Ensure we don't compact the most recent blocks prematurely. We allow a job to remain if: - // - its range is before the most recent block - // - its range is at least 1 job length in the past + // - its range is in the past // - its max compaction level is 1 // - it fully covers the range - highestMaxTime := getMaxTime(blocks) - for idx := 0; idx < len(jobs); { job := jobs[idx] - // If the job covers a range before the most recent block, it's fine. - if job.rangeEnd <= highestMaxTime { - idx++ - continue - } - - // If the job covers a range at least 1 job length in the past, it's fine. - if job.rangeEnd+job.rangeLength() <= time.Now().UnixMilli() { + // If the job covers a range in the past, it's fine. + if isRangeInThePast(job, highestMaxTime, now) { idx++ continue } @@ -182,7 +193,7 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, cfgProv } // If the job covers the full range, it's fine. - if job.maxTime()-job.minTime() == job.rangeLength() { + if job.coversFullRange() { idx++ continue } @@ -209,6 +220,29 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, cfgProv return jobs } +// isRangeInThePast returns whether the job range is no longer the one in-order ingestion is filling, +// either because more recent blocks already exist or because the range ended at least one range +// length ago. +func isRangeInThePast(j *job, highestMaxTime, now int64) bool { + return j.rangeEnd <= highestMaxTime || j.rangeEnd+j.rangeLength() <= now +} + +// canSkipElapsedIntermediateRange returns whether the candidate job can replace the conflicting job for a +// smaller range, merging the larger range in a single pass instead of writing the intermediate block +// first. Ranges are planned in increasing order, so the candidate always covers the larger range. +func canSkipElapsedIntermediateRange(candidate, conflicting *job, smallestRange, highestMaxTime, now int64) bool { + // The smallest range is not an intermediate range. Letting it merge in isolation helps lower + // the block count faster. This also rules out split jobs, which only exist for that range. + if conflicting.rangeLength() <= smallestRange { + return false + } + + // Replace only once the candidate range is in the past, which guarantees the candidate isn't + // dropped later as a premature compaction. Dropping it would leave the blocks of the replaced + // job without any planned job. + return isRangeInThePast(candidate, highestMaxTime, now) +} + // planCompactionByRange analyzes the input blocks and returns a list of compaction jobs to // compact blocks for the given compaction time range. Input blocks MUST be sorted by MinTime. func planCompactionByRange(userID string, blocks []*block.Meta, tr int64, isSmallestRange bool, shardCount, splitGroups uint32) (jobs []*job) { diff --git a/pkg/compactor/split_merge_grouper_test.go b/pkg/compactor/split_merge_grouper_test.go index 24f228ea530..db744bdef94 100644 --- a/pkg/compactor/split_merge_grouper_test.go +++ b/pkg/compactor/split_merge_grouper_test.go @@ -33,12 +33,13 @@ func TestPlanCompaction(t *testing.T) { } tests := map[string]struct { - ranges []int64 - shardCount uint32 - oooShardCount uint32 - splitGroups uint32 - blocks []*block.Meta - expected []*job + ranges []int64 + shardCount uint32 + oooShardCount uint32 + splitGroups uint32 + skipElapsedIntermediateRanges bool + blocks []*block.Meta + expected []*job }{ "no input blocks": { ranges: []int64{20}, @@ -549,6 +550,80 @@ func TestPlanCompaction(t *testing.T) { }}, }, }, + "should skip the intermediate range when a single job for it stands in the way": { + ranges: []int64{10, 20, 40}, + shardCount: 1, + skipElapsedIntermediateRanges: true, + blocks: []*block.Meta{ + // Already compacted on the 2nd level range [0, 20]. + {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + // Still on the 1st level range, covering the rest of the 3rd level range [0, 40]. + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + expected: []*job{ + {userID: userID, stage: stageMerge, shardID: "1_of_1", blocksGroup: blocksGroup{ + rangeStart: 0, + rangeEnd: 40, + blocks: []*block.Meta{ + {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + }}, + }, + }, + "should NOT skip the intermediate range if disabled": { + ranges: []int64{10, 20, 40}, + shardCount: 1, + skipElapsedIntermediateRanges: false, + blocks: []*block.Meta{ + {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + expected: []*job{ + {userID: userID, stage: stageMerge, shardID: "1_of_1", blocksGroup: blocksGroup{ + rangeStart: 20, + rangeEnd: 40, + blocks: []*block.Meta{ + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + }}, + }, + }, + "should NOT skip the intermediate range if more than one job for it stands in the way": { + ranges: []int64{10, 20, 40}, + shardCount: 1, + skipElapsedIntermediateRanges: true, + blocks: []*block.Meta{ + // Both halves of the 3rd level range [0, 40] are still on the 1st level range, so + // two 2nd level jobs stand in the way. + {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 10}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 10, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block4, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + expected: []*job{ + {userID: userID, stage: stageMerge, shardID: "1_of_1", blocksGroup: blocksGroup{ + rangeStart: 0, + rangeEnd: 20, + blocks: []*block.Meta{ + {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 10}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 10, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + }}, + {userID: userID, stage: stageMerge, shardID: "1_of_1", blocksGroup: blocksGroup{ + rangeStart: 20, + rangeEnd: 40, + blocks: []*block.Meta{ + {BlockMeta: tsdb.BlockMeta{ULID: block3, MinTime: 20, MaxTime: 30}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + {BlockMeta: tsdb.BlockMeta{ULID: block4, MinTime: 30, MaxTime: 40}, Thanos: block.ThanosMeta{Labels: map[string]string{block.CompactorShardIDExternalLabel: "1_of_1"}}}, + }, + }}, + }, + }, } for testName, testData := range tests { @@ -557,7 +632,7 @@ func TestPlanCompaction(t *testing.T) { cfg.splitAndMergeShards[userID] = int(testData.shardCount) cfg.oooSplitAndMergeShards[userID] = int(testData.oooShardCount) cfg.splitGroups[userID] = int(testData.splitGroups) - actual := planCompaction(userID, testData.blocks, testData.ranges, cfg) + actual := planCompaction(userID, testData.blocks, testData.ranges, testData.skipElapsedIntermediateRanges, cfg) // Print the actual jobs (useful for debugging if tests fail). t.Logf("got %d jobs:", len(actual)) @@ -675,6 +750,57 @@ func TestPlanSplitting(t *testing.T) { } } +func TestCanSkipElapsedIntermediateRange(t *testing.T) { + var ( + twoHours = 2 * time.Hour.Milliseconds() + halfDay = 12 * time.Hour.Milliseconds() + day = 24 * time.Hour.Milliseconds() + ) + + // The candidate merges a whole day, replacing the job for the day's second half. + candidate := &job{stage: stageMerge, blocksGroup: blocksGroup{rangeStart: 0, rangeEnd: day}} + + tests := map[string]struct { + conflictingRangeStart int64 + highestMaxTime int64 + now int64 + expected bool + }{ + "should skip the intermediate range once blocks for a more recent range exist": { + conflictingRangeStart: halfDay, + highestMaxTime: day + twoHours, + now: day + twoHours, + expected: true, + }, + "should skip the intermediate range once the candidate range ended a range length ago": { + conflictingRangeStart: halfDay, + highestMaxTime: day - twoHours, + now: 2 * day, + expected: true, + }, + "should NOT skip the intermediate range while the candidate range is not in the past yet": { + conflictingRangeStart: halfDay, + highestMaxTime: day - twoHours, + now: day - twoHours, + expected: false, + }, + "should NOT skip the smallest range, which is never an intermediate one": { + conflictingRangeStart: day - twoHours, + highestMaxTime: day + twoHours, + now: day + twoHours, + expected: false, + }, + } + + for testName, testData := range tests { + t.Run(testName, func(t *testing.T) { + conflicting := &job{stage: stageMerge, blocksGroup: blocksGroup{rangeStart: testData.conflictingRangeStart, rangeEnd: day}} + + assert.Equal(t, testData.expected, canSkipElapsedIntermediateRange(candidate, conflicting, twoHours, testData.highestMaxTime, testData.now)) + }) + } +} + func TestGroupBlocksByShardID(t *testing.T) { block1 := ulid.MustNew(1, nil) block2 := ulid.MustNew(2, nil) @@ -903,7 +1029,7 @@ func TestSplitAndMergeGrouper_Groups_OOOShardCount(t *testing.T) { cfg := newMockConfigProvider() cfg.splitGroups[userID] = 1 - g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, cfg, log.NewNopLogger()) + g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, false, cfg, log.NewNopLogger()) jobs, err := g.Groups(map[ulid.ULID]*block.Meta{ block1: {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}}, block2: {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 0, MaxTime: 20}, Thanos: block.ThanosMeta{Labels: map[string]string{block.OutOfOrderExternalLabel: block.OutOfOrderExternalLabelValue}}}, @@ -924,7 +1050,7 @@ func TestSplitAndMergeGrouper_Groups_OOOShardCount(t *testing.T) { cfg := newMockConfigProvider() cfg.splitAndMergeShards[userID] = 8 cfg.splitGroups[userID] = 1 - g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, cfg, log.NewNopLogger()) + g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, false, cfg, log.NewNopLogger()) jobs, err := g.Groups(map[ulid.ULID]*block.Meta{ block1: {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}}, block2: {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 0, MaxTime: 20}}, @@ -950,7 +1076,7 @@ func TestSplitAndMergeGrouper_Groups_OOOShardCount(t *testing.T) { cfg.splitAndMergeShards[userID] = 8 cfg.oooSplitAndMergeShards[userID] = 2 cfg.splitGroups[userID] = 1 - g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, cfg, log.NewNopLogger()) + g := NewSplitAndMergeGrouper(userID, []int64{20, 40}, false, cfg, log.NewNopLogger()) jobs, err := g.Groups(map[ulid.ULID]*block.Meta{ block1: {BlockMeta: tsdb.BlockMeta{ULID: block1, MinTime: 0, MaxTime: 20}}, block2: {BlockMeta: tsdb.BlockMeta{ULID: block2, MinTime: 0, MaxTime: 20}}, diff --git a/pkg/compactor/split_merge_job.go b/pkg/compactor/split_merge_job.go index 3a6b6888eae..a80f55bac78 100644 --- a/pkg/compactor/split_merge_job.go +++ b/pkg/compactor/split_merge_job.go @@ -130,6 +130,11 @@ func (g blocksGroup) maxTime() int64 { return max } +// coversFullRange returns whether the blocks in the group span the entire group range. +func (g blocksGroup) coversFullRange() bool { + return g.maxTime()-g.minTime() == g.rangeLength() +} + // maxCompactionLevel returns the highest Compaction.Level across all blocks in the group. func (g blocksGroup) maxCompactionLevel() int { maxLevel := g.blocks[0].Compaction.Level diff --git a/tools/compaction-planner/main.go b/tools/compaction-planner/main.go index a7fc64af43e..ad546e476e6 100644 --- a/tools/compaction-planner/main.go +++ b/tools/compaction-planner/main.go @@ -32,13 +32,14 @@ func main() { flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError) cfg := struct { - bucket bucket.Config - userID string - blockRanges mimir_tsdb.DurationList - shardCount int - oooShardCount int - splitGroups int - sorting string + bucket bucket.Config + userID string + blockRanges mimir_tsdb.DurationList + skipElapsedIntermediateRanges bool + shardCount int + oooShardCount int + splitGroups int + sorting string }{} logger := gokitlog.NewNopLogger() @@ -47,6 +48,7 @@ func main() { cfg.bucket.RegisterFlags(flag.CommandLine) cfg.blockRanges = mimir_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour} flag.Var(&cfg.blockRanges, "block-ranges", "List of compaction time ranges.") + flag.BoolVar(&cfg.skipElapsedIntermediateRanges, "skip-elapsed-intermediate-block-ranges", false, "Merge blocks directly into a larger compaction range when a single compaction job for an intermediate range is the only one standing in the way.") flag.StringVar(&cfg.userID, "user", "", "User (tenant)") flag.IntVar(&cfg.shardCount, "shard-count", 4, "Shard count") flag.IntVar(&cfg.oooShardCount, "ooo-shard-count", 0, "Shard count for out-of-order blocks (0 to use shard-count)") @@ -103,7 +105,7 @@ func main() { oooShardCount: cfg.oooShardCount, splitGroupsCount: cfg.splitGroups, } - grouper := compactor.NewSplitAndMergeGrouper(cfg.userID, cfg.blockRanges.ToMilliseconds(), cfgProvider, logger) + grouper := compactor.NewSplitAndMergeGrouper(cfg.userID, cfg.blockRanges.ToMilliseconds(), cfg.skipElapsedIntermediateRanges, cfgProvider, logger) jobs, err := grouper.Groups(metas) if err != nil { log.Fatalln("failed to plan compaction:", err)