diff --git a/CHANGELOG.md b/CHANGELOG.md index 3497bb1a6ea..fdbebb5038d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ * [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 * [BUGFIX] Ingester: Staleness markers written by the Ruler are no longer reported as discarded samples for the tenant. #7887 +* [BUGFIX] Compactor: Delete block deletion markers left in the global `markers/` location of a tenant after the block they refer to was removed. They were skipped at every bucket index update and kept the tenant location in the bucket forever. #7890 ## 1.22.0 in progress * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792 diff --git a/pkg/storage/tsdb/bucketindex/updater.go b/pkg/storage/tsdb/bucketindex/updater.go index e917a53fb25..40c56d80f32 100644 --- a/pkg/storage/tsdb/bucketindex/updater.go +++ b/pkg/storage/tsdb/bucketindex/updater.go @@ -254,6 +254,7 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark if errors.Is(err, ErrBlockDeletionMarkNotFound) { // This could happen if the block is permanently deleted between the "list objects" and now. level.Warn(w.logger).Log("msg", "skipped missing block deletion mark when updating bucket index", "block", id.String()) + w.deleteOrphanedGlobalDeletionMark(ctx, id) continue } if errors.Is(err, ErrBlockDeletionMarkCorrupted) { @@ -270,6 +271,35 @@ func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark return out, deletedBlocks, totalBlocksBlocksMarkedForNoCompaction, nil } +// errStopIteration is used to stop a bucket iteration as soon as the first object is found. +var errStopIteration = errors.New("stop iteration") + +// deleteOrphanedGlobalDeletionMark removes the block deletion mark from the global markers location +// when the block it refers to no longer exists in the storage. Without this, the stale marker would +// be listed (and skipped) at every update of the bucket index, and would keep the tenant's location in +// the storage forever. The marker is removed only if the block location is empty, so that a block +// which is being marked for deletion is never affected. This is best effort: failures are only logged. +func (w *Updater) deleteOrphanedGlobalDeletionMark(ctx context.Context, id ulid.ULID) { + hasObjects := false + err := w.bkt.Iter(ctx, id.String()+objstore.DirDelim, func(string) error { + hasObjects = true + return errStopIteration + }, objstore.WithRecursiveIter()) + if err != nil && !errors.Is(err, errStopIteration) { + level.Warn(w.logger).Log("msg", "failed to check whether the block of an orphaned deletion mark exists", "block", id.String(), "err", err) + return + } + if hasObjects { + return + } + + if err := w.bkt.Delete(ctx, BlockDeletionMarkFilepath(id)); err != nil && !w.bkt.IsObjNotFoundErr(err) { + level.Warn(w.logger).Log("msg", "failed to delete orphaned block deletion mark", "block", id.String(), "err", err) + return + } + level.Info(w.logger).Log("msg", "deleted orphaned block deletion mark of a block which no longer exists", "block", id.String()) +} + func (w *Updater) updateBlockDeletionMarkIndexEntry(ctx context.Context, id ulid.ULID) (*BlockDeletionMark, error) { m := metadata.DeletionMark{} diff --git a/pkg/storage/tsdb/bucketindex/updater_test.go b/pkg/storage/tsdb/bucketindex/updater_test.go index 1b1116cf32d..b57423abfa6 100644 --- a/pkg/storage/tsdb/bucketindex/updater_test.go +++ b/pkg/storage/tsdb/bucketindex/updater_test.go @@ -318,6 +318,48 @@ func TestUpdater_UpdateIndex_ShouldSkipBlockMarkedForDeletionWithMissingGlobalMa assert.Empty(t, nonCompactBlocks) } +func TestUpdater_UpdateIndex_ShouldDeleteOrphanedGlobalDeletionMarks(t *testing.T) { + const userID = "user-1" + + bkt, _ := testutil.PrepareFilesystemBucket(t) + + ctx := context.Background() + logger := log.NewNopLogger() + + bkt = BucketWithGlobalMarkers(bkt) + block1 := testutil.MockStorageBlock(t, bkt, userID, 10, 20) + block2 := testutil.MockStorageBlock(t, bkt, userID, 20, 30) + testutil.MockStorageDeletionMark(t, bkt, userID, block2) + + // Orphaned marker: the block does not exist at all. + orphanID := ulid.MustNew(ulid.Now()+1, nil) + mark, err := json.Marshal(metadata.DeletionMark{ID: orphanID, Version: metadata.DeletionMarkVersion1, DeletionTime: time.Now().Add(-24 * time.Hour).Unix()}) + require.NoError(t, err) + orphanPath := path.Join(userID, BlockDeletionMarkFilepath(orphanID)) + require.NoError(t, bkt.Upload(ctx, orphanPath, bytes.NewReader(mark))) + + // Global marker of a block which still has files but no local marker is left untouched. + partialID := ulid.MustNew(ulid.Now()+2, nil) + partialPath := path.Join(userID, BlockDeletionMarkFilepath(partialID)) + require.NoError(t, bkt.Upload(ctx, partialPath, bytes.NewReader(mark))) + require.NoError(t, bkt.Upload(ctx, path.Join(userID, partialID.String(), "index"), strings.NewReader("x"))) + + w := NewUpdater(bkt, userID, nil, logger) + idx, _, _, err := w.UpdateIndex(ctx, nil) + require.NoError(t, err) + assertBucketIndexEqual(t, idx, bkt, userID, + []tsdb.BlockMeta{block1, block2}, + []*metadata.DeletionMark{{ID: block2.ULID, Version: metadata.DeletionMarkVersion1, DeletionTime: idx.BlockDeletionMarks[0].DeletionTime}}) + + exists, err := bkt.Exists(ctx, orphanPath) + require.NoError(t, err) + assert.False(t, exists) + + exists, err = bkt.Exists(ctx, partialPath) + require.NoError(t, err) + assert.True(t, exists) +} + func TestUpdater_UpdateIndex_NoTenantInTheBucket(t *testing.T) { const userID = "user-1"