Configurable block listing strategy (#5828)

* make block listing strategy configurable. Change default strategy from Recursive to Concurrent

Signed-off-by: Ben Ye <[email protected]>

* update docs

Signed-off-by: Ben Ye <[email protected]>

---------

Signed-off-by: Ben Ye <[email protected]>

Author: yeya24

CHANGELOG.md

@@ -9,7 +9,7 @@
 * [CHANGE] Ruler: Remove `cortex_ruler_write_requests_total`, `cortex_ruler_write_requests_failed_total`, `cortex_ruler_queries_total`, `cortex_ruler_queries_failed_total`, and `cortex_ruler_query_seconds_total` metrics for the tenant when the ruler deletes the manager for the tenant. #5772
 * [CHANGE] Main: Mark `mem-ballast-size-bytes` flag as deprecated. #5816
 * [CHANGE] Querier: Mark `-querier.ingester-streaming` flag as deprecated. Now query ingester streaming is always enabled. #5817
-* [CHANGE] AlertManager API: Removal of all api/v1/ endpoints following [2970](https://github.com/prometheus/alertmanager/pull/2970). [5841]
+* [CHANGE] Compactor/Bucket Store: Added `-blocks-storage.bucket-store.block-discovery-strategy` to configure different block listing strategy. Reverted the current recursive block listing mechanism and use the strategy `Concurrent` as in 1.15. #5828
 * [FEATURE] Ingester: Add per-tenant new metric `cortex_ingester_tsdb_data_replay_duration_seconds`. #5477
 * [FEATURE] Query Frontend/Scheduler: Add query priority support. #5605
 * [FEATURE] Tracing: Add `kuberesolver` to resolve endpoints address with `kubernetes://` prefix as Kubernetes service. #5731

docs/blocks-storage/querier.md

@@ -1314,6 +1314,17 @@ blocks_storage:
       # CLI flag: -blocks-storage.bucket-store.bucket-index.max-stale-period
       [max_stale_period: <duration> | default = 1h]
 
+    # One of concurrent, recursive, bucket_index. When set to concurrent, stores
+    # will concurrently issue one call per directory to discover active blocks
+    # in the bucket. The recursive strategy iterates through all objects in the
+    # bucket, recursively traversing into each directory. This avoids N+1 calls
+    # at the expense of having slower bucket iterations. bucket_index strategy
+    # can be used in Compactor only and utilizes the existing bucket index to
+    # fetch block IDs to sync. This avoids iterating the bucket but can be
+    # impacted by delays of cleaner creating bucket index.
+    # CLI flag: -blocks-storage.bucket-store.block-discovery-strategy
+    [block_discovery_strategy: <string> | default = "concurrent"]
+
     # Max size - in bytes - of a chunks pool, used to reduce memory allocations.
     # The pool is shared across all tenants. 0 to disable the limit.
     # CLI flag: -blocks-storage.bucket-store.max-chunk-pool-bytes

docs/blocks-storage/store-gateway.md

@@ -1423,6 +1423,17 @@ blocks_storage:
       # CLI flag: -blocks-storage.bucket-store.bucket-index.max-stale-period
       [max_stale_period: <duration> | default = 1h]
 
+    # One of concurrent, recursive, bucket_index. When set to concurrent, stores
+    # will concurrently issue one call per directory to discover active blocks
+    # in the bucket. The recursive strategy iterates through all objects in the
+    # bucket, recursively traversing into each directory. This avoids N+1 calls
+    # at the expense of having slower bucket iterations. bucket_index strategy
+    # can be used in Compactor only and utilizes the existing bucket index to
+    # fetch block IDs to sync. This avoids iterating the bucket but can be
+    # impacted by delays of cleaner creating bucket index.
+    # CLI flag: -blocks-storage.bucket-store.block-discovery-strategy
+    [block_discovery_strategy: <string> | default = "concurrent"]
+
     # Max size - in bytes - of a chunks pool, used to reduce memory allocations.
     # The pool is shared across all tenants. 0 to disable the limit.
     # CLI flag: -blocks-storage.bucket-store.max-chunk-pool-bytes

docs/configuration/config-file-reference.md

@@ -1850,6 +1850,17 @@ bucket_store:
     # CLI flag: -blocks-storage.bucket-store.bucket-index.max-stale-period
     [max_stale_period: <duration> | default = 1h]
 
+  # One of concurrent, recursive, bucket_index. When set to concurrent, stores
+  # will concurrently issue one call per directory to discover active blocks in
+  # the bucket. The recursive strategy iterates through all objects in the
+  # bucket, recursively traversing into each directory. This avoids N+1 calls at
+  # the expense of having slower bucket iterations. bucket_index strategy can be
+  # used in Compactor only and utilizes the existing bucket index to fetch block
+  # IDs to sync. This avoids iterating the bucket but can be impacted by delays
+  # of cleaner creating bucket index.
+  # CLI flag: -blocks-storage.bucket-store.block-discovery-strategy
+  [block_discovery_strategy: <string> | default = "concurrent"]
+
   # Max size - in bytes - of a chunks pool, used to reduce memory allocations.
   # The pool is shared across all tenants. 0 to disable the limit.
   # CLI flag: -blocks-storage.bucket-store.max-chunk-pool-bytes

pkg/compactor/compactor.go

@@ -827,22 +827,26 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	// out of order chunks or index file too big.
 	noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(ulogger, bucket, c.compactorCfg.MetaSyncConcurrency)
 
-	var blockIDsFetcher block.Lister
-	var fetcherULogger log.Logger
-	if c.storageCfg.BucketStore.BucketIndex.Enabled {
-		fetcherULogger = log.With(ulogger, "blockIdsFetcher", "BucketIndexBlockIDsFetcher")
-		blockIDsFetcher = bucketindex.NewBlockIDsFetcher(fetcherULogger, c.bucketClient, userID, c.limits)
-
-	} else {
-		fetcherULogger = log.With(ulogger, "blockIdsFetcher", "BaseBlockIDsFetcher")
-		blockIDsFetcher = block.NewRecursiveLister(fetcherULogger, bucket)
+	var blockLister block.Lister
+	switch cortex_tsdb.BlockDiscoveryStrategy(c.storageCfg.BucketStore.BlockDiscoveryStrategy) {
+	case cortex_tsdb.ConcurrentDiscovery:
+		blockLister = block.NewConcurrentLister(ulogger, bucket)
+	case cortex_tsdb.RecursiveDiscovery:
+		blockLister = block.NewRecursiveLister(ulogger, bucket)
+	case cortex_tsdb.BucketIndexDiscovery:
+		if !c.storageCfg.BucketStore.BucketIndex.Enabled {
+			return cortex_tsdb.ErrInvalidBucketIndexBlockDiscoveryStrategy
+		}
+		blockLister = bucketindex.NewBlockLister(ulogger, c.bucketClient, userID, c.limits)
+	default:
+		return cortex_tsdb.ErrBlockDiscoveryStrategy
 	}
 
 	fetcher, err := block.NewMetaFetcher(
-		fetcherULogger,
+		ulogger,
 		c.compactorCfg.MetaSyncConcurrency,
 		bucket,
-		blockIDsFetcher,
+		blockLister,
 		c.metaSyncDirForUser(userID),
 		reg,
 		// List of filters to apply (order matters).

pkg/compactor/compactor_test.go

@@ -1673,6 +1673,7 @@ func prepareConfig() Config {
 func prepare(t *testing.T, compactorCfg Config, bucketClient objstore.InstrumentedBucket, limits *validation.Limits) (*Compactor, *tsdbCompactorMock, *tsdbPlannerMock, *concurrency.SyncBuffer, prometheus.Gatherer) {
 	storageCfg := cortex_tsdb.BlocksStorageConfig{}
 	flagext.DefaultValues(&storageCfg)
+	storageCfg.BucketStore.BlockDiscoveryStrategy = string(cortex_tsdb.RecursiveDiscovery)
 
 	// Create a temporary directory for compactor data.
 	compactorCfg.DataDir = t.TempDir()

pkg/querier/blocks_finder_bucket_scan.go

@@ -43,6 +43,8 @@ type BucketScanBlocksFinderConfig struct {
 	ConsistencyDelay         time.Duration
 	IgnoreDeletionMarksDelay time.Duration
 	IgnoreBlocksWithin       time.Duration
+
+	BlockDiscoveryStrategy string
 }
 
 // BucketScanBlocksFinder is a BlocksFinder implementation periodically scanning the bucket to discover blocks.
@@ -384,12 +386,25 @@ func (d *BucketScanBlocksFinder) createMetaFetcher(userID string) (block.Metadat
 		filters = append(filters, storegateway.NewIgnoreNonQueryableBlocksFilter(d.logger, d.cfg.IgnoreBlocksWithin))
 	}
 
-	blockIdsFetcher := block.NewRecursiveLister(userLogger, userBucket)
+	var (
+		err         error
+		blockLister block.Lister
+	)
+	switch cortex_tsdb.BlockDiscoveryStrategy(d.cfg.BlockDiscoveryStrategy) {
+	case cortex_tsdb.ConcurrentDiscovery:
+		blockLister = block.NewConcurrentLister(userLogger, userBucket)
+	case cortex_tsdb.RecursiveDiscovery:
+		blockLister = block.NewRecursiveLister(userLogger, userBucket)
+	case cortex_tsdb.BucketIndexDiscovery:
+		return nil, nil, nil, cortex_tsdb.ErrInvalidBucketIndexBlockDiscoveryStrategy
+	default:
+		return nil, nil, nil, cortex_tsdb.ErrBlockDiscoveryStrategy
+	}
 	f, err := block.NewMetaFetcher(
 		userLogger,
 		d.cfg.MetasConcurrency,
 		userBucket,
-		blockIdsFetcher,
+		blockLister,
 		// The fetcher stores cached metas in the "meta-syncer/" sub directory.
 		filepath.Join(d.cfg.CacheDir, userID),
 		userReg,

pkg/querier/blocks_finder_bucket_scan_test.go

@@ -521,5 +521,6 @@ func prepareBucketScanBlocksFinderConfig() BucketScanBlocksFinderConfig {
 		MetasConcurrency:         10,
 		IgnoreDeletionMarksDelay: time.Hour,
 		IgnoreBlocksWithin:       10 * time.Hour, // All blocks created in the last 10 hour shouldn't be scanned.
+		BlockDiscoveryStrategy:   string(cortex_tsdb.RecursiveDiscovery),
 	}
 }

pkg/querier/blocks_store_queryable.go

@@ -219,6 +219,7 @@ func NewBlocksStoreQueryableFromConfig(querierCfg Config, gatewayCfg storegatewa
 			CacheDir:                 storageCfg.BucketStore.SyncDir,
 			IgnoreDeletionMarksDelay: storageCfg.BucketStore.IgnoreDeletionMarksDelay,
 			IgnoreBlocksWithin:       storageCfg.BucketStore.IgnoreBlocksWithin,
+			BlockDiscoveryStrategy:   storageCfg.BucketStore.BlockDiscoveryStrategy,
 		}, bucketClient, limits, logger, reg)
 	}
 

pkg/storage/tsdb/bucketindex/block_ids_fetcher.go

@@ -13,40 +13,40 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 )
 
-type BlockIDsFetcher struct {
-	logger              log.Logger
-	bkt                 objstore.Bucket
-	userID              string
-	cfgProvider         bucket.TenantConfigProvider
-	baseBlockIDsFetcher block.Lister
+type BlockLister struct {
+	logger      log.Logger
+	bkt         objstore.Bucket
+	userID      string
+	cfgProvider bucket.TenantConfigProvider
+	baseLister  block.Lister
 }
 
-func NewBlockIDsFetcher(logger log.Logger, bkt objstore.Bucket, userID string, cfgProvider bucket.TenantConfigProvider) *BlockIDsFetcher {
+func NewBlockLister(logger log.Logger, bkt objstore.Bucket, userID string, cfgProvider bucket.TenantConfigProvider) *BlockLister {
 	userBkt := bucket.NewUserBucketClient(userID, bkt, cfgProvider)
-	baseBlockIDsFetcher := block.NewRecursiveLister(logger, userBkt)
-	return &BlockIDsFetcher{
-		logger:              logger,
-		bkt:                 bkt,
-		userID:              userID,
-		cfgProvider:         cfgProvider,
-		baseBlockIDsFetcher: baseBlockIDsFetcher,
+	baseLister := block.NewConcurrentLister(logger, userBkt)
+	return &BlockLister{
+		logger:      logger,
+		bkt:         bkt,
+		userID:      userID,
+		cfgProvider: cfgProvider,
+		baseLister:  baseLister,
 	}
 }
 
-func (f *BlockIDsFetcher) GetActiveAndPartialBlockIDs(ctx context.Context, ch chan<- ulid.ULID) (partialBlocks map[ulid.ULID]bool, err error) {
+func (f *BlockLister) GetActiveAndPartialBlockIDs(ctx context.Context, ch chan<- ulid.ULID) (partialBlocks map[ulid.ULID]bool, err error) {
 	// Fetch the bucket index.
 	idx, err := ReadIndex(ctx, f.bkt, f.userID, f.cfgProvider, f.logger)
 	if errors.Is(err, ErrIndexNotFound) {
 		// This is a legit case happening when the first blocks of a tenant have recently been uploaded by ingesters
 		// and their bucket index has not been created yet.
 		// Fallback to BaseBlockIDsFetcher.
-		return f.baseBlockIDsFetcher.GetActiveAndPartialBlockIDs(ctx, ch)
+		return f.baseLister.GetActiveAndPartialBlockIDs(ctx, ch)
 	}
 	if errors.Is(err, ErrIndexCorrupted) {
 		// In case a single tenant bucket index is corrupted, we want to return empty active blocks and parital blocks, so skipping this compaction cycle
 		level.Error(f.logger).Log("msg", "corrupted bucket index found", "user", f.userID, "err", err)
 		// Fallback to BaseBlockIDsFetcher.
-		return f.baseBlockIDsFetcher.GetActiveAndPartialBlockIDs(ctx, ch)
+		return f.baseLister.GetActiveAndPartialBlockIDs(ctx, ch)
 	}
 
 	if errors.Is(err, bucket.ErrCustomerManagedKeyAccessDenied) {