diff --git a/CHANGELOG.md b/CHANGELOG.md index ae39687a31..5e0de86c4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,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. #7920 * [ENHANCEMENT] Update Go version to 1.27.3, which includes security fixes to the `crypto/x509` package (CVE-2026-111943). #7906 #7923 * [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 diff --git a/docs/blocks-storage/compactor.md b/docs/blocks-storage/compactor.md index 10abc9c819..23c03136fb 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 b68f0174ab..5bdae1e692 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 bca40f73d2..3b03e8477d 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 692b8bb3b7..123327911d 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 70da4774ea..a033257f95 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 0000000000..065a8ec17f --- /dev/null +++ b/pkg/compactor/sharded_block_populator_test.go @@ -0,0 +1,271 @@ +package compactor + +import ( + "context" + "fmt" + "sort" + "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" + "go.uber.org/atomic" +) + +// 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 := 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 := 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)) + } + } + + 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 18db514cdd..49ab06cf72 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 c5c933ef87..776aa5ee08 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.",