Let blocks_cleaner delete blocks concurrently

Signed-off-by: wangguoliang <[email protected]>

Author: wgliang

CHANGELOG.md

@@ -5,6 +5,7 @@
 * [ENHANCEMENT] Querier: limit series query to only ingesters if `start` param is not specified. #4976
 * [ENHANCEMENT] Query-frontend/scheduler: add a new limit `frontend.max-outstanding-requests-per-tenant` for configuring queue size per tenant. Started deprecating two flags `-query-scheduler.max-outstanding-requests-per-tenant` and `-querier.max-outstanding-requests-per-tenant`, and change their value default to 0. Now if both the old flag and new flag are specified, the old flag's queue size will be picked. #5005
 * [ENHANCEMENT] Query-tee: Add `/api/v1/query_exemplars` API endpoint support. #5010
+* [ENHANCEMENT] Let blocks_cleaner delete blocks concurrently. And added `-compactor.blocks-delete-concurrency` allowing to configure number of go routines for deleting blocks during compaction. #5028
 * [FEATURE] Querier/Query Frontend: support Prometheus /api/v1/status/buildinfo API. #4978
 * [FEATURE] Ingester: Add active series to all_user_stats page. #4972
 * [FEATURE] Ingester: Added `-blocks-storage.tsdb.head-chunks-write-queue-size` allowing to configure the size of the in-memory queue used before flushing chunks to the disk . #5000

docs/blocks-storage/compactor.md

@@ -162,6 +162,11 @@ compactor:
   # CLI flag: -compactor.blocks-fetch-concurrency
   [blocks_fetch_concurrency: <int> | default = 3]
 
+  # Number of goroutines to use when deleting blocks from object storage when
+  # compacting.
+  # CLI flag: -compactor.blocks-delete-concurrency
+  [blocks_delete_concurrency: <int> | default = 16]
+
   # When enabled, at compactor startup the bucket will be scanned and all found
   # deletion marks inside the block location will be copied to the markers
   # global location too. This option can (and should) be safely disabled as soon

docs/configuration/config-file-reference.md

@@ -3811,6 +3811,11 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.blocks-fetch-concurrency
 [blocks_fetch_concurrency: <int> | default = 3]
 
+# Number of goroutines to use when deleting blocks from object storage when
+# compacting.
+# CLI flag: -compactor.blocks-delete-concurrency
+[blocks_delete_concurrency: <int> | default = 16]
+
 # When enabled, at compactor startup the bucket will be scanned and all found
 # deletion marks inside the block location will be copied to the markers global
 # location too. This option can (and should) be safely disabled as soon as the

pkg/compactor/blocks_cleaner.go

@@ -3,6 +3,7 @@ package compactor
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/go-kit/log"
@@ -24,12 +25,17 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
+const (
+	defaultDeleteBlocksConcurrency = 16
+)
+
 type BlocksCleanerConfig struct {
 	DeletionDelay                      time.Duration
 	CleanupInterval                    time.Duration
 	CleanupConcurrency                 int
 	BlockDeletionMarksMigrationEnabled bool          // TODO Discuss whether we should remove it in Cortex 1.8.0 and document that upgrading to 1.7.0 before 1.8.0 is required.
 	TenantCleanupDelay                 time.Duration // Delay before removing tenant deletion mark and "debug".
+	DeleteBlocksConcurrency            int
 }
 
 type BlocksCleaner struct {
@@ -344,23 +350,34 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b
 
 	// Delete blocks marked for deletion. We iterate over a copy of deletion marks because
 	// we'll need to manipulate the index (removing blocks which get deleted).
+	blocksToDelete := make([]interface{}, 0, len(idx.BlockDeletionMarks))
+	var mux sync.Mutex
 	for _, mark := range idx.BlockDeletionMarks.Clone() {
 		if time.Since(mark.GetDeletionTime()).Seconds() <= c.cfg.DeletionDelay.Seconds() {
 			continue
 		}
+		blocksToDelete = append(blocksToDelete, mark.ID)
+	}
 
-		if err := block.Delete(ctx, userLogger, userBucket, mark.ID); err != nil {
+	// Concurrently deletes blocks marked for deletion, and removes blocks from index.
+	_ = concurrency.ForEach(ctx, blocksToDelete, c.cfg.DeleteBlocksConcurrency, func(ctx context.Context, job interface{}) error {
+		blockID := job.(ulid.ULID)
+
+		if err := block.Delete(ctx, userLogger, userBucket, blockID); err != nil {
 			c.blocksFailedTotal.Inc()
-			level.Warn(userLogger).Log("msg", "failed to delete block marked for deletion", "block", mark.ID, "err", err)
-			continue
+			level.Warn(userLogger).Log("msg", "failed to delete block marked for deletion", "block", blockID, "err", err)
+			return nil
 		}
 
 		// Remove the block from the bucket index too.
-		idx.RemoveBlock(mark.ID)
+		mux.Lock()
+		idx.RemoveBlock(blockID)
+		mux.Unlock()
 
 		c.blocksCleanedTotal.Inc()
-		level.Info(userLogger).Log("msg", "deleted block marked for deletion", "block", mark.ID)
-	}
+		level.Info(userLogger).Log("msg", "deleted block marked for deletion", "block", blockID)
+		return nil
+	})
 
 	// Partial blocks with a deletion mark can be cleaned up. This is a best effort, so we don't return
 	// error if the cleanup of partial blocks fail.
@@ -383,39 +400,51 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b
 }
 
 // cleanUserPartialBlocks delete partial blocks which are safe to be deleted. The provided partials map
-// is updated accordingly.
+// and index are updated accordingly.
 func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map[ulid.ULID]error, idx *bucketindex.Index, userBucket objstore.InstrumentedBucket, userLogger log.Logger) {
+	// Collect all blocks with missing meta.json into buffered channel.
+	blocks := make([]interface{}, 0, len(partials))
+
 	for blockID, blockErr := range partials {
 		// We can safely delete only blocks which are partial because the meta.json is missing.
 		if !errors.Is(blockErr, bucketindex.ErrBlockMetaNotFound) {
 			continue
 		}
+		blocks = append(blocks, blockID)
+	}
+
+	var mux sync.Mutex
 
+	_ = concurrency.ForEach(ctx, blocks, c.cfg.DeleteBlocksConcurrency, func(ctx context.Context, job interface{}) error {
+		blockID := job.(ulid.ULID)
 		// 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) {
-			continue
+			return nil
 		}
 		if err != nil {
 			level.Warn(userLogger).Log("msg", "error reading partial block deletion mark", "block", blockID, "err", err)
-			continue
+			return nil
 		}
 
 		// Hard-delete partial blocks having a deletion mark, even if the deletion threshold has not
 		// been reached yet.
 		if err := block.Delete(ctx, userLogger, userBucket, blockID); err != nil {
 			c.blocksFailedTotal.Inc()
 			level.Warn(userLogger).Log("msg", "error deleting partial block marked for deletion", "block", blockID, "err", err)
-			continue
+			return nil
 		}
 
 		// Remove the block from the bucket index too.
+		mux.Lock()
 		idx.RemoveBlock(blockID)
 		delete(partials, blockID)
+		mux.Unlock()
 
 		c.blocksCleanedTotal.Inc()
 		level.Info(userLogger).Log("msg", "deleted partial block marked for deletion", "block", blockID)
-	}
+		return nil
+	})
 }
 
 // applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period.

pkg/compactor/blocks_cleaner_test.go

@@ -115,6 +115,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		CleanupConcurrency:                 options.concurrency,
 		BlockDeletionMarksMigrationEnabled: options.markersMigrationEnabled,
 		TenantCleanupDelay:                 options.tenantDeletionDelay,
+		DeleteBlocksConcurrency:            defaultDeleteBlocksConcurrency,
 	}
 
 	reg := prometheus.NewPedanticRegistry()
@@ -256,9 +257,10 @@ func TestBlocksCleaner_ShouldContinueOnBlockDeletionFailure(t *testing.T) {
 	}
 
 	cfg := BlocksCleanerConfig{
-		DeletionDelay:      deletionDelay,
-		CleanupInterval:    time.Minute,
-		CleanupConcurrency: 1,
+		DeletionDelay:           deletionDelay,
+		CleanupInterval:         time.Minute,
+		CleanupConcurrency:      1,
+		DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency,
 	}
 
 	logger := log.NewNopLogger()
@@ -316,9 +318,10 @@ func TestBlocksCleaner_ShouldRebuildBucketIndexOnCorruptedOne(t *testing.T) {
 	require.NoError(t, bucketClient.Upload(ctx, path.Join(userID, bucketindex.IndexCompressedFilename), strings.NewReader("invalid!}")))
 
 	cfg := BlocksCleanerConfig{
-		DeletionDelay:      deletionDelay,
-		CleanupInterval:    time.Minute,
-		CleanupConcurrency: 1,
+		DeletionDelay:           deletionDelay,
+		CleanupInterval:         time.Minute,
+		CleanupConcurrency:      1,
+		DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency,
 	}
 
 	logger := log.NewNopLogger()
@@ -365,9 +368,10 @@ func TestBlocksCleaner_ShouldRemoveMetricsForTenantsNotBelongingAnymoreToTheShar
 	createTSDBBlock(t, bucketClient, "user-2", 30, 40, nil)
 
 	cfg := BlocksCleanerConfig{
-		DeletionDelay:      time.Hour,
-		CleanupInterval:    time.Minute,
-		CleanupConcurrency: 1,
+		DeletionDelay:           time.Hour,
+		CleanupInterval:         time.Minute,
+		CleanupConcurrency:      1,
+		DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency,
 	}
 
 	ctx := context.Background()
@@ -496,9 +500,10 @@ func TestBlocksCleaner_ShouldRemoveBlocksOutsideRetentionPeriod(t *testing.T) {
 	block4 := createTSDBBlock(t, bucketClient, "user-2", ts(-8), ts(-6), nil)
 
 	cfg := BlocksCleanerConfig{
-		DeletionDelay:      time.Hour,
-		CleanupInterval:    time.Minute,
-		CleanupConcurrency: 1,
+		DeletionDelay:           time.Hour,
+		CleanupInterval:         time.Minute,
+		CleanupConcurrency:      1,
+		DeleteBlocksConcurrency: defaultDeleteBlocksConcurrency,
 	}
 
 	ctx := context.Background()

pkg/compactor/compactor.go

@@ -181,6 +181,7 @@ type Config struct {
 	SkipBlocksWithOutOfOrderChunksEnabled bool                     `yaml:"skip_blocks_with_out_of_order_chunks_enabled"`
 	BlockFilesConcurrency                 int                      `yaml:"block_files_concurrency"`
 	BlocksFetchConcurrency                int                      `yaml:"blocks_fetch_concurrency"`
+	BlocksDeleteConcurrency               int                      `yaml:"blocks_delete_concurrency"`
 
 	// Whether the migration of block deletion marks to the global markers location is enabled.
 	BlockDeletionMarksMigrationEnabled bool `yaml:"block_deletion_marks_migration_enabled"`
@@ -236,6 +237,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.SkipBlocksWithOutOfOrderChunksEnabled, "compactor.skip-blocks-with-out-of-order-chunks-enabled", false, "When enabled, mark blocks containing index with out-of-order chunks for no compact instead of halting the compaction.")
 	f.IntVar(&cfg.BlockFilesConcurrency, "compactor.block-files-concurrency", 10, "Number of goroutines to use when fetching/uploading block files from object storage.")
 	f.IntVar(&cfg.BlocksFetchConcurrency, "compactor.blocks-fetch-concurrency", 3, "Number of goroutines to use when fetching blocks from object storage when compacting.")
+	f.IntVar(&cfg.BlocksDeleteConcurrency, "compactor.blocks-delete-concurrency", 16, "Number of goroutines to use when deleting blocks from object storage when compacting.")
 
 	f.Var(&cfg.EnabledTenants, "compactor.enabled-tenants", "Comma separated list of tenants that can be compacted. If specified, only these tenants will be compacted by compactor, otherwise all tenants can be compacted. Subject to sharding.")
 	f.Var(&cfg.DisabledTenants, "compactor.disabled-tenants", "Comma separated list of tenants that cannot be compacted by this compactor. If specified, and compactor would normally pick given tenant for compaction (via -compactor.enabled-tenants or sharding), it will be ignored instead.")
@@ -505,6 +507,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 		CleanupConcurrency:                 c.compactorCfg.CleanupConcurrency,
 		BlockDeletionMarksMigrationEnabled: c.compactorCfg.BlockDeletionMarksMigrationEnabled,
 		TenantCleanupDelay:                 c.compactorCfg.TenantCleanupDelay,
+		DeleteBlocksConcurrency:            c.compactorCfg.BlocksDeleteConcurrency,
 	}, c.bucketClient, c.usersScanner, c.cfgProvider, c.parentLogger, c.registerer)
 
 	// Initialize the compactors ring if sharding is enabled.