diff --git a/cmd/pyroscope/help-all.txt.tmpl b/cmd/pyroscope/help-all.txt.tmpl index 5324c26455..0b66795479 100644 --- a/cmd/pyroscope/help-all.txt.tmpl +++ b/cmd/pyroscope/help-all.txt.tmpl @@ -51,6 +51,8 @@ Usage of ./pyroscope: Duration after which the blocks marked for deletion will be filtered out while fetching blocks. The idea of ignore-deletion-marks-delay is to ignore blocks that are marked for deletion with some delay. This ensures store can still serve blocks that are meant to be deleted but do not have a replacement yet. (default 30m0s) -blocks-storage.bucket-store.meta-sync-concurrency int Number of Go routines to use when syncing block meta files from object storage per tenant. (default 20) + -blocks-storage.bucket-store.symbol-cache-max-bytes int + [experimental] Max size - in bytes - of the in-process cache of decoded symbol tables, shared across all blocks and tenants. Avoids re-decoding a block's symbols on every query. 0 disables the cache. -blocks-storage.bucket-store.sync-dir string Directory to store synchronized pyroscope block headers. This directory is not required to be persisted between restarts, but it's highly recommended in order to improve the store-gateway startup time. (default "./data/pyroscope-sync/") -blocks-storage.bucket-store.sync-interval duration diff --git a/docs/sources/configure-server/reference-configuration-parameters/index.md b/docs/sources/configure-server/reference-configuration-parameters/index.md index 192d7e4ed3..40a1c1644e 100644 --- a/docs/sources/configure-server/reference-configuration-parameters/index.md +++ b/docs/sources/configure-server/reference-configuration-parameters/index.md @@ -2056,6 +2056,12 @@ bucket_store: # have a replacement yet. # CLI flag: -blocks-storage.bucket-store.ignore-deletion-marks-delay [ignore_deletion_mark_delay: | default = 30m] + + # (experimental) Max size - in bytes - of the in-process cache of decoded + # symbol tables, shared across all blocks and tenants. Avoids re-decoding a + # block's symbols on every query. 0 disables the cache. + # CLI flag: -blocks-storage.bucket-store.symbol-cache-max-bytes + [symbol_cache_max_bytes: | default = 0] ``` ### compactor diff --git a/pkg/compactor/bucket_compactor.go b/pkg/compactor/bucket_compactor.go index dfbc5aad87..b5cb50f339 100644 --- a/pkg/compactor/bucket_compactor.go +++ b/pkg/compactor/bucket_compactor.go @@ -340,6 +340,18 @@ func newCompactorMetrics(r prometheus.Registerer) *CompactorMetrics { return m } +// isLoneIntermediateShardedBlock reports whether the compaction input is a +// single sharded block that has not been merged past the deduplication level. +// The split-and-merge grouper plans such a job on purpose, to promote a block +// that has no sibling to merge with. +func isLoneIntermediateShardedBlock(readers []phlaredb.BlockReader) bool { + if len(readers) != 1 { + return false + } + meta := readers[0].Meta() + return isIntermediateShardedBlock(&meta) +} + func (c *BlockCompactor) CompactWithSplitting(ctx context.Context, dest string, dirs []string, shardCount, stageSize uint64) ([]ulid.ULID, error) { defer func() { if err := recover(); err != nil { @@ -410,11 +422,15 @@ func (c *BlockCompactor) CompactWithSplitting(ctx context.Context, dest string, c.metrics.Split.WithLabelValues(fmt.Sprintf("%d", currentLevel)).Observe(float64(shardCount)) metas, err := phlaredb.CompactWithSplitting(ctx, phlaredb.CompactWithSplittingOpts{ - Src: readers, - Dst: dest, - SplitCount: shardCount, - StageSize: stageSize, - SplitBy: c.splitBy, + Src: readers, + Dst: dest, + SplitCount: shardCount, + StageSize: stageSize, + SplitBy: c.splitBy, + // A merge job with a single input block is planned only to promote a + // sharded block that never got merged past the deduplication level, so + // the single-block compaction is deliberate here. + AllowSingleBlock: isLoneIntermediateShardedBlock(readers), DownsamplerEnabled: c.downsamplerEnabled, Logger: c.logger, }) diff --git a/pkg/compactor/split_merge_grouper.go b/pkg/compactor/split_merge_grouper.go index 3b483b9900..bf630a14da 100644 --- a/pkg/compactor/split_merge_grouper.go +++ b/pkg/compactor/split_merge_grouper.go @@ -124,6 +124,10 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, shardCo // Sort blocks by min time. sortMetasByMinTime(mainBlocks) + // Remember where this group's jobs start, so the premature-compaction + // filter below can be applied to them alone. + groupStart := len(jobs) + for _, tr := range ranges { nextJob: for _, job := range planCompactionByRange(userID, mainBlocks, tr, tr == ranges[0], shardCount, splitGroups) { @@ -140,31 +144,36 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, shardCo jobs = append(jobs, job) } } - } - - // Ensure we don't compact the most recent blocks prematurely when another one of - // the same size still fits in the range. To do it, we consider a job valid only - // if its range is before the most recent block or if it fully covers the range. - highestMaxTime := getMaxTime(blocks) - for idx := 0; idx < len(jobs); { - job := jobs[idx] + // Ensure we don't compact the most recent blocks prematurely when another one of + // the same size still fits in the range. To do it, we consider a job valid only + // if its range is before the most recent block or if it fully covers the range. + // + // The reference point is the most recent block *of this group*: groups are + // independent streams of data, often written by unrelated services on + // unrelated schedules, so taking the maximum across all of them lets the + // group with the freshest data gate compaction for every other group. + highestMaxTime := getMaxTime(mainBlocks) + + for idx := groupStart; 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 before the most recent block, it's fine. - if job.rangeEnd <= highestMaxTime { - idx++ - continue - } + // If the job covers the full range, it's fine. + if job.maxTime()-job.minTime() == job.rangeLength() { + idx++ + continue + } - // If the job covers the full range, it's fine. - if job.maxTime()-job.minTime() == job.rangeLength() { - idx++ - continue + // We have found a job which would compact recent blocks prematurely, + // so we need to filter it out. + jobs = append(jobs[:idx], jobs[idx+1:]...) } - - // We have found a job which would compact recent blocks prematurely, - // so we need to filter it out. - jobs = append(jobs[:idx], jobs[idx+1:]...) } // Jobs will be sorted later using configured job sorting algorithm. @@ -200,8 +209,17 @@ func planCompactionByRange(userID string, blocks []*block.Meta, tr int64, isSmal // (or we're not processing the smallest time range, or splitting is disabled). // Then, we can check if there's any group of blocks to be merged together for each shard. for shardID, shardBlocks := range groupBlocksByShardID(group.blocks) { - // No merging to do if there are less than 2 blocks. - if len(shardBlocks) < 2 { + // Normally there is no merging to do if there are less than 2 blocks: + // a block that has no sibling in its shard is already as compacted as + // it can get, and re-compacting it on its own would never terminate. + // + // The exception is a lone block that is still an intermediate + // compaction artefact, i.e. below the level at which data is + // deduplicated. Such a block is not directly queryable (the querier + // prefers its ancestors, which may already have been deleted), so it + // must be merged on its own to be promoted past the deduplication + // level. This terminates because the merge output is one level higher. + if len(shardBlocks) < 2 && !isOrphanedIntermediateBlock(shardCount, shardID, shardBlocks) { continue } @@ -221,6 +239,32 @@ func planCompactionByRange(userID string, blocks []*block.Meta, tr int64, isSmal return jobs } +// isOrphanedIntermediateBlock reports whether the given single-block shard group +// holds a sharded block that is stuck below the deduplication level, and so has +// to be merged on its own to become queryable. +func isOrphanedIntermediateBlock(shardCount uint32, shardID string, shardBlocks []*block.Meta) bool { + if shardCount == 0 || shardID == "" || len(shardBlocks) != 1 { + return false + } + return isIntermediateShardedBlock(shardBlocks[0]) +} + +// isIntermediateShardedBlock reports whether the block is the sharded output of +// the split stage that has not been merged past the deduplication level yet. +// +// Such a block is only queryable at the cost of deduplicating it against its +// siblings, and only for as long as its ancestors have not been deleted, so the +// compactor has to promote it even though it has no sibling to merge with. +func isIntermediateShardedBlock(meta *block.Meta) bool { + if meta == nil || meta.Labels[sharding.CompactorShardIDLabel] == "" { + return false + } + // A sharded block is always the output of at least the split stage, so a zero + // level means the metadata is missing rather than that the block is new. + compactionLevel := int32(meta.Compaction.Level) + return compactionLevel > 0 && block.IsIntermediate(compactionLevel, true) +} + // planSplitting returns a job to split the blocks in the input group or nil if there's nothing to do because // all blocks in the group have already been split. func planSplitting(userID string, group blocksGroup, splitGroups uint32) []*job { diff --git a/pkg/compactor/split_merge_grouper_test.go b/pkg/compactor/split_merge_grouper_test.go index b86b23ab6c..0688fc7779 100644 --- a/pkg/compactor/split_merge_grouper_test.go +++ b/pkg/compactor/split_merge_grouper_test.go @@ -509,6 +509,67 @@ func TestPlanCompaction(t *testing.T) { }}, }, }, + "should merge a lone sharded block still below the deduplication level, so that it can be promoted": { + ranges: []int64{20, 40}, + shardCount: 2, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 2}}, + }, + expected: []*job{ + {userID: userID, stage: stageMerge, shardID: "1_of_2", blocksGroup: blocksGroup{ + rangeStart: 0, + rangeEnd: 20, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 2}}, + }, + }}, + }, + }, + "should not merge a lone sharded block that already reached the deduplication level": { + ranges: []int64{20, 40}, + shardCount: 2, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 3}}, + }, + expected: nil, + }, + "should not merge a lone sharded block with no compaction metadata": { + ranges: []int64{20, 40}, + shardCount: 2, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}}, + }, + expected: nil, + }, + "should not compact a group's most recent blocks prematurely just because another group has fresher blocks": { + ranges: []int64{10, 20}, + blocks: []*block.Meta{ + // Group "a" has not reached the end of the [20, 30) range yet. + {ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}}, + {ULID: block2, MinTime: 25, MaxTime: 28, Labels: map[string]string{"group": "a"}}, + // Group "b" is further ahead in time, but that says nothing about group "a". + {ULID: block3, MinTime: 40, MaxTime: 50, Labels: map[string]string{"group": "b"}}, + }, + expected: nil, + }, + "should compact a group whose range is closed, regardless of other groups": { + ranges: []int64{10, 20}, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}}, + {ULID: block2, MinTime: 25, MaxTime: 30, Labels: map[string]string{"group": "a"}}, + {ULID: block3, MinTime: 40, MaxTime: 50, Labels: map[string]string{"group": "b"}}, + }, + expected: []*job{ + {userID: userID, stage: stageMerge, blocksGroup: blocksGroup{ + rangeStart: 20, + rangeEnd: 30, + blocks: []*block.Meta{ + {ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}}, + {ULID: block2, MinTime: 25, MaxTime: 30, Labels: map[string]string{"group": "a"}}, + }, + }}, + }, + }, } for testName, testData := range tests { diff --git a/pkg/phlaredb/block/dedup.go b/pkg/phlaredb/block/dedup.go new file mode 100644 index 0000000000..ab329e425c --- /dev/null +++ b/pkg/phlaredb/block/dedup.go @@ -0,0 +1,38 @@ +package block + +// deduplicationLevelSharded and deduplicationLevelUnsharded are the compaction +// levels at or above which a block's contents are fully deduplicated. +// +// With split-and-merge sharding enabled, level 2 is the intermediate split +// stage: each group is split into split_shards parts which are not yet +// deduplicated, so the first authoritative level is 3. Without sharding, +// compaction deduplicates in a single step and level 2 is authoritative. +const ( + deduplicationLevelUnsharded int32 = 2 + deduplicationLevelSharded int32 = 3 +) + +// DeduplicationLevel returns the compaction level at or above which a block is +// authoritative for its time range: its contents are deduplicated and it fully +// replaces its ancestors. +// +// Both the compactor and the querier must agree on this value. The querier uses +// it to decide whether a block can be served as-is or has to be merged with +// deduplication; the compactor uses it to avoid leaving blocks in a state the +// querier will not serve. +func DeduplicationLevel(sharded bool) int32 { + if sharded { + return deduplicationLevelSharded + } + return deduplicationLevelUnsharded +} + +// IsIntermediate reports whether a block at the given compaction level is an +// intermediate compaction artefact rather than an authoritative block, i.e. it +// sits below the deduplication level. +// +// Intermediate blocks are still readable, but they may overlap with other +// blocks, so a query that includes one must deduplicate. +func IsIntermediate(level int32, sharded bool) bool { + return level < DeduplicationLevel(sharded) +} diff --git a/pkg/phlaredb/compact.go b/pkg/phlaredb/compact.go index 99b67492ba..f02b79874b 100644 --- a/pkg/phlaredb/compact.go +++ b/pkg/phlaredb/compact.go @@ -59,6 +59,12 @@ type CompactWithSplittingOpts struct { SplitBy SplitByFunc DownsamplerEnabled bool Logger log.Logger + + // AllowSingleBlock permits compacting a single source block without + // splitting it. That is normally a no-op and therefore rejected, but the + // compactor uses it to promote a block to the next compaction level when it + // has no sibling to merge with. + AllowSingleBlock bool } func Compact(ctx context.Context, src []BlockReader, dst string) (meta block.Meta, err error) { @@ -80,7 +86,7 @@ func Compact(ctx context.Context, src []BlockReader, dst string) (meta block.Met func CompactWithSplitting(ctx context.Context, opts CompactWithSplittingOpts) ( []block.Meta, error, ) { - if len(opts.Src) <= 1 && opts.SplitCount == 1 { + if len(opts.Src) == 0 || (len(opts.Src) == 1 && opts.SplitCount == 1 && !opts.AllowSingleBlock) { return nil, errors.New("not enough blocks to compact") } if opts.SplitCount == 0 { diff --git a/pkg/phlaredb/compact_test.go b/pkg/phlaredb/compact_test.go index 7f99a43f2f..943ec5c9a9 100644 --- a/pkg/phlaredb/compact_test.go +++ b/pkg/phlaredb/compact_test.go @@ -191,6 +191,57 @@ func TestCompactWithDownsampling(t *testing.T) { assert.True(t, querier.metrics.profileTableAccess.DeleteLabelValues("profiles.parquet")) } +// Compacting a single block on its own is normally a no-op, but the compactor +// relies on it to promote a block that has no sibling to merge with to the next +// compaction level. +func TestCompactWithSplitting_SingleBlock(t *testing.T) { + ctx := context.Background() + + b := newBlock(t, func() []*testhelper.ProfileBuilder { + return profileSeriesGenerator(t, time.Unix(1, 0), time.Unix(10, 0), time.Second, "job", "a") + }) + + t.Run("rejected by default", func(t *testing.T) { + _, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{ + Src: []BlockReader{b}, + Dst: t.TempDir(), + SplitCount: 1, + SplitBy: SplitByFingerprint, + Logger: log.NewNopLogger(), + }) + require.ErrorContains(t, err, "not enough blocks to compact") + }) + + t.Run("allowed explicitly", func(t *testing.T) { + compacted, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{ + Src: []BlockReader{b}, + Dst: t.TempDir(), + SplitCount: 1, + SplitBy: SplitByFingerprint, + AllowSingleBlock: true, + Logger: log.NewNopLogger(), + }) + require.NoError(t, err) + require.Len(t, compacted, 1) + require.Equal(t, b.Meta().MinTime, compacted[0].MinTime) + require.Equal(t, b.Meta().MaxTime, compacted[0].MaxTime) + // The point of the promotion: the output sits one level higher. + require.Equal(t, b.Meta().Compaction.Level+1, compacted[0].Compaction.Level) + }) + + t.Run("no source blocks is always rejected", func(t *testing.T) { + _, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{ + Src: nil, + Dst: t.TempDir(), + SplitCount: 1, + SplitBy: SplitByFingerprint, + AllowSingleBlock: true, + Logger: log.NewNopLogger(), + }) + require.ErrorContains(t, err, "not enough blocks to compact") + }) +} + func TestCompactWithSplitting(t *testing.T) { ctx := context.Background() diff --git a/pkg/querier/replication.go b/pkg/querier/replication.go index 41a7689d2d..c08c3881bc 100644 --- a/pkg/querier/replication.go +++ b/pkg/querier/replication.go @@ -17,6 +17,7 @@ import ( ingestv1 "github.com/grafana/pyroscope/api/gen/proto/go/ingester/v1" typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1" + "github.com/grafana/pyroscope/v2/pkg/phlaredb/block" "github.com/grafana/pyroscope/v2/pkg/phlaredb/sharding" "github.com/grafana/pyroscope/v2/pkg/util" "github.com/grafana/pyroscope/v2/pkg/util/spanlogger" @@ -216,10 +217,16 @@ func (r *replicasPerBlockID) hasShardedBlocks() bool { } // pruneIncompleteShardedBlocks drops sharded blocks for any window instant that -// is missing a shard. It must run before pruneSupersededBlocks, so an incomplete -// set's lower-level ancestors survive as a fallback. deduplicationLevel is the -// level at/above which blocks are deduplicated (3 when sharded). -func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int32) error { +// is missing a shard. Only blocks at or above minLevel are counted towards +// completeness. +// +// It runs twice. First with minLevel = deduplicationLevel, before +// pruneSupersededBlocks, so that an incomplete set of deduplicated blocks is +// pruned while its lower-level ancestors are still around to serve as a +// fallback. Then again with minLevel = 0, after pruneSupersededBlocks, to check +// that whatever survived the collapse - which may include intermediate blocks +// kept as a last resort - still covers every shard. +func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(minLevel int32) error { // Completeness is checked per sharding (shard count) and per time instant, not // by compaction level: a window's shards can legitimately sit at different // levels (a partial late re-merge advances only some), so keying on level would @@ -230,9 +237,11 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int // covering that instant. Coverage (minTime <= t < maxTime), rather than an exact // minTime match, keeps a wider block (a shard merged to a longer span) counted // for the later windows it overlaps, so sibling shards' blocks there are not - // orphaned and silently pruned. Only blocks >= deduplicationLevel count; - // intermediate lower blocks are never served (pruneSupersededBlocks drops them) - // so must not satisfy a shard. + // orphaned and silently pruned. Only blocks >= minLevel count: + // intermediate lower blocks are preferentially dropped by + // pruneSupersededBlocks, so they must not satisfy a shard here. They are also + // never pruned here, which is what lets pruneSupersededBlocks keep them as a + // last resort when their ancestors are gone. type shardedBlock struct { id string shard uint64 @@ -249,7 +258,7 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int if !ok { continue } - if meta.Compaction == nil || meta.Compaction.Level < deduplicationLevel { + if meta.Compaction == nil || meta.Compaction.Level < minLevel { continue } byShardCount[shardCount] = append(byShardCount[shardCount], @@ -320,37 +329,83 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int return nil } -// prunes blocks that are contained by a higher compaction level block -func (r *replicasPerBlockID) pruneSupersededBlocks(sharded bool) error { +// pruneSupersededBlocks removes blocks whose data is fully contained in a block +// at or above the deduplication level, and collapses intermediate compaction +// artefacts when a complete lower-level fallback is still available. +// +// deduplicationLevel is the level at/above which a block is authoritative for +// its range (see block.DeduplicationLevel). +func (r *replicasPerBlockID) pruneSupersededBlocks(deduplicationLevel int32) error { + // First pass: authoritative blocks supersede every ancestor they were built + // from, including any intermediate blocks recorded as parents. for blockID := range r.m { meta, ok := r.meta[blockID] if !ok { return fmt.Errorf("meta missing for block id %s", blockID) } - if meta.Compaction == nil { + if meta.Compaction == nil || meta.Compaction.Level < deduplicationLevel { continue } - if meta.Compaction.Level < 2 { + for _, ancestor := range meta.Compaction.Parents { + r.removeBlock(ancestor) + } + for _, ancestor := range meta.Compaction.Sources { + r.removeBlock(ancestor) + } + } + + // Second pass: intermediate blocks (compacted at least once, but still below + // the deduplication level – the split stage of split-and-merge compaction). + // Reading them is more expensive than reading their ancestors: the split + // stage fans a group out into shard_count blocks, which is typically + // _significantly_ more blocks than it consumed, and they are not yet + // deduplicated either way. So we prefer the ancestors – but only if they are + // all still there. If any of them is gone, the intermediate block is the only + // remaining copy of that data and dropping it loses the range entirely; keep + // it instead and let the query deduplicate. + for blockID := range r.m { + meta, ok := r.meta[blockID] + if !ok { + return fmt.Errorf("meta missing for block id %s", blockID) + } + // Level < 2 means the block has not been compacted yet: it has no + // ancestors to fall back to and is served as-is. + if meta.Compaction == nil || meta.Compaction.Level < 2 || meta.Compaction.Level >= deduplicationLevel { continue } - // At split phase of compaction, L2 is an intermediate step where we - // split each group into split_shards parts, thus there will be up to - // groups_num * split_shards blocks, which is typically _significantly_ - // greater that the number of source blocks. Moreover, these blocks are - // not yet deduplicated, therefore we should prefer L1 blocks over them. - // As an optimisation, we drop all L2 blocks. - if sharded && meta.Compaction.Level == 2 { + if r.ancestorsLive(meta) { r.removeBlock(blockID) continue } - for _, blockID := range meta.Compaction.Parents { - r.removeBlock(blockID) + level.Warn(r.logger).Log( + "msg", "querying intermediate compaction level block: its ancestors are no longer available", + "block", blockID, + "compaction_level", meta.Compaction.Level, + "deduplication_level", deduplicationLevel, + ) + } + + return nil +} + +// ancestorsLive reports whether the block's ancestors are all still present and +// can therefore serve its range in its place. A block with no recorded +// ancestors has no known fallback. +func (r *replicasPerBlockID) ancestorsLive(meta *typesv1.BlockInfo) bool { + if len(meta.Compaction.Sources) == 0 && len(meta.Compaction.Parents) == 0 { + return false + } + for _, ancestor := range meta.Compaction.Sources { + if _, ok := r.m[ancestor]; !ok { + return false } - for _, blockID := range meta.Compaction.Sources { - r.removeBlock(blockID) + } + for _, ancestor := range meta.Compaction.Parents { + if _, ok := r.m[ancestor]; !ok { + return false } } - return nil + return true } type blockPlanEntry struct { @@ -391,12 +446,9 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla sharded := r.hasShardedBlocks() // Depending on whether split sharding is used, the compaction level at - // which the data gets deduplicated differs: if split sharding is enabled, - // we deduplicate at level 3, and at level 2 otherwise. - var deduplicationLevel int32 = 2 - if sharded { - deduplicationLevel = 3 - } + // which the data gets deduplicated differs. The compactor uses the same + // threshold, so the two stay in agreement. + deduplicationLevel := block.DeduplicationLevel(sharded) // Prune incomplete sharded sets first, so that any lower-level ancestors // survive as a fallback, then collapse the surviving blocks per shard. @@ -405,11 +457,19 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla return nil } - if err := r.pruneSupersededBlocks(sharded); err != nil { + if err := r.pruneSupersededBlocks(deduplicationLevel); err != nil { level.Warn(r.logger).Log("msg", "block planning failed to prune superseded blocks", "err", err) return nil } + // Re-check completeness over what survived: pruneSupersededBlocks may have + // kept intermediate blocks that the first check did not count, and it may + // have removed ancestors that were covering a shard. + if err := r.pruneIncompleteShardedBlocks(0); err != nil { + level.Warn(r.logger).Log("msg", "block planning failed to prune incomplete sharded blocks", "err", err) + return nil + } + // now we go through all blocks and choose the replicas that we want to query for blockID, replicas := range r.m { // skip if we have no replicas, then block is already contained i an higher compaction level one diff --git a/pkg/querier/replication_test.go b/pkg/querier/replication_test.go index 43c256a4e9..d2e0f38cb3 100644 --- a/pkg/querier/replication_test.go +++ b/pkg/querier/replication_test.go @@ -476,10 +476,11 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { }, { // Mid-merge: shard 0 at L3, shard 1 only in an intermediate L2 block, no - // L1. The L2 block gets dropped by superseding, so it must not count shard - // 1 as present - otherwise we'd serve shard 0 alone (half the data). The - // set is incomplete -> everything pruned (transient empty, not undercount). - name: "do not let an intermediate L2 block satisfy shard completeness", + // L1. The L2 block must not count shard 1 as present at the deduplicated + // level - otherwise we'd serve shard 0 alone (half the data). The L3 block + // is therefore pruned, and the intermediate L2 blocks left behind do cover + // both shards, so they are served with deduplication. + name: "an intermediate L2 block does not satisfy shard completeness at the deduplicated level", inputs: func(r *replicasPerBlockID) { t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ @@ -507,6 +508,37 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { }, }, storeGatewayInstance) }, + validators: []validatorFunc{ + validatePlanBlockIDs("s0-l2", "s1-l2"), + validatePlanDeduplication(true), + }, + }, + { + // Same shape, but shard 0 has no intermediate block left behind: pruning + // the incomplete L3 set leaves shard 1 alone, which would be half the + // data, so everything is pruned (transient empty, not undercount). + name: "prune a surviving intermediate block that does not cover every shard", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0-l3"). + withCompactionLevel(3). + withCompactorShard(0, 2). + withMinTime(t1, time.Hour). + info(), + // shard 1 only exists as an intermediate L2 block. + newBlockInfo("s1-l2"). + withCompactionLevel(2). + withCompactorShard(1, 2). + withMinTime(t1, time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, validators: []validatorFunc{ validatePlanBlockIDs(), }, @@ -560,6 +592,121 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { validatePlanBlocksOnReplica("ingester-0", "b"), }, }, + { + // The split stage of compaction produces intermediate blocks that are + // preferentially dropped in favour of their ancestors. If the ancestors + // are gone, dropping them loses the range entirely, so they must be + // served (with deduplication) instead. + name: "keep intermediate sharded blocks whose ancestors are gone", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("s0", "s1"), + validatePlanDeduplication(true), + }, + }, + { + name: "drop intermediate sharded blocks whose ancestors are all still available", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("a").withMinTime(t1, 30*time.Minute).info(), + newBlockInfo("b").withMinTime(t1, time.Hour).info(), + + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("a", "b"), + validatePlanDeduplication(true), + }, + }, + { + // A block at or above the deduplication level supersedes everything it + // was built from, including the intermediate blocks recorded as parents. + name: "drop intermediate sharded blocks superseded by a deduplicated block", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + + newBlockInfo("m0"). + withMinTime(t1, time.Hour). + withCompactionLevel(3). + withCompactionSources("a", "b"). + withCompactionParents("s0"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("m1"). + withMinTime(t1, time.Hour). + withCompactionLevel(3). + withCompactionSources("a", "b"). + withCompactionParents("s1"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("m0", "m1"), + validatePlanDeduplication(false), + }, + }, } { t.Run(tc.name, func(t *testing.T) { r := newReplicasPerBlockID(log.NewNopLogger()) @@ -617,3 +764,53 @@ func Test_pruneIncompleteShardedBlocks_logging(t *testing.T) { require.NotContains(t, buf.String(), warnMsg, "must not warn for a complete window") }) } + +// Serving an intermediate compaction level block is a last resort: it means the +// compactor left the data in a state it never promoted out of, so it must be +// visible in the logs. +func Test_pruneSupersededBlocks_logging(t *testing.T) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + + const warnMsg = "its ancestors are no longer available" + + t.Run("warns when an intermediate block has to be served", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(0, 2).info(), + newBlockInfo("s1").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(1, 2).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.NotEmpty(t, plan) + require.Contains(t, buf.String(), warnMsg) + }) + + t.Run("does not warn when the ancestors are still available", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("a").withMinTime(t1, time.Hour).info(), + newBlockInfo("s0").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(0, 2).info(), + newBlockInfo("s1").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(1, 2).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.NotEmpty(t, plan) + require.NotContains(t, buf.String(), warnMsg) + }) +}