Merge pull request #8410 from thanos-io/partial_delete_marked

compact: ignore blocks with deletion mark in partial deletes

Author: GiedriusS

cmd/thanos/compact.go

@@ -442,7 +442,7 @@ func runCompact(
 			return errors.Wrap(err, "syncing metas")
 		}
 
-		compact.BestEffortCleanAbortedPartialUploads(ctx, logger, sy.Partial(), insBkt, compactMetrics.partialUploadDeleteAttempts, compactMetrics.blocksCleaned, compactMetrics.blockCleanupFailures)
+		compact.BestEffortCleanAbortedPartialUploads(ctx, logger, sy.Partial(), insBkt, compactMetrics.partialUploadDeleteAttempts, compactMetrics.blocksCleaned, compactMetrics.blockCleanupFailures, ignoreDeletionMarkFilter.DeletionMarkBlocks())
 		if err := blocksCleaner.DeleteMarkedBlocks(ctx); err != nil {
 			return errors.Wrap(err, "cleaning marked blocks")
 		}

cmd/thanos/tools_bucket.go

@@ -917,7 +917,7 @@ func registerBucketCleanup(app extkingpin.AppClause, objStoreConfig *extflag.Pat
 
 		level.Info(logger).Log("msg", "synced blocks done")
 
-		compact.BestEffortCleanAbortedPartialUploads(ctx, logger, sy.Partial(), insBkt, stubCounter, stubCounter, stubCounter)
+		compact.BestEffortCleanAbortedPartialUploads(ctx, logger, sy.Partial(), insBkt, stubCounter, stubCounter, stubCounter, ignoreDeletionMarkFilter.DeletionMarkBlocks())
 		if err := blocksCleaner.DeleteMarkedBlocks(ctx); err != nil {
 			return errors.Wrap(err, "error cleaning blocks")
 		}

pkg/compact/clean.go

@@ -17,6 +17,7 @@ import (
 	"github.com/thanos-io/objstore"
 
 	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
 )
 
 const (
@@ -61,10 +62,19 @@ func BestEffortCleanAbortedPartialUploads(
 	deleteAttempts prometheus.Counter,
 	blockCleanups prometheus.Counter,
 	blockCleanupFailures prometheus.Counter,
+	deletionMarkBlocks map[ulid.ULID]*metadata.DeletionMark,
 ) {
 	level.Info(logger).Log("msg", "started cleaning of aborted partial uploads")
 
 	for id := range partial {
+		// NOTE(GiedriusS): we start to delete blocks from meta.json so at that point they are marked as partial.
+		// If you have multiple compactor shards then they might try to delete the same block here.
+		// We do not want to delete blocks that are marked for deletion, as they are already scheduled for deletion in the blocks cleaner.
+		if _, ok := deletionMarkBlocks[id]; ok {
+			level.Debug(logger).Log("msg", "ignoring block marked for deletion", "block", id)
+			continue
+		}
+
 		lastModifiedTime, err := getOldestModifiedTime(ctx, id, bkt)
 		if err != nil {
 			level.Warn(logger).Log("msg", "failed to get last modified time for block; falling back to block creation time", "block", id, "err", err)

pkg/compact/clean_test.go

@@ -72,7 +72,7 @@ func TestBestEffortCleanAbortedPartialUploads(t *testing.T) {
 	_, partial, err := metaFetcher.Fetch(ctx)
 	testutil.Ok(t, err)
 
-	BestEffortCleanAbortedPartialUploads(ctx, logger, partial, bkt, deleteAttempts, blockCleanups, blockCleanupFailures)
+	BestEffortCleanAbortedPartialUploads(ctx, logger, partial, bkt, deleteAttempts, blockCleanups, blockCleanupFailures, map[ulid.ULID]*metadata.DeletionMark{})
 	testutil.Equals(t, 1.0, promtest.ToFloat64(deleteAttempts))
 	testutil.Equals(t, 1.0, promtest.ToFloat64(blockCleanups))
 	testutil.Equals(t, 0.0, promtest.ToFloat64(blockCleanupFailures))
@@ -88,6 +88,23 @@ func TestBestEffortCleanAbortedPartialUploads(t *testing.T) {
 	exists, err = bkt.Exists(ctx, path.Join(shouldIgnoreID2.String(), "chunks", "000001"))
 	testutil.Ok(t, err)
 	testutil.Equals(t, true, exists)
+
+	// It has been marked for deletion, so it should not be deleted.
+	shouldDeleteID, err = ulid.New(uint64(time.Now().Add(-PartialUploadThresholdAge-1*time.Hour).Unix()*1000), nil)
+	testutil.Ok(t, err)
+
+	fakeChunk.Reset()
+	fakeChunk.Write([]byte{0, 1, 2, 3})
+	testutil.Ok(t, bkt.Upload(ctx, path.Join(shouldDeleteID.String(), "chunks", "000001"), &fakeChunk))
+	testutil.Ok(t, mb.ChangeLastModified(path.Join(shouldDeleteID.String(), "chunks", "000001"), time.Now().Add(-PartialUploadThresholdAge-1*time.Hour)))
+
+	BestEffortCleanAbortedPartialUploads(ctx, logger, partial, bkt, deleteAttempts, blockCleanups, blockCleanupFailures, map[ulid.ULID]*metadata.DeletionMark{
+		shouldDeleteID: {},
+	})
+
+	exists, err = bkt.Exists(ctx, path.Join(shouldDeleteID.String(), "chunks", "000001"))
+	testutil.Ok(t, err)
+	testutil.Equals(t, true, exists)
 }
 
 func TestGetLastModifiedTime(t *testing.T) {