Make the compaction group graceful period configurable per block range

CHANGELOG.md

@@ -5,6 +5,7 @@
 * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828
 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873
 * [FEATURE] Storage: Add support for Alibaba Cloud OSS as a backend for blocks, ruler, and alertmanager storage. Configured via `-<prefix>.oss.*` flags with `backend: oss`. #7907
+* [ENHANCEMENT] Compactor: Add `-compactor.block-ranges-compaction-delay` to configure the compaction group graceful period per `-compactor.block-ranges` entry, matched by position. A larger block range can now be held back until its sub-ranges have settled, instead of every range sharing one period derived from `-compactor.cleanup-interval`. Leaving it empty, and any range without an entry, keeps the current behaviour of twice `-compactor.cleanup-interval`. The delay is a lower bound: groups only become eligible on a compaction run, so the effective wait can be up to `-compactor.compaction-interval` longer. Only used by the partitioning compaction strategy. #7916
 * [ENHANCEMENT] Ring: Add the `cortex_ring_member_zonal_ownership_percent{name, member, zone}` metric, reporting the fraction of its own zone's token space owned by each ring member. With zone-aware replication a write is routed within a single zone, so the existing `cortex_ring_member_ownership_percent`, which is computed over the merged token list of all zones, does not reflect the hash range a member actually serves. Reported when ring detailed metrics and zone awareness are both enabled. #7910
 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873

docs/blocks-storage/compactor.md

@@ -92,6 +92,17 @@ compactor:
   # CLI flag: -compactor.block-ranges
   [block_ranges: <list of duration> | default = 2h0m0s,12h0m0s,24h0m0s]
 
+  # Per compaction time range, the minimum age the end of a compaction group's
+  # time range must reach before the group becomes eligible for compaction.
+  # Positionally matched against -compactor.block-ranges: it must either be
+  # empty or have the same number of entries, and no entry may be negative. When
+  # empty, every range defaults to twice -compactor.cleanup-interval. This is a
+  # lower bound: a group only becomes eligible on a compaction run, so the
+  # effective wait can be up to -compactor.compaction-interval longer. Only used
+  # by the partitioning compaction strategy.
+  # CLI flag: -compactor.block-ranges-compaction-delay
+  [block_ranges_compaction_delay: <list of duration> | default = ]
+
   # Number of Go routines to use when syncing block index and chunks files from
   # the long term storage.
   # CLI flag: -compactor.block-sync-concurrency

docs/configuration/config-file-reference.md

@@ -3103,6 +3103,17 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.block-ranges
 [block_ranges: <list of duration> | default = 2h0m0s,12h0m0s,24h0m0s]
 
+# Per compaction time range, the minimum age the end of a compaction group's
+# time range must reach before the group becomes eligible for compaction.
+# Positionally matched against -compactor.block-ranges: it must either be empty
+# or have the same number of entries, and no entry may be negative. When empty,
+# every range defaults to twice -compactor.cleanup-interval. This is a lower
+# bound: a group only becomes eligible on a compaction run, so the effective
+# wait can be up to -compactor.compaction-interval longer. Only used by the
+# partitioning compaction strategy.
+# CLI flag: -compactor.block-ranges-compaction-delay
+[block_ranges_compaction_delay: <list of duration> | default = ]
+
 # Number of Go routines to use when syncing block index and chunks files from
 # the long term storage.
 # CLI flag: -compactor.block-sync-concurrency

pkg/compactor/compactor.go

@@ -52,8 +52,10 @@ const (
 )
 
 var (
-	errInvalidBlockRanges = "compactor block range periods should be divisible by the previous one, but %s is not divisible by %s"
-	RingOp                = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil)
+	errInvalidBlockRanges                   = "compactor block range periods should be divisible by the previous one, but %s is not divisible by %s"
+	errInvalidBlockRangesCompactionDelayLen = "compactor block ranges compaction delay must have the same number of entries as compactor block ranges, but got %d and %d"
+	errNegativeBlockRangesCompactionDelay   = "compactor block ranges compaction delay cannot be negative, but got %s at position %d"
+	RingOp                                  = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil)
 
 	supportedShardingStrategies              = []string{util.ShardingStrategyDefault, util.ShardingStrategyShuffle}
 	errInvalidShardingStrategy               = errors.New("invalid sharding strategy")
@@ -255,6 +257,7 @@ type Limits interface {
 // Config holds the Compactor config.
 type Config struct {
 	BlockRanges                           cortex_tsdb.DurationList `yaml:"block_ranges"`
+	BlockRangesCompactionDelay            cortex_tsdb.DurationList `yaml:"block_ranges_compaction_delay"`
 	BlockSyncConcurrency                  int                      `yaml:"block_sync_concurrency"`
 	MetaSyncConcurrency                   int                      `yaml:"meta_sync_concurrency"`
 	ConsistencyDelay                      time.Duration            `yaml:"consistency_delay"`
@@ -317,6 +320,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.retryMaxBackoff = time.Minute
 
 	f.Var(&cfg.BlockRanges, "compactor.block-ranges", "List of compaction time ranges.")
+	f.Var(&cfg.BlockRangesCompactionDelay, "compactor.block-ranges-compaction-delay", "Per compaction time range, the minimum age the end of a compaction group's time range must reach before the group becomes eligible for compaction. Positionally matched against -compactor.block-ranges: it must either be empty or have the same number of entries, and no entry may be negative. When empty, every range defaults to twice -compactor.cleanup-interval. This is a lower bound: a group only becomes eligible on a compaction run, so the effective wait can be up to -compactor.compaction-interval longer. Only used by the partitioning compaction strategy.")
 	f.DurationVar(&cfg.ConsistencyDelay, "compactor.consistency-delay", 0, fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.PartialUploadThresholdAge))
 	f.IntVar(&cfg.BlockSyncConcurrency, "compactor.block-sync-concurrency", 20, "Number of Go routines to use when syncing block index and chunks files from the long term storage.")
 	f.IntVar(&cfg.MetaSyncConcurrency, "compactor.meta-sync-concurrency", 20, "Number of Go routines to use when syncing block meta files from the long term storage.")
@@ -365,6 +369,19 @@ func (cfg *Config) Validate(limits validation.Limits) error {
 		}
 	}
 
+	// The compaction delay is matched to block ranges by position, so it must either be
+	// left empty (every range falls back to the default delay) or cover every range.
+	if len(cfg.BlockRangesCompactionDelay) > 0 {
+		if len(cfg.BlockRangesCompactionDelay) != len(cfg.BlockRanges) {
+			return errors.Errorf(errInvalidBlockRangesCompactionDelayLen, len(cfg.BlockRangesCompactionDelay), len(cfg.BlockRanges))
+		}
+		for i, delay := range cfg.BlockRangesCompactionDelay {
+			if delay < 0 {
+				return errors.Errorf(errNegativeBlockRangesCompactionDelay, delay.String(), i)
+			}
+		}
+	}
+
 	// Make sure a valid sharding strategy is being used
 	if !slices.Contains(supportedShardingStrategies, cfg.ShardingStrategy) {
 		return errInvalidShardingStrategy

pkg/compactor/compactor_test.go

@@ -145,6 +145,53 @@ func TestConfig_Validate(t *testing.T) {
 			initLimits: func(_ *validation.Limits) {},
 			expected:   errInvalidTenantShardSize.Error(),
 		},
+		"should pass with an empty block ranges compaction delay": {
+			setup: func(cfg *Config) {
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   "",
+		},
+		"should pass with a block ranges compaction delay entry per block range": {
+			setup: func(cfg *Config) {
+				cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{2 * time.Hour, 150 * time.Minute, 30 * time.Minute}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   "",
+		},
+		"should pass with a zero block ranges compaction delay entry": {
+			setup: func(cfg *Config) {
+				cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{0, 0, 0}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   "",
+		},
+		"should fail with fewer block ranges compaction delay entries than block ranges": {
+			setup: func(cfg *Config) {
+				cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{2 * time.Hour}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   errors.Errorf(errInvalidBlockRangesCompactionDelayLen, 1, 3).Error(),
+		},
+		"should fail with more block ranges compaction delay entries than block ranges": {
+			setup: func(cfg *Config) {
+				cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour}
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{2 * time.Hour, 150 * time.Minute, 30 * time.Minute}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   errors.Errorf(errInvalidBlockRangesCompactionDelayLen, 3, 2).Error(),
+		},
+		"should fail with a negative block ranges compaction delay entry": {
+			setup: func(cfg *Config) {
+				cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
+				cfg.BlockRangesCompactionDelay = cortex_tsdb.DurationList{2 * time.Hour, -1 * time.Minute, 30 * time.Minute}
+			},
+			initLimits: func(_ *validation.Limits) {},
+			expected:   errors.Errorf(errNegativeBlockRangesCompactionDelay, (-1 * time.Minute).String(), 1).Error(),
+		},
 	}
 
 	for testName, testData := range tests {

pkg/compactor/partition_compaction_grouper.go

@@ -29,6 +29,14 @@ var (
 	DUMMY_BLOCK_ID = ulid.ULID{}
 )
 
+// defaultCompactionDelayCleanupIntervalMultiplier is how many cleanup intervals a
+// compaction group's time range end must be in the past before the group becomes
+// eligible for compaction, when -compactor.block-ranges-compaction-delay does not
+// configure a delay for that range. The cleanup interval also drives how often the
+// bucket index is refreshed, so waiting for two of them keeps grouping from acting on
+// a bucket index that is missing recently uploaded blocks.
+const defaultCompactionDelayCleanupIntervalMultiplier = 2
+
 type PartitionCompactionGrouper struct {
 	ctx                      context.Context
 	logger                   log.Logger
@@ -283,7 +291,7 @@ func (g *PartitionCompactionGrouper) groupBlocksByRange(blocks []*metadata.Meta,
 		}
 
 		if len(group.blocks) > 1 {
-			if time.UnixMilli(group.rangeEnd).Before(time.Now().Add(-2 * g.compactorCfg.CleanupInterval)) {
+			if time.UnixMilli(group.rangeEnd).Before(time.Now().Add(-g.compactionDelayForRange(tr))) {
 				ret = append(ret, group)
 			}
 		}
@@ -292,6 +300,24 @@ func (g *PartitionCompactionGrouper) groupBlocksByRange(blocks []*metadata.Meta,
 	return ret
 }
 
+// compactionDelayForRange returns the minimum age the end of a compaction group's time
+// range must reach before the group becomes eligible for compaction. The configured
+// delays are matched to the configured block ranges by position; any range without a
+// configured delay falls back to defaultCompactionDelayCleanupIntervalMultiplier times
+// the cleanup interval, which keeps the grouping decision at least two bucket index
+// refreshes behind the blocks it is grouping.
+func (g *PartitionCompactionGrouper) compactionDelayForRange(tr int64) time.Duration {
+	if len(g.compactorCfg.BlockRangesCompactionDelay) == len(g.compactorCfg.BlockRanges) {
+		for i, blockRange := range g.compactorCfg.BlockRanges {
+			if blockRange.Milliseconds() == tr {
+				return g.compactorCfg.BlockRangesCompactionDelay[i]
+			}
+		}
+	}
+
+	return defaultCompactionDelayCleanupIntervalMultiplier * g.compactorCfg.CleanupInterval
+}
+
 func (g *PartitionCompactionGrouper) sortBlockGroups(groups []blocksGroupWithPartition) {
 	// Ensure groups are sorted by smallest range, oldest min time first. The rationale
 	// is that we wanna favor smaller ranges first (ie. to deduplicate samples sooner