Compactor: delete orphaned global block deletion marks

CHANGELOG.md

@@ -10,6 +10,7 @@
 * [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
+* [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. #7885
 
 ## 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

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{}
 

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"