From c3546db2b3d25c61223ab8caf58cd8fe3ec6c426 Mon Sep 17 00:00:00 2001 From: Alex Le Date: Fri, 6 Nov 2026 23:59:22 +0000 Subject: [PATCH 1/4] Compactor: add block merge fan-in to parallelise the compaction merge A partitioning compaction builds one series set per source block and merges them all in a single flat merge. Advancing that merge reads the winning block's series entry off disk, and every one of those reads is issued from the one goroutine that advances the merge heap, so the disk sees a queue depth near one however much IOPS the volume has provisioned. A compaction over a large number of blocks is then bound by read latency rather than by disk capacity, and provisioning more IOPS does not help it. Add -compactor.block-merge-fan-in to group the per-block series sets into contiguous subtrees before the final merge. Each subtree is wrapped in a background chunk series set, so it advances in its own goroutine and the reads behind the subtrees overlap. The flag is disabled by default (0), which keeps the single flat merge unchanged. BenchmarkMergeFanIn measures the effect. 64 sets, 1024 series replicated 3 ways, so 3136 set advances per op, with latency injected per advance: latency fan-in merge goroutines ns/op ns/advance vs flat 0 0 (off) 1 7519760 2398 -- 0 4 16 8696905 2773 0.86x 0 8 8 9024411 2878 0.83x 0 16 4 8390016 2675 0.90x 1ms 0 (off) 1 3451540907 1100619 -- 1ms 16 4 914257427 291536 3.8x 1ms 8 8 496123534 158203 6.9x 1ms 4 16 287013931 91522 12.0x The ns/advance column is the mechanism: a flat merge pays the full 1.10 ms per advance because every read queues behind one goroutine, and 16 goroutines bring that to 91.5 us without changing the read volume at all. Speedup tracks roughly 0.75x the goroutine count; the shortfall is the root merge's own serial handoff. Note the fan-in is inverted against goroutine count, since a smaller fan-in produces more subtrees. With no read latency to hide, the change costs 10-17%, which is why it is off by default. A workload whose blocks fit in page cache should leave it off. time.Sleep cannot resolve latencies below roughly 100us, so the benchmark reports ns/advance to show the latency a row actually achieved rather than the latency it asked for. TestMergeSubtreesMatchesFlatMerge asserts the grouped merge yields identical labels and samples to the flat merge across several fan-ins, including one that leaves a short final group. It compares samples flattened across chunk boundaries on purpose: a merge tree may lay the same data out in different chunks, which changes the output block's chunk count but not its data. Signed-off-by: Alex Le --- CHANGELOG.md | 1 + docs/blocks-storage/compactor.md | 9 + docs/configuration/config-file-reference.md | 9 + pkg/compactor/background_chunks_series_set.go | 14 +- pkg/compactor/compactor.go | 15 +- pkg/compactor/sharded_block_populator.go | 34 ++- pkg/compactor/sharded_block_populator_test.go | 271 ++++++++++++++++++ .../sharded_compaction_lifecycle_callback.go | 10 +- schemas/cortex-config-schema.json | 6 + 9 files changed, 360 insertions(+), 9 deletions(-) create mode 100644 pkg/compactor/sharded_block_populator_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index ad3907ad588..4606166d4bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ * [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 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889 * [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909 +* [ENHANCEMENT] Compactor: Add `-compactor.block-merge-fan-in` to group a partitioning compaction's per-block series sets into subtrees, each advanced by its own goroutine, before the final merge. A flat merge issues every index read from the single goroutine that advances the merge heap, so a compaction over a large number of blocks drives the disk at a queue depth near one however much IOPS it has provisioned, and its wall clock is bound by read latency rather than by disk capacity. Disabled by default (`0`), which keeps the single flat merge. #TODO * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880 diff --git a/docs/blocks-storage/compactor.md b/docs/blocks-storage/compactor.md index 10abc9c8195..23c03136fb2 100644 --- a/docs/blocks-storage/compactor.md +++ b/docs/blocks-storage/compactor.md @@ -113,6 +113,15 @@ compactor: # CLI flag: -compactor.meta-sync-concurrency [meta_sync_concurrency: | default = 20] + # Number of per-block series sets merged per subtree during compaction. Each + # subtree advances in its own goroutine, so a group of blocks larger than this + # value no longer issues every index read from a single goroutine. Only + # affects partitioning compaction. 0 disables it and keeps one flat merge over + # all blocks. Useful range is 32 to 128; lower values create more goroutines + # without adding disk concurrency. + # CLI flag: -compactor.block-merge-fan-in + [block_merge_fan_in: | default = 0] + # Minimum age of fresh (non-compacted) blocks before they are being processed. # Malformed blocks older than the maximum of consistency-delay and 48h0m0s # will be removed. diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index b68f0174abb..5bdae1e6923 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3124,6 +3124,15 @@ The `compactor_config` configures the compactor for the blocks storage. # CLI flag: -compactor.meta-sync-concurrency [meta_sync_concurrency: | default = 20] +# Number of per-block series sets merged per subtree during compaction. Each +# subtree advances in its own goroutine, so a group of blocks larger than this +# value no longer issues every index read from a single goroutine. Only affects +# partitioning compaction. 0 disables it and keeps one flat merge over all +# blocks. Useful range is 32 to 128; lower values create more goroutines without +# adding disk concurrency. +# CLI flag: -compactor.block-merge-fan-in +[block_merge_fan_in: | default = 0] + # Minimum age of fresh (non-compacted) blocks before they are being processed. # Malformed blocks older than the maximum of consistency-delay and 48h0m0s will # be removed. diff --git a/pkg/compactor/background_chunks_series_set.go b/pkg/compactor/background_chunks_series_set.go index bca40f73d2e..3b03e8477db 100644 --- a/pkg/compactor/background_chunks_series_set.go +++ b/pkg/compactor/background_chunks_series_set.go @@ -46,10 +46,22 @@ func (b *backgrounChunkSeriesSet) run(ctx context.Context) { } } +// defaultBackgroundChunkSeriesSetBuffer is the lookahead used for a single +// background set. A merge tree holds one buffer per subtree, so inner levels +// pass a smaller value to bound total series held in memory. +const defaultBackgroundChunkSeriesSetBuffer = 1000 + func NewBackgroundChunkSeriesSet(ctx context.Context, cs storage.ChunkSeriesSet) storage.ChunkSeriesSet { + return NewBackgroundChunkSeriesSetWithBuffer(ctx, cs, defaultBackgroundChunkSeriesSetBuffer) +} + +func NewBackgroundChunkSeriesSetWithBuffer(ctx context.Context, cs storage.ChunkSeriesSet, buffer int) storage.ChunkSeriesSet { + if buffer < 1 { + buffer = 1 + } r := &backgrounChunkSeriesSet{ cs: cs, - nextSet: make(chan storage.ChunkSeries, 1000), + nextSet: make(chan storage.ChunkSeries, buffer), } go func() { diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go index 692b8bb3b7f..123327911d1 100644 --- a/pkg/compactor/compactor.go +++ b/pkg/compactor/compactor.go @@ -59,6 +59,7 @@ var ( supportedShardingStrategies = []string{util.ShardingStrategyDefault, util.ShardingStrategyShuffle} errInvalidShardingStrategy = errors.New("invalid sharding strategy") + errInvalidBlockMergeFanIn = errors.New("invalid block merge fan-in, the value must not be negative") errInvalidTenantShardSize = errors.New("invalid tenant shard size, the value must be greater than 0") supportedCompactionStrategies = []string{util.CompactionStrategyDefault, util.CompactionStrategyPartitioning} errInvalidCompactionStrategy = errors.New("invalid compaction strategy") @@ -173,16 +174,17 @@ var ( return NewPartitionCompactionBlockDeletableChecker() } - DefaultCompactionLifecycleCallbackFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ int, _ string, _ string, _ *compactorMetrics) compact.CompactionLifecycleCallback { + DefaultCompactionLifecycleCallbackFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ int, _ int, _ string, _ string, _ *compactorMetrics) compact.CompactionLifecycleCallback { return compact.DefaultCompactionLifecycleCallback{} } - ShardedCompactionLifecycleCallbackFactory = func(ctx context.Context, userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, compactDir string, userID string, compactorMetrics *compactorMetrics) compact.CompactionLifecycleCallback { + ShardedCompactionLifecycleCallbackFactory = func(ctx context.Context, userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, blockMergeFanIn int, compactDir string, userID string, compactorMetrics *compactorMetrics) compact.CompactionLifecycleCallback { return NewShardedCompactionLifecycleCallback( ctx, userBucket, logger, metaSyncConcurrency, + blockMergeFanIn, compactDir, userID, compactorMetrics, @@ -236,6 +238,7 @@ type CompactionLifecycleCallbackFactory func( userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, + blockMergeFanIn int, compactDir string, userID string, compactorMetrics *compactorMetrics, @@ -260,6 +263,7 @@ type Config struct { BlockRangesCompactionDelay cortex_tsdb.DurationList `yaml:"block_ranges_compaction_delay"` BlockSyncConcurrency int `yaml:"block_sync_concurrency"` MetaSyncConcurrency int `yaml:"meta_sync_concurrency"` + BlockMergeFanIn int `yaml:"block_merge_fan_in"` ConsistencyDelay time.Duration `yaml:"consistency_delay"` DataDir string `yaml:"data_dir"` CompactionInterval time.Duration `yaml:"compaction_interval"` @@ -324,6 +328,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) { 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.") + f.IntVar(&cfg.BlockMergeFanIn, "compactor.block-merge-fan-in", 0, "Number of per-block series sets merged per subtree during compaction. Each subtree advances in its own goroutine, so a group of blocks larger than this value no longer issues every index read from a single goroutine. Only affects partitioning compaction. 0 disables it and keeps one flat merge over all blocks. Useful range is 32 to 128; lower values create more goroutines without adding disk concurrency.") f.StringVar(&cfg.DataDir, "compactor.data-dir", "./data", "Data directory in which to cache blocks and process compactions") f.DurationVar(&cfg.CompactionInterval, "compactor.compaction-interval", time.Hour, "The frequency at which the compaction runs") f.IntVar(&cfg.CompactionRetries, "compactor.compaction-retries", 3, "How many times to retry a failed compaction within a single compaction run.") @@ -382,6 +387,10 @@ func (cfg *Config) Validate(limits validation.Limits) error { } } + if cfg.BlockMergeFanIn < 0 { + return errInvalidBlockMergeFanIn + } + // Make sure a valid sharding strategy is being used if !slices.Contains(supportedShardingStrategies, cfg.ShardingStrategy) { return errInvalidShardingStrategy @@ -1114,7 +1123,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error { c.blocksPlannerFactory(currentCtx, bucket, ulogger, c.compactorCfg, noCompactMarkerFilter, c.ringLifecycler, userID, c.blockVisitMarkerReadFailed, c.blockVisitMarkerWriteFailed, c.compactorMetrics, ignoreDeletionMarkFilter), c.blocksCompactor, c.blockDeletableCheckerFactory(currentCtx, bucket, ulogger), - c.compactionLifecycleCallbackFactory(currentCtx, bucket, ulogger, c.compactorCfg.MetaSyncConcurrency, c.compactDirForUser(userID), userID, c.compactorMetrics), + c.compactionLifecycleCallbackFactory(currentCtx, bucket, ulogger, c.compactorCfg.MetaSyncConcurrency, c.compactorCfg.BlockMergeFanIn, c.compactDirForUser(userID), userID, c.compactorMetrics), c.compactDirForUser(userID), bucket, c.compactorCfg.CompactionConcurrency, diff --git a/pkg/compactor/sharded_block_populator.go b/pkg/compactor/sharded_block_populator.go index 70da4774eab..a033257f953 100644 --- a/pkg/compactor/sharded_block_populator.go +++ b/pkg/compactor/sharded_block_populator.go @@ -23,7 +23,10 @@ import ( type ShardedBlockPopulator struct { partitionCount int partitionID int - logger log.Logger + // blockMergeFanIn groups the per-block series sets into subtrees before the + // final merge. 0 or 1 keeps the single flat merge. + blockMergeFanIn int + logger log.Logger } // PopulateBlock fills the index and chunk writers with new data gathered as the union @@ -131,7 +134,7 @@ func (c ShardedBlockPopulator) PopulateBlock(ctx context.Context, metrics *tsdb. iCtx, cancel := context.WithCancel(ctx) // Merge series using specified chunk series merger. // The default one is the compacting series merger. - set = NewBackgroundChunkSeriesSet(iCtx, storage.NewMergeChunkSeriesSet(sets, 0, mergeFunc)) + set = NewBackgroundChunkSeriesSet(iCtx, storage.NewMergeChunkSeriesSet(c.mergeSubtrees(iCtx, sets, mergeFunc), 0, mergeFunc)) defer cancel() } @@ -222,3 +225,30 @@ func (c ShardedBlockPopulator) PopulateBlock(ctx context.Context, metrics *tsdb. return nil } + +// mergeSubtrees groups sets into subtrees, each advanced by its own goroutine, +// and returns the subtree roots for the caller to merge. +// +// A flat merge over every block issues all of its index reads from the single +// goroutine that advances the heap, so the disk sees a queue depth near one +// whatever it has provisioned. Grouping gives one in-flight read per subtree. +// Returns sets unchanged when disabled or when there is nothing to group. +func (c ShardedBlockPopulator) mergeSubtrees(ctx context.Context, sets []storage.ChunkSeriesSet, mergeFunc storage.VerticalChunkSeriesMergeFunc) []storage.ChunkSeriesSet { + fanIn := c.blockMergeFanIn + if fanIn < 2 || len(sets) <= fanIn { + return sets + } + + subtrees := (len(sets) + fanIn - 1) / fanIn + // Split the single-set lookahead across the subtrees so total series held + // in memory stays comparable to the flat merge. + buffer := max(defaultBackgroundChunkSeriesSetBuffer/subtrees, 8) + + roots := make([]storage.ChunkSeriesSet, 0, subtrees) + for i := 0; i < len(sets); i += fanIn { + group := sets[i:min(i+fanIn, len(sets))] + roots = append(roots, NewBackgroundChunkSeriesSetWithBuffer(ctx, storage.NewMergeChunkSeriesSet(group, 0, mergeFunc), buffer)) + } + level.Info(c.logger).Log("msg", "grouped series sets into merge subtrees", "series_sets_count", len(sets), "subtrees", len(roots), "fan_in", fanIn, "subtree_buffer", buffer) + return roots +} diff --git a/pkg/compactor/sharded_block_populator_test.go b/pkg/compactor/sharded_block_populator_test.go new file mode 100644 index 00000000000..0246facf2f5 --- /dev/null +++ b/pkg/compactor/sharded_block_populator_test.go @@ -0,0 +1,271 @@ +package compactor + +import ( + "context" + "fmt" + "sort" + "sync/atomic" + "testing" + "time" + + "github.com/go-kit/log" + "github.com/prometheus/prometheus/model/histogram" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/storage" + "github.com/prometheus/prometheus/tsdb/chunkenc" + "github.com/prometheus/prometheus/tsdb/chunks" + "github.com/prometheus/prometheus/util/annotations" + "github.com/stretchr/testify/require" +) + +// testSample is a float-only chunks.Sample; the package's own implementation is +// unexported. +type testSample struct { + t int64 + v float64 +} + +func (s testSample) T() int64 { return s.t } +func (s testSample) F() float64 { return s.v } +func (s testSample) H() *histogram.Histogram { return nil } +func (s testSample) FH() *histogram.FloatHistogram { return nil } +func (s testSample) Type() chunkenc.ValueType { return chunkenc.ValFloat } +func (s testSample) Copy() chunks.Sample { return s } + +// sliceChunkSeriesSet serves a pre-sorted slice of series once. +type sliceChunkSeriesSet struct { + series []storage.ChunkSeries + idx int +} + +func newSliceChunkSeriesSet(series []storage.ChunkSeries) *sliceChunkSeriesSet { + return &sliceChunkSeriesSet{series: series, idx: -1} +} + +func (s *sliceChunkSeriesSet) Next() bool { + s.idx++ + return s.idx < len(s.series) +} +func (s *sliceChunkSeriesSet) At() storage.ChunkSeries { return s.series[s.idx] } +func (s *sliceChunkSeriesSet) Err() error { return nil } +func (s *sliceChunkSeriesSet) Warnings() annotations.Annotations { return nil } + +type collectedSeries struct { + labels string + samples []testSample +} + +// collect flattens samples across chunk boundaries on purpose. A merge tree may +// lay the same samples out in different chunks than a flat merge, which changes +// the output block's chunk count but not its data; the samples are what must +// match. +func collect(t *testing.T, set storage.ChunkSeriesSet) []collectedSeries { + t.Helper() + var out []collectedSeries + for set.Next() { + s := set.At() + cur := collectedSeries{labels: s.Labels().String()} + chkIter := s.Iterator(nil) + for chkIter.Next() { + var valIter chunkenc.Iterator + valIter = chkIter.At().Chunk.Iterator(valIter) + for valIter.Next() == chunkenc.ValFloat { + ts, v := valIter.At() + cur.samples = append(cur.samples, testSample{t: ts, v: v}) + } + require.NoError(t, valIter.Err()) + } + require.NoError(t, chkIter.Err()) + out = append(out, cur) + } + require.NoError(t, set.Err()) + return out +} + +// buildSets returns setCount sorted sets. Each series is replicated into three +// sets with partially overlapping sample windows, and a given timestamp always +// carries the same value, which is how ingester replicas present the same +// series to the compactor. +func buildSets(setCount, seriesCount int) []storage.ChunkSeriesSet { + perSet := make([][]storage.ChunkSeries, setCount) + for i := 0; i < seriesCount; i++ { + lbls := labels.FromStrings(labels.MetricName, "test_metric", "series", fmt.Sprintf("%03d", i)) + for _, offset := range []int{0, 1, 3} { + set := (i + offset) % setCount + samples := make([]chunks.Sample, 0, 12) + start := int64(offset) * 40 + for k := int64(0); k < 12; k++ { + ts := start + k*10 + samples = append(samples, testSample{t: ts, v: float64(ts) * 1.5}) + } + perSet[set] = append(perSet[set], storage.NewListChunkSeriesFromSamples(lbls, samples)) + } + } + + sets := make([]storage.ChunkSeriesSet, 0, setCount) + for _, series := range perSet { + sort.Slice(series, func(a, b int) bool { + return labels.Compare(series[a].Labels(), series[b].Labels()) < 0 + }) + sets = append(sets, newSliceChunkSeriesSet(series)) + } + return sets +} + +func TestMergeSubtreesMatchesFlatMerge(t *testing.T) { + const ( + setCount = 9 + seriesCount = 40 + ) + mergeFunc := storage.NewCompactingChunkSeriesMerger(storage.ChainedSeriesMerge) + + flat := collect(t, storage.NewMergeChunkSeriesSet(buildSets(setCount, seriesCount), 0, mergeFunc)) + require.Len(t, flat, seriesCount) + + // Fan-ins that divide the set count evenly and that do not, so the short + // final group is covered too. + for _, fanIn := range []int{2, 3, 4, 8} { + t.Run(fmt.Sprintf("fan_in_%d", fanIn), func(t *testing.T) { + c := ShardedBlockPopulator{blockMergeFanIn: fanIn, logger: log.NewNopLogger()} + sets := buildSets(setCount, seriesCount) + roots := c.mergeSubtrees(context.Background(), sets, mergeFunc) + require.Less(t, len(roots), len(sets), "fan-in should have grouped the sets") + + tree := collect(t, storage.NewMergeChunkSeriesSet(roots, 0, mergeFunc)) + require.Equal(t, flat, tree) + }) + } +} + +func TestMergeSubtreesDisabled(t *testing.T) { + mergeFunc := storage.NewCompactingChunkSeriesMerger(storage.ChainedSeriesMerge) + for _, fanIn := range []int{0, 1, -1} { + t.Run(fmt.Sprintf("fan_in_%d", fanIn), func(t *testing.T) { + c := ShardedBlockPopulator{blockMergeFanIn: fanIn, logger: log.NewNopLogger()} + sets := buildSets(4, 4) + require.Equal(t, sets, c.mergeSubtrees(context.Background(), sets, mergeFunc), + "a disabled fan-in must return the sets untouched") + }) + } +} + +func TestMergeSubtreesGrouping(t *testing.T) { + mergeFunc := storage.NewCompactingChunkSeriesMerger(storage.ChainedSeriesMerge) + for _, tc := range []struct { + sets, fanIn, expected int + }{ + {sets: 10, fanIn: 5, expected: 2}, + {sets: 10, fanIn: 3, expected: 4}, + {sets: 10, fanIn: 10, expected: 10}, // not larger than fan-in, left flat + {sets: 10, fanIn: 11, expected: 10}, + {sets: 2761, fanIn: 64, expected: 44}, + } { + t.Run(fmt.Sprintf("sets_%d_fan_in_%d", tc.sets, tc.fanIn), func(t *testing.T) { + c := ShardedBlockPopulator{blockMergeFanIn: tc.fanIn, logger: log.NewNopLogger()} + sets := make([]storage.ChunkSeriesSet, tc.sets) + for i := range sets { + sets[i] = newSliceChunkSeriesSet(nil) + } + require.Len(t, c.mergeSubtrees(context.Background(), sets, mergeFunc), tc.expected) + }) + } +} + +// latencySet models the index read that backs a merge advance. Advancing a +// real BlockChunkSeriesSet reads the winning block's series entry off disk, and +// that read is what a flat merge serialises onto one goroutine. +type latencySet struct { + storage.ChunkSeriesSet + latency time.Duration + advances *atomic.Int64 +} + +func (l *latencySet) Next() bool { + l.advances.Add(1) + if l.latency > 0 { + time.Sleep(l.latency) + } + return l.ChunkSeriesSet.Next() +} + +func buildLatencySets(setCount, seriesCount int, latency time.Duration, advances *atomic.Int64) []storage.ChunkSeriesSet { + sets := buildSets(setCount, seriesCount) + for i, s := range sets { + sets[i] = &latencySet{ChunkSeriesSet: s, latency: latency, advances: advances} + } + return sets +} + +// BenchmarkMergeFanIn compares the flat merge against the subtree merge. +// +// latency=0 isolates the cost of the extra goroutines and channel handoffs, +// where the flat merge is expected to win. The non-zero cases are the ones the +// flag exists for: with per-advance read latency the flat merge serialises +// every read onto one goroutine, and grouping overlaps them. +// +// Latencies below roughly 100us are not worth running: time.Sleep cannot +// resolve them, so every row collapses onto the timer floor and the comparison +// says nothing. 1ms is close to a real EBS gp3 read. +// +// Run with a fixed iteration count for stable numbers, and do not pin to one +// CPU, which inflates every sleep. The latency cases are slow by construction: +// +// go test -run XXX -bench BenchmarkMergeFanIn -benchtime 5x ./pkg/compactor/ +func BenchmarkMergeFanIn(b *testing.B) { + const ( + setCount = 64 + seriesCount = 1024 + ) + mergeFunc := storage.NewCompactingChunkSeriesMerger(storage.ChainedSeriesMerge) + + for _, latency := range []time.Duration{0, time.Millisecond} { + for _, fanIn := range []int{0, 4, 8, 16} { + name := fmt.Sprintf("latency=%v/fan_in=%d", latency, fanIn) + b.Run(name, func(b *testing.B) { + c := ShardedBlockPopulator{blockMergeFanIn: fanIn, logger: log.NewNopLogger()} + var advances atomic.Int64 + // Goroutines that advance sets, which is the read concurrency + // on offer. A flat merge has one however many sets it holds. + mergeGoroutines := 1 + if fanIn >= 2 && setCount > fanIn { + mergeGoroutines = (setCount + fanIn - 1) / fanIn + } + for i := 0; i < b.N; i++ { + b.StopTimer() + sets := buildLatencySets(setCount, seriesCount, latency, &advances) + ctx, cancel := context.WithCancel(context.Background()) + b.StartTimer() + + // Mirrors PopulateBlock: the root is always wrapped, only + // the layer underneath it changes. + set := NewBackgroundChunkSeriesSet(ctx, storage.NewMergeChunkSeriesSet( + c.mergeSubtrees(ctx, sets, mergeFunc), 0, mergeFunc)) + + got := 0 + for set.Next() { + chkIter := set.At().Iterator(nil) + for chkIter.Next() { + } + if err := chkIter.Err(); err != nil { + b.Fatal(err) + } + got++ + } + if got != seriesCount { + b.Fatalf("got %d series, want %d", got, seriesCount) + } + cancel() + } + // ns/advance is the latency actually achieved per set advance, + // which is the guard against trusting a row whose sleeps never + // resolved to the requested latency. + b.ReportMetric(float64(mergeGoroutines), "merge_goroutines") + n := advances.Load() + if n > 0 && b.N > 0 { + b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(n), "ns/advance") + b.ReportMetric(float64(n)/float64(b.N), "advances/op") + } + }) + } + } +} diff --git a/pkg/compactor/sharded_compaction_lifecycle_callback.go b/pkg/compactor/sharded_compaction_lifecycle_callback.go index 18db514cdd7..49ab06cf721 100644 --- a/pkg/compactor/sharded_compaction_lifecycle_callback.go +++ b/pkg/compactor/sharded_compaction_lifecycle_callback.go @@ -22,6 +22,7 @@ type ShardedCompactionLifecycleCallback struct { userBucket objstore.InstrumentedBucket logger log.Logger metaSyncConcurrency int + blockMergeFanIn int compactDir string userID string compactorMetrics *compactorMetrics @@ -34,6 +35,7 @@ func NewShardedCompactionLifecycleCallback( userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, + blockMergeFanIn int, compactDir string, userID string, compactorMetrics *compactorMetrics, @@ -43,6 +45,7 @@ func NewShardedCompactionLifecycleCallback( userBucket: userBucket, logger: logger, metaSyncConcurrency: metaSyncConcurrency, + blockMergeFanIn: blockMergeFanIn, compactDir: compactDir, userID: userID, compactorMetrics: compactorMetrics, @@ -100,9 +103,10 @@ func (c *ShardedCompactionLifecycleCallback) GetBlockPopulator(_ context.Context }) } populateBlockFunc := ShardedBlockPopulator{ - partitionCount: partitionInfo.PartitionCount, - partitionID: partitionInfo.PartitionID, - logger: logger, + partitionCount: partitionInfo.PartitionCount, + partitionID: partitionInfo.PartitionID, + blockMergeFanIn: c.blockMergeFanIn, + logger: logger, } return populateBlockFunc, nil } diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index c5c933ef875..776aa5ee08d 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -3747,6 +3747,12 @@ "type": "number", "x-cli-flag": "compactor.block-files-concurrency" }, + "block_merge_fan_in": { + "default": 0, + "description": "Number of per-block series sets merged per subtree during compaction. Each subtree advances in its own goroutine, so a group of blocks larger than this value no longer issues every index read from a single goroutine. Only affects partitioning compaction. 0 disables it and keeps one flat merge over all blocks. Useful range is 32 to 128; lower values create more goroutines without adding disk concurrency.", + "type": "number", + "x-cli-flag": "compactor.block-merge-fan-in" + }, "block_ranges": { "default": "2h0m0s,12h0m0s,24h0m0s", "description": "List of compaction time ranges.", From 1e8b5d7996eb83a306c673565d00c71bb431516f Mon Sep 17 00:00:00 2001 From: Alex Le Date: Sat, 7 Nov 2026 00:05:58 +0000 Subject: [PATCH 2/4] Compactor: reference the PR number in the changelog entry Signed-off-by: Alex Le --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4606166d4bd..6750e9d1022 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ * [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 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889 * [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909 -* [ENHANCEMENT] Compactor: Add `-compactor.block-merge-fan-in` to group a partitioning compaction's per-block series sets into subtrees, each advanced by its own goroutine, before the final merge. A flat merge issues every index read from the single goroutine that advances the merge heap, so a compaction over a large number of blocks drives the disk at a queue depth near one however much IOPS it has provisioned, and its wall clock is bound by read latency rather than by disk capacity. Disabled by default (`0`), which keeps the single flat merge. #TODO +* [ENHANCEMENT] Compactor: Add `-compactor.block-merge-fan-in` to group a partitioning compaction's per-block series sets into subtrees, each advanced by its own goroutine, before the final merge. A flat merge issues every index read from the single goroutine that advances the merge heap, so a compaction over a large number of blocks drives the disk at a queue depth near one however much IOPS it has provisioned, and its wall clock is bound by read latency rather than by disk capacity. Disabled by default (`0`), which keeps the single flat merge. #7920 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880 From dc0a8573138611071b98e30a9fd403227f9c375d Mon Sep 17 00:00:00 2001 From: Alex Le Date: Sat, 7 Nov 2026 00:42:21 +0000 Subject: [PATCH 3/4] Compactor: use range over int in the merge fan-in test golangci-lint's modernize linter rejects the three-clause loops, and the package's existing tests already use the range form. Signed-off-by: Alex Le --- pkg/compactor/sharded_block_populator_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/compactor/sharded_block_populator_test.go b/pkg/compactor/sharded_block_populator_test.go index 0246facf2f5..78620c8bed4 100644 --- a/pkg/compactor/sharded_block_populator_test.go +++ b/pkg/compactor/sharded_block_populator_test.go @@ -88,14 +88,14 @@ func collect(t *testing.T, set storage.ChunkSeriesSet) []collectedSeries { // series to the compactor. func buildSets(setCount, seriesCount int) []storage.ChunkSeriesSet { perSet := make([][]storage.ChunkSeries, setCount) - for i := 0; i < seriesCount; i++ { + for i := range seriesCount { lbls := labels.FromStrings(labels.MetricName, "test_metric", "series", fmt.Sprintf("%03d", i)) for _, offset := range []int{0, 1, 3} { set := (i + offset) % setCount samples := make([]chunks.Sample, 0, 12) start := int64(offset) * 40 - for k := int64(0); k < 12; k++ { - ts := start + k*10 + for k := range 12 { + ts := start + int64(k)*10 samples = append(samples, testSample{t: ts, v: float64(ts) * 1.5}) } perSet[set] = append(perSet[set], storage.NewListChunkSeriesFromSamples(lbls, samples)) From 4286625a931ae857d873e4c70ba7f0df61b04a1e Mon Sep 17 00:00:00 2001 From: Alex Le Date: Sat, 7 Nov 2026 00:55:32 +0000 Subject: [PATCH 4/4] Compactor: use go.uber.org/atomic in the merge fan-in benchmark make lint runs faillint with sync/atomic blocklisted in favour of go.uber.org/atomic, which this package already uses elsewhere. The Int64 Add and Load API is the same and its zero value is usable. Signed-off-by: Alex Le --- pkg/compactor/sharded_block_populator_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/compactor/sharded_block_populator_test.go b/pkg/compactor/sharded_block_populator_test.go index 78620c8bed4..065a8ec17fd 100644 --- a/pkg/compactor/sharded_block_populator_test.go +++ b/pkg/compactor/sharded_block_populator_test.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "sort" - "sync/atomic" "testing" "time" @@ -16,6 +15,7 @@ import ( "github.com/prometheus/prometheus/tsdb/chunks" "github.com/prometheus/prometheus/util/annotations" "github.com/stretchr/testify/require" + "go.uber.org/atomic" ) // testSample is a float-only chunks.Sample; the package's own implementation is