compactor: allow skipping intermediate range in compaction

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.*`

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
 }

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

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(

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.

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")

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)
 }