Repository: cortexproject/cortex
Pull request #5342 (2023-05-10 21:24:55)

Make sure partial block with only visit marker can be cleaned
**What this PR does**:
Partial block with only visit marker should be deleted even there is no deletion marker.

**Checklist**
- [X] Tests updated
- [ ] Documentation added
- [X] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f5cb0555f2..0ac212caa3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,7 @@
 * [BUGFIX] Alertmanager: Route web-ui requests to the alertmanager distributor when sharding is enabled. #5293
 * [BUGFIX] Storage: Bucket index updater should ignore meta not found for partial blocks. #5343
 * [BUGFIX] Ring: Add JOINING state to read operation. #5346
+* [BUGFIX] Compactor: Partial block with only visit marker should be deleted even there is no deletion marker. #5342
 
 ## 1.15.1 2023-04-26
 
diff --git a/pkg/compactor/block_visit_marker.go b/pkg/compactor/block_visit_marker.go
index 57cfd782d6..b391421fd3 100644
--- a/pkg/compactor/block_visit_marker.go
+++ b/pkg/compactor/block_visit_marker.go
@@ -30,6 +30,7 @@ const (
 var (
 	ErrorBlockVisitMarkerNotFound  = errors.New("block visit marker not found")
 	ErrorUnmarshalBlockVisitMarker = errors.New("unmarshal block visit marker JSON")
+	ErrorNotBlockVisitMarker       = errors.New("file is not block visit marker")
 )
 
 type BlockVisitMarker struct {
@@ -142,3 +143,11 @@ heartBeat:
 	}
 	level.Info(logger).Log("msg", fmt.Sprintf("stop heart beat for blocks: %s", blocksInfo))
 }
+
+func IsBlockVisitMarker(path string) bool {
+	return strings.HasSuffix(path, BlockVisitMarkerFile)
+}
+
+func IsNotBlockVisitMarkerError(err error) bool {
+	return errors.Is(err, ErrorNotBlockVisitMarker)
+}
diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go
index 59e7fd1c40..3259944987 100644
--- a/pkg/compactor/blocks_cleaner.go
+++ b/pkg/compactor/blocks_cleaner.go
@@ -419,9 +419,23 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map
 		// We can safely delete only partial blocks with a deletion mark.
 		err := metadata.ReadMarker(ctx, userLogger, userBucket, blockID.String(), &metadata.DeletionMark{})
 		if errors.Is(err, metadata.ErrorMarkerNotFound) {
-			return nil
-		}
-		if err != nil {
+			//If only visit marker exists in the block, we can safely delete it.
+			isEmpty := true
+			notVisitMarkerError := userBucket.ReaderWithExpectedErrs(IsNotBlockVisitMarkerError).Iter(ctx, blockID.String(), func(file string) error {
+				isEmpty = false
+				if !IsBlockVisitMarker(file) {
+					// return error here to fail iteration fast
+					// to avoid going through all files
+					return ErrorNotBlockVisitMarker
+				}
+				return nil
+			})
+			if isEmpty || notVisitMarkerError != nil {
+				// skip deleting partial block if block directory
+				// is empty or non visit marker file exists
+				return nil
+			}
+		} else if err != nil {
 			level.Warn(userLogger).Log("msg", "error reading partial block deletion mark", "block", blockID, "err", err)
 			return nil
 		}


Test patch:
diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go
index 8220b9286d..52f16d741a 100644
--- a/pkg/compactor/blocks_cleaner_test.go
+++ b/pkg/compactor/blocks_cleaner_test.go
@@ -79,11 +79,13 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 	block6 := createTSDBBlock(t, bucketClient, "user-1", 40, 50, nil)
 	block7 := createTSDBBlock(t, bucketClient, "user-2", 10, 20, nil)
 	block8 := createTSDBBlock(t, bucketClient, "user-2", 40, 50, nil)
+	block11 := ulid.MustNew(11, rand.Reader)
 	createDeletionMark(t, bucketClient, "user-1", block2, now.Add(-deletionDelay).Add(time.Hour))             // Block hasn't reached the deletion threshold yet.
 	createDeletionMark(t, bucketClient, "user-1", block3, now.Add(-deletionDelay).Add(-time.Hour))            // Block reached the deletion threshold.
 	createDeletionMark(t, bucketClient, "user-1", block4, now.Add(-deletionDelay).Add(time.Hour))             // Partial block hasn't reached the deletion threshold yet.
 	createDeletionMark(t, bucketClient, "user-1", block5, now.Add(-deletionDelay).Add(-time.Hour))            // Partial block reached the deletion threshold.
 	require.NoError(t, bucketClient.Delete(ctx, path.Join("user-1", block6.String(), metadata.MetaFilename))) // Partial block without deletion mark.
+	createBlockVisitMarker(t, bucketClient, "user-1", block11)                                                // Partial block only has visit marker.
 	createDeletionMark(t, bucketClient, "user-2", block7, now.Add(-deletionDelay).Add(-time.Hour))            // Block reached the deletion threshold.
 
 	// Blocks for user-3, marked for deletion.
@@ -147,6 +149,8 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		{path: path.Join("user-1", bucketindex.BlockDeletionMarkFilepath(block5)), expectedExists: false},
 		// Should not delete a partial block without deletion mark.
 		{path: path.Join("user-1", block6.String(), "index"), expectedExists: true},
+		// Should delete a partial block with only visit marker.
+		{path: path.Join("user-1", block11.String(), BlockVisitMarkerFile), expectedExists: false},
 		// Should completely delete blocks for user-3, marked for deletion
 		{path: path.Join("user-3", block9.String(), metadata.MetaFilename), expectedExists: false},
 		{path: path.Join("user-3", block9.String(), "index"), expectedExists: false},
@@ -166,7 +170,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 	assert.Equal(t, float64(1), testutil.ToFloat64(cleaner.runsStarted))
 	assert.Equal(t, float64(1), testutil.ToFloat64(cleaner.runsCompleted))
 	assert.Equal(t, float64(0), testutil.ToFloat64(cleaner.runsFailed))
-	assert.Equal(t, float64(6), testutil.ToFloat64(cleaner.blocksCleanedTotal))
+	assert.Equal(t, float64(7), testutil.ToFloat64(cleaner.blocksCleanedTotal))
 	assert.Equal(t, float64(0), testutil.ToFloat64(cleaner.blocksFailedTotal))
 
 	// Check the updated bucket index.
@@ -179,7 +183,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		{
 			userID:         "user-1",
 			expectedIndex:  true,
-			expectedBlocks: []ulid.ULID{block1, block2 /* deleted: block3, block4, block5, partial: block6 */},
+			expectedBlocks: []ulid.ULID{block1, block2 /* deleted: block3, block4, block5, block11, partial: block6 */},
 			expectedMarks:  []ulid.ULID{block2},
 		}, {
 			userID:         "user-2",
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 3810f7c511..c4622e4da7 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -1423,6 +1423,14 @@ func createNoCompactionMark(t *testing.T, bkt objstore.Bucket, userID string, bl
 	require.NoError(t, bkt.Upload(context.Background(), markPath, strings.NewReader(content)))
 }
 
+func createBlockVisitMarker(t *testing.T, bkt objstore.Bucket, userID string, blockID ulid.ULID) {
+	content := mockBlockVisitMarker()
+	blockPath := path.Join(userID, blockID.String())
+	markPath := path.Join(blockPath, BlockVisitMarkerFile)
+
+	require.NoError(t, bkt.Upload(context.Background(), markPath, strings.NewReader(content)))
+}
+
 func findCompactorByUserID(compactors []*Compactor, logs []*concurrency.SyncBuffer, userID string) (*Compactor, *concurrency.SyncBuffer, error) {
 	var compactor *Compactor
 	var log *concurrency.SyncBuffer
@@ -1680,6 +1688,21 @@ func mockBlockMetaJSONWithTime(id string, orgID string, minTime int64, maxTime i
 	return string(content)
 }
 
+func mockBlockVisitMarker() string {
+	blockVisitMarker := BlockVisitMarker{
+		CompactorID: "dummy",
+		VisitTime:   time.Now().Unix(),
+		Version:     1,
+	}
+
+	content, err := json.Marshal(blockVisitMarker)
+	if err != nil {
+		panic("failed to marshal mocked block visit marker")
+	}
+
+	return string(content)
+}
+
 func TestCompactor_DeleteLocalSyncFiles(t *testing.T) {
 	numUsers := 10