Compactor: Download and Upload block files in parallel

CHANGELOG.md

@@ -1,7 +1,7 @@
 # Changelog
 
 ## master / unreleased
-
+* [FEATURE] Compactor: Added `-compactor.block-files-concurrency` allowing to configure number of go routines for download/upload block files during compaction. #4784
 
 ## 1.13.0 in progress
 * [CHANGE] Changed default for `-ingester.min-ready-duration` from 1 minute to 15 seconds. #4539

docs/blocks-storage/compactor.md

@@ -152,6 +152,11 @@ compactor:
   # CLI flag: -compactor.skip-blocks-with-out-of-order-chunks-enabled
   [skip_blocks_with_out_of_order_chunks_enabled: <boolean> | default = false]
 
+  # Number of goroutines to use when fetching/uploading block files from object
+  # storage.
+  # CLI flag: -compactor.block-files-concurrency
+  [block_files_concurrency: <int> | default = 10]
+
   # 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

@@ -5317,6 +5317,11 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.skip-blocks-with-out-of-order-chunks-enabled
 [skip_blocks_with_out_of_order_chunks_enabled: <boolean> | default = false]
 
+# Number of goroutines to use when fetching/uploading block files from object
+# storage.
+# CLI flag: -compactor.block-files-concurrency
+[block_files_concurrency: <int> | default = 10]
+
 # 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

go.mod

@@ -56,7 +56,7 @@ require (
 	github.com/sony/gobreaker v0.4.1
 	github.com/spf13/afero v1.6.0
 	github.com/stretchr/testify v1.7.2
-	github.com/thanos-io/thanos v0.27.0-rc.0.0.20220707121322-5f31fc5ad310
+	github.com/thanos-io/thanos v0.27.0-rc.0.0.20220712060227-ca4fe82d74f6
 	github.com/uber/jaeger-client-go v2.30.0+incompatible
 	github.com/weaveworks/common v0.0.0-20210913144402-035033b78a78
 	go.etcd.io/bbolt v1.3.6

go.sum

@@ -2040,8 +2040,8 @@ github.com/thanos-io/thanos v0.13.1-0.20210224074000-659446cab117/go.mod h1:kdqF
 github.com/thanos-io/thanos v0.13.1-0.20210226164558-03dace0a1aa1/go.mod h1:gMCy4oCteKTT7VuXVvXLTPGzzjovX1VPE5p+HgL1hyU=
 github.com/thanos-io/thanos v0.13.1-0.20210401085038-d7dff0c84d17/go.mod h1:zU8KqE+6A+HksK4wiep8e/3UvCZLm+Wrw9AqZGaAm9k=
 github.com/thanos-io/thanos v0.22.0/go.mod h1:SZDWz3phcUcBr4MYFoPFRvl+Z9Nbi45HlwQlwSZSt+Q=
-github.com/thanos-io/thanos v0.27.0-rc.0.0.20220707121322-5f31fc5ad310 h1:mhZ/O25pK/mQ/v+5OIn2wGcvEq5XW5SnNPKmSGHyikA=
-github.com/thanos-io/thanos v0.27.0-rc.0.0.20220707121322-5f31fc5ad310/go.mod h1:v47D+NAz9g9AcUsp96zA0Cj/rzYFwu6i2Vcbz8c0NdY=
+github.com/thanos-io/thanos v0.27.0-rc.0.0.20220712060227-ca4fe82d74f6 h1:6jxorvQx1sk912ykRj29pizu8U480zKmjZb7HnQe+hM=
+github.com/thanos-io/thanos v0.27.0-rc.0.0.20220712060227-ca4fe82d74f6/go.mod h1:sklyj/ttQrL8iY3g/pQEAdIhayKW4HvOpbA7TEYK0Xs=
 github.com/themihai/gomemcache v0.0.0-20180902122335-24332e2d58ab h1:7ZR3hmisBWw77ZpO1/o86g+JV3VKlk3d48jopJxzTjU=
 github.com/themihai/gomemcache v0.0.0-20180902122335-24332e2d58ab/go.mod h1:eheTFp954zcWZXCU8d0AT76ftsQOTo4DTqkN/h3k1MY=
 github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=

pkg/compactor/compactor.go

@@ -63,7 +63,8 @@ var (
 			blocksMarkedForDeletion,
 			garbageCollectedBlocks,
 			blocksMarkedForNoCompaction,
-			metadata.NoneFunc)
+			metadata.NoneFunc,
+			cfg.BlockFilesConcurrency)
 	}
 
 	ShuffleShardingGrouperFactory = func(ctx context.Context, cfg Config, bkt objstore.Bucket, logger log.Logger, reg prometheus.Registerer, blocksMarkedForDeletion, blocksMarkedForNoCompaction, garbageCollectedBlocks prometheus.Counter, remainingPlannedCompactions prometheus.Gauge, ring *ring.Ring, ringLifecycle *ring.Lifecycler, limits Limits, userID string) compact.Grouper {
@@ -82,7 +83,8 @@ var (
 			ring,
 			ringLifecycle.Addr,
 			limits,
-			userID)
+			userID,
+			cfg.BlockFilesConcurrency)
 	}
 
 	DefaultBlocksCompactorFactory = func(ctx context.Context, cfg Config, logger log.Logger, reg prometheus.Registerer) (compact.Compactor, PlannerFactory, error) {
@@ -163,6 +165,7 @@ type Config struct {
 	DeletionDelay                         time.Duration            `yaml:"deletion_delay"`
 	TenantCleanupDelay                    time.Duration            `yaml:"tenant_cleanup_delay"`
 	SkipBlocksWithOutOfOrderChunksEnabled bool                     `yaml:"skip_blocks_with_out_of_order_chunks_enabled"`
+	BlockFilesConcurrency                 int                      `yaml:"block_files_concurrency"`
 
 	// Whether the migration of block deletion marks to the global markers location is enabled.
 	BlockDeletionMarksMigrationEnabled bool `yaml:"block_deletion_marks_migration_enabled"`
@@ -212,6 +215,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.TenantCleanupDelay, "compactor.tenant-cleanup-delay", 6*time.Hour, "For tenants marked for deletion, this is time between deleting of last block, and doing final cleanup (marker files, debug files) of the tenant.")
 	f.BoolVar(&cfg.BlockDeletionMarksMigrationEnabled, "compactor.block-deletion-marks-migration-enabled", false, "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 compactor has successfully run at least once.")
 	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.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.")

pkg/compactor/shuffle_sharding_grouper.go

@@ -40,6 +40,7 @@ type ShuffleShardingGrouper struct {
 	compactorCfg                Config
 	limits                      Limits
 	userID                      string
+	blockFilesConcurrency       int
 
 	ring               ring.ReadRing
 	ringLifecyclerAddr string
@@ -61,6 +62,7 @@ func NewShuffleShardingGrouper(
 	ringLifecyclerAddr string,
 	limits Limits,
 	userID string,
+	blockFilesConcurrency int,
 ) *ShuffleShardingGrouper {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -98,11 +100,12 @@ func NewShuffleShardingGrouper(
 			Name: "thanos_compact_group_vertical_compactions_total",
 			Help: "Total number of group compaction attempts that resulted in a new block based on overlapping blocks.",
 		}, []string{"group"}),
-		compactorCfg:       compactorCfg,
-		ring:               ring,
-		ringLifecyclerAddr: ringLifecyclerAddr,
-		limits:             limits,
-		userID:             userID,
+		compactorCfg:          compactorCfg,
+		ring:                  ring,
+		ringLifecyclerAddr:    ringLifecyclerAddr,
+		limits:                limits,
+		userID:                userID,
+		blockFilesConcurrency: blockFilesConcurrency,
 	}
 }
 
@@ -180,6 +183,7 @@ func (g *ShuffleShardingGrouper) Groups(blocks map[ulid.ULID]*metadata.Meta) (re
 				g.blocksMarkedForDeletion,
 				g.blocksMarkedForNoCompact,
 				g.hashFunc,
+				g.blockFilesConcurrency,
 			)
 			if err != nil {
 				return nil, errors.Wrap(err, "create compaction group")

pkg/compactor/shuffle_sharding_grouper_test.go

@@ -240,7 +240,8 @@ func TestShuffleShardingGrouper_Groups(t *testing.T) {
 				ring,
 				"test-addr",
 				overrides,
-				"")
+				"",
+				10)
 			actual, err := g.Groups(testData.blocks)
 			require.NoError(t, err)
 			require.Len(t, actual, len(testData.expected))