make old bucket.DeletePrefix concurrent

Signed-off-by: Alex Le <[email protected]>

Author: alexqyle

pkg/compactor/blocks_cleaner.go

@@ -501,7 +501,7 @@ func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLog
 		return err
 	}
 
-	if deleted, err := bucket.DeletePrefixConcurrent(ctx, userBucket, bucketindex.MarkersPathname, userLogger, defaultDeleteBlocksConcurrency); err != nil {
+	if deleted, err := bucket.DeletePrefix(ctx, userBucket, bucketindex.MarkersPathname, userLogger, defaultDeleteBlocksConcurrency); err != nil {
 		return errors.Wrap(err, "failed to delete marker files")
 	} else if deleted > 0 {
 		level.Info(userLogger).Log("msg", "deleted marker files for tenant marked for deletion", "count", deleted)
@@ -513,15 +513,15 @@ func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLog
 }
 
 func (c *BlocksCleaner) deleteNonDataFiles(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket) error {
-	if deleted, err := bucket.DeletePrefixConcurrent(ctx, userBucket, block.DebugMetas, userLogger, defaultDeleteBlocksConcurrency); err != nil {
+	if deleted, err := bucket.DeletePrefix(ctx, userBucket, block.DebugMetas, userLogger, defaultDeleteBlocksConcurrency); err != nil {
 		return errors.Wrap(err, "failed to delete "+block.DebugMetas)
 	} else if deleted > 0 {
 		level.Info(userLogger).Log("msg", "deleted files under "+block.DebugMetas+" for tenant marked for deletion", "count", deleted)
 	}
 
 	if c.cfg.CompactionStrategy == util.CompactionStrategyPartitioning {
 		// Clean up partitioned group info files
-		if deleted, err := bucket.DeletePrefixConcurrent(ctx, userBucket, PartitionedGroupDirectory, userLogger, defaultDeleteBlocksConcurrency); err != nil {
+		if deleted, err := bucket.DeletePrefix(ctx, userBucket, PartitionedGroupDirectory, userLogger, defaultDeleteBlocksConcurrency); err != nil {
 			return errors.Wrap(err, "failed to delete "+PartitionedGroupDirectory)
 		} else if deleted > 0 {
 			level.Info(userLogger).Log("msg", "deleted files under "+PartitionedGroupDirectory+" for tenant marked for deletion", "count", deleted)
@@ -781,7 +781,7 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 
 		if extraInfo.status.CanDelete || extraInfo.status.DeleteVisitMarker {
 			// Remove partition visit markers
-			if _, err := bucket.DeletePrefixConcurrent(ctx, userBucket, GetPartitionVisitMarkerDirectoryPath(partitionedGroupInfo.PartitionedGroupID), userLogger, defaultDeleteBlocksConcurrency); err != nil {
+			if _, err := bucket.DeletePrefix(ctx, userBucket, GetPartitionVisitMarkerDirectoryPath(partitionedGroupInfo.PartitionedGroupID), userLogger, defaultDeleteBlocksConcurrency); err != nil {
 				level.Warn(userLogger).Log("msg", "failed to delete partition visit markers for partitioned group", "partitioned_group_info", partitionedGroupInfoFile, "err", err)
 			} else {
 				level.Info(userLogger).Log("msg", "deleted partition visit markers for partitioned group", "partitioned_group_info", partitionedGroupInfoFile)

pkg/storage/bucket/bucket_util.go

@@ -15,11 +15,7 @@ import (
 // DeletePrefix removes all objects with given prefix, recursively.
 // It returns number of deleted objects.
 // If deletion of any object fails, it returns error and stops.
-func DeletePrefix(ctx context.Context, bkt objstore.Bucket, prefix string, logger log.Logger) (int, error) {
-	return DeletePrefixConcurrent(ctx, bkt, prefix, logger, 1)
-}
-
-func DeletePrefixConcurrent(ctx context.Context, bkt objstore.Bucket, prefix string, logger log.Logger, maxConcurrency int) (int, error) {
+func DeletePrefix(ctx context.Context, bkt objstore.Bucket, prefix string, logger log.Logger, maxConcurrency int) (int, error) {
 	keys, err := ListPrefixes(ctx, bkt, prefix, logger)
 	if err != nil {
 		return 0, err

pkg/storage/bucket/bucket_util_test.go

@@ -22,7 +22,7 @@ func TestDeletePrefix(t *testing.T) {
 	require.NoError(t, mem.Upload(context.Background(), "prefix/sub2/4", strings.NewReader("hello")))
 	require.NoError(t, mem.Upload(context.Background(), "outside/obj", strings.NewReader("hello")))
 
-	del, err := DeletePrefix(context.Background(), mem, "prefix", log.NewNopLogger())
+	del, err := DeletePrefix(context.Background(), mem, "prefix", log.NewNopLogger(), 1)
 	require.NoError(t, err)
 	assert.Equal(t, 4, del)
 	assert.Equal(t, 2, len(mem.Objects()))
@@ -42,7 +42,7 @@ func TestDeletePrefixConcurrent(t *testing.T) {
 		require.NoError(t, mem.Upload(context.Background(), fmt.Sprintf("prefix/sub/%d", i), strings.NewReader(fmt.Sprintf("hello%d", i))))
 	}
 
-	del, err := DeletePrefixConcurrent(context.Background(), mem, "prefix", log.NewNopLogger(), 100)
+	del, err := DeletePrefix(context.Background(), mem, "prefix", log.NewNopLogger(), 100)
 	require.NoError(t, err)
 	assert.Equal(t, 4+n, del)
 	assert.Equal(t, 2, len(mem.Objects()))