Optimize cleaner run time (#6815)

* optimize cleaner run time

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

* make old bucket.DeletePrefix concurrent

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

* added metric to record failure pushing to job channel. updated CHANGELOG. put more logs with timing values

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

---------

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

Author: alexqyle

CHANGELOG.md

@@ -40,6 +40,7 @@
 * [ENHANCEMENT] Compactor, Store Gateway: Introduce user scanner strategy and user index. #6780
 * [ENHANCEMENT] Querier: Support chunks cache for parquet queryable. #6805
 * [ENHANCEMENT] Parquet Storage: Add some metrics for parquet blocks and converter. #6809
+* [ENHANCEMENT] Compactor: Optimize cleaner run time. #6815
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

pkg/compactor/blocks_cleaner.go

@@ -82,6 +82,7 @@ type BlocksCleaner struct {
 	remainingPlannedCompactions       *prometheus.GaugeVec
 	inProgressCompactions             *prometheus.GaugeVec
 	oldestPartitionGroupOffset        *prometheus.GaugeVec
+	enqueueJobFailed                  *prometheus.CounterVec
 }
 
 func NewBlocksCleaner(
@@ -186,6 +187,10 @@ func NewBlocksCleaner(
 		remainingPlannedCompactions: remainingPlannedCompactions,
 		inProgressCompactions:       inProgressCompactions,
 		oldestPartitionGroupOffset:  oldestPartitionGroupOffset,
+		enqueueJobFailed: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_compactor_enqueue_cleaner_job_failed_total",
+			Help: "Total number of cleaner jobs failed to be enqueued.",
+		}, []string{"user_status"}),
 	}
 
 	c.Service = services.NewBasicService(c.starting, c.loop, nil)
@@ -243,13 +248,25 @@ func (c *BlocksCleaner) loop(ctx context.Context) error {
 				continue
 			}
 			cleanJobTimestamp := time.Now().Unix()
-			usersChan <- &cleanerJob{
+
+			select {
+			case usersChan <- &cleanerJob{
 				users:     activeUsers,
 				timestamp: cleanJobTimestamp,
+			}:
+			default:
+				level.Warn(c.logger).Log("msg", "unable to push cleaning job to usersChan")
+				c.enqueueJobFailed.WithLabelValues(activeStatus).Inc()
 			}
-			deleteChan <- &cleanerJob{
+
+			select {
+			case deleteChan <- &cleanerJob{
 				users:     deletedUsers,
 				timestamp: cleanJobTimestamp,
+			}:
+			default:
+				level.Warn(c.logger).Log("msg", "unable to push deletion job to deleteChan")
+				c.enqueueJobFailed.WithLabelValues(deletedStatus).Inc()
 			}
 
 		case <-ctx.Done():
@@ -392,10 +409,18 @@ func (c *BlocksCleaner) obtainVisitMarkerManager(ctx context.Context, userLogger
 }
 
 // Remove blocks and remaining data for tenant marked for deletion.
-func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) error {
-
+func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) (returnErr error) {
+	startTime := time.Now()
 	level.Info(userLogger).Log("msg", "deleting blocks for tenant marked for deletion")
+	defer func() {
+		if returnErr != nil {
+			level.Warn(userLogger).Log("msg", "failed deleting tenant marked for deletion", "err", returnErr)
+		} else {
+			level.Info(userLogger).Log("msg", "completed deleting tenant marked for deletion", "duration", time.Since(startTime), "duration_ms", time.Since(startTime).Milliseconds())
+		}
+	}()
 
+	begin := time.Now()
 	// We immediately delete the bucket index, to signal to its consumers that
 	// the tenant has "no blocks" in the storage.
 	if err := bucketindex.DeleteIndex(ctx, c.bucketClient, userID, c.cfgProvider); err != nil {
@@ -465,6 +490,7 @@ func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLog
 	if deletedBlocks.Load() > 0 {
 		level.Info(userLogger).Log("msg", "deleted blocks for tenant marked for deletion", "deletedBlocks", deletedBlocks.Load())
 	}
+	level.Info(userLogger).Log("msg", "completed deleting blocks for tenant marked for deletion", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 
 	mark, err := cortex_tsdb.ReadTenantDeletionMark(ctx, c.bucketClient, userID)
 	if err != nil {
@@ -491,10 +517,11 @@ func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userLog
 		return err
 	}
 
-	if deleted, err := bucket.DeletePrefix(ctx, userBucket, bucketindex.MarkersPathname, userLogger); err != nil {
+	begin = time.Now()
+	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)
+		level.Info(userLogger).Log("msg", "deleted marker files for tenant marked for deletion", "count", deleted, "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 	}
 	if err := cortex_tsdb.DeleteTenantDeletionMark(ctx, c.bucketClient, userID); err != nil {
 		return errors.Wrap(err, "failed to delete tenant deletion mark")
@@ -503,18 +530,20 @@ 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.DeletePrefix(ctx, userBucket, block.DebugMetas, userLogger); err != nil {
+	begin := time.Now()
+	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)
+		level.Info(userLogger).Log("msg", "deleted files under "+block.DebugMetas+" for tenant marked for deletion", "count", deleted, "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 	}
 
 	if c.cfg.CompactionStrategy == util.CompactionStrategyPartitioning {
+		begin = time.Now()
 		// Clean up partitioned group info files
-		if deleted, err := bucket.DeletePrefix(ctx, userBucket, PartitionedGroupDirectory, userLogger); 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)
+			level.Info(userLogger).Log("msg", "deleted files under "+PartitionedGroupDirectory+" for tenant marked for deletion", "count", deleted, "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 		}
 	}
 	return nil
@@ -531,7 +560,7 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userLogger log.Logger, us
 		if returnErr != nil {
 			level.Warn(userLogger).Log("msg", "failed blocks cleanup and maintenance", "err", returnErr)
 		} else {
-			level.Info(userLogger).Log("msg", "completed blocks cleanup and maintenance", "duration", time.Since(startTime))
+			level.Info(userLogger).Log("msg", "completed blocks cleanup and maintenance", "duration", time.Since(startTime), "duration_ms", time.Since(startTime).Milliseconds())
 		}
 		c.tenantCleanDuration.WithLabelValues(userID).Set(time.Since(startTime).Seconds())
 	}()
@@ -771,7 +800,7 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 
 		if extraInfo.status.CanDelete || extraInfo.status.DeleteVisitMarker {
 			// Remove partition visit markers
-			if _, err := bucket.DeletePrefix(ctx, userBucket, GetPartitionVisitMarkerDirectoryPath(partitionedGroupInfo.PartitionedGroupID), userLogger); 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

@@ -7,27 +7,45 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/thanos-io/objstore"
+	"go.uber.org/atomic"
+
+	"github.com/cortexproject/cortex/pkg/util/concurrency"
 )
 
 // 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) {
-	result := 0
-	err := bkt.Iter(ctx, prefix, func(name string) error {
-		if strings.HasSuffix(name, objstore.DirDelim) {
-			deleted, err := DeletePrefix(ctx, bkt, name, logger)
-			result += deleted
-			return err
-		}
+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
+	}
 
+	result := atomic.NewInt32(0)
+	err = concurrency.ForEach(ctx, concurrency.CreateJobsFromStrings(keys), maxConcurrency, func(ctx context.Context, key interface{}) error {
+		name := key.(string)
 		if err := bkt.Delete(ctx, name); err != nil {
 			return err
 		}
-		result++
+		result.Inc()
 		level.Debug(logger).Log("msg", "deleted file", "file", name)
 		return nil
 	})
 
-	return result, err
+	return int(result.Load()), err
+}
+
+func ListPrefixes(ctx context.Context, bkt objstore.Bucket, prefix string, logger log.Logger) ([]string, error) {
+	var keys []string
+	err := bkt.Iter(ctx, prefix, func(name string) error {
+		if strings.HasSuffix(name, objstore.DirDelim) {
+			moreKeys, err := ListPrefixes(ctx, bkt, name, logger)
+			keys = append(keys, moreKeys...)
+			return err
+		}
+
+		keys = append(keys, name)
+		return nil
+	})
+	return keys, err
 }

pkg/storage/bucket/bucket_util_test.go

@@ -2,6 +2,7 @@ package bucket
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -21,8 +22,28 @@ 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()))
 }
+
+func TestDeletePrefixConcurrent(t *testing.T) {
+	mem := objstore.NewInMemBucket()
+
+	require.NoError(t, mem.Upload(context.Background(), "obj", strings.NewReader("hello")))
+	require.NoError(t, mem.Upload(context.Background(), "prefix/1", strings.NewReader("hello")))
+	require.NoError(t, mem.Upload(context.Background(), "prefix/2", strings.NewReader("hello")))
+	require.NoError(t, mem.Upload(context.Background(), "prefix/sub1/3", strings.NewReader("hello")))
+	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")))
+	n := 10000
+	for i := 0; i < n; i++ {
+		require.NoError(t, mem.Upload(context.Background(), fmt.Sprintf("prefix/sub/%d", i), strings.NewReader(fmt.Sprintf("hello%d", i))))
+	}
+
+	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()))
+}