Update/thanos blocks (#4787)

* go mod Signed-off-by: alanprot <[email protected]> * Adding compactor.blocks-fetch-concurrency Signed-off-by: alanprot <[email protected]> * Docs Signed-off-by: alanprot <[email protected]> * Changelog Signed-off-by: alanprot <[email protected]> * Fix test Signed-off-by: Alan Protasio <[email protected]> * Update thanos to fix the race condition Signed-off-by: alanprot <[email protected]>
cortexproject · Jul 18, 2022 · 2575a0a · 2575a0a
1 parent e8e5432
commit 2575a0a
Show file tree

Hide file tree

Showing 15 changed files with 218 additions and 175 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## master / unreleased
 * [FEATURE] Compactor: Added `-compactor.block-files-concurrency` allowing to configure number of go routines for download/upload block files during compaction. #4784
 * [ENHANCEMENT] Querier/Ruler: Retry store-gateway in case of unexpected failure, instead of failing the query. #4532
+* [FEATURE] Compactor: Added -compactor.blocks-fetch-concurrency` allowing to configure number of go routines for blocks during compaction. #4787
 
 ## 1.13.0 2022-07-14
 * [CHANGE] Changed default for `-ingester.min-ready-duration` from 1 minute to 15 seconds. #4539

diff --git a/docs/blocks-storage/compactor.md b/docs/blocks-storage/compactor.md
@@ -157,6 +157,11 @@ compactor:
   # CLI flag: -compactor.block-files-concurrency
   [block_files_concurrency: <int> | default = 10]
 
+  # Number of goroutines to use when fetching blocks from object storage when
+  # compacting.
+  # CLI flag: -compactor.blocks-fetch-concurrency
+  [blocks_fetch_concurrency: <int> | default = 3]
+
   # 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

diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md
@@ -5322,6 +5322,11 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.block-files-concurrency
 [block_files_concurrency: <int> | default = 10]
 
+# Number of goroutines to use when fetching blocks from object storage when
+# compacting.
+# CLI flag: -compactor.blocks-fetch-concurrency
+[blocks_fetch_concurrency: <int> | default = 3]
+
 # 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

diff --git a/go.mod b/go.mod
@@ -39,7 +39,7 @@ require (
 	github.com/hashicorp/memberlist v0.3.1
 	github.com/json-iterator/go v1.1.12
 	github.com/lib/pq v1.3.0
-	github.com/minio/minio-go/v7 v7.0.30
+	github.com/minio/minio-go/v7 v7.0.32-0.20220706200439-ef3e45ed9cdb
 	github.com/mitchellh/go-wordwrap v1.0.0
 	github.com/ncw/swift v1.0.52
 	github.com/oklog/ulid v1.3.1
@@ -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.20220712060227-ca4fe82d74f6
+	github.com/thanos-io/thanos v0.27.0-rc.0.0.20220715025542-84880ea6d7f8
 	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

diff --git a/go.sum b/go.sum
@@ -1551,8 +1551,8 @@ github.com/minio/minio-go/v6 v6.0.44/go.mod h1:qD0lajrGW49lKZLtXKtCB4X/qkMf0a5tB
 github.com/minio/minio-go/v6 v6.0.56/go.mod h1:KQMM+/44DSlSGSQWSfRrAZ12FVMmpWNuX37i2AX0jfI=
 github.com/minio/minio-go/v7 v7.0.2/go.mod h1:dJ80Mv2HeGkYLH1sqS/ksz07ON6csH3S6JUMSQ2zAns=
 github.com/minio/minio-go/v7 v7.0.10/go.mod h1:td4gW1ldOsj1PbSNS+WYK43j+P1XVhX/8W8awaYlBFo=
-github.com/minio/minio-go/v7 v7.0.30 h1:Re+qlwA+LB3mgFGYbztVPzlEjKtGzRVV5Sk38np858k=
-github.com/minio/minio-go/v7 v7.0.30/go.mod h1:/sjRKkKIA75CKh1iu8E3qBy7ktBmCCDGII0zbXGwbUk=
+github.com/minio/minio-go/v7 v7.0.32-0.20220706200439-ef3e45ed9cdb h1:J7jRWqlD+K3Tp4YbLWcyBKiHoNRy49JR5HA4RetFrAY=
+github.com/minio/minio-go/v7 v7.0.32-0.20220706200439-ef3e45ed9cdb/go.mod h1:/sjRKkKIA75CKh1iu8E3qBy7ktBmCCDGII0zbXGwbUk=
 github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
 github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
 github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
@@ -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.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/thanos-io/thanos v0.27.0-rc.0.0.20220715025542-84880ea6d7f8 h1:kdpa2sXUyxQy3KAk00T+4R9rZFCX5vfdwFmf+TWk5BE=
+github.com/thanos-io/thanos v0.27.0-rc.0.0.20220715025542-84880ea6d7f8/go.mod h1:Ga4NVtqoPVGjZmKiu5Wz2x6V9o4ohdeV9ZAcztvH13M=
 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=

diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
@@ -64,7 +64,8 @@ var (
 			garbageCollectedBlocks,
 			blocksMarkedForNoCompaction,
 			metadata.NoneFunc,
-			cfg.BlockFilesConcurrency)
+			cfg.BlockFilesConcurrency,
+			cfg.BlocksFetchConcurrency)
 	}
 
 	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 {
@@ -84,7 +85,8 @@ var (
 			ringLifecycle.Addr,
 			limits,
 			userID,
-			cfg.BlockFilesConcurrency)
+			cfg.BlockFilesConcurrency,
+			cfg.BlocksFetchConcurrency)
 	}
 
 	DefaultBlocksCompactorFactory = func(ctx context.Context, cfg Config, logger log.Logger, reg prometheus.Registerer) (compact.Compactor, PlannerFactory, error) {
@@ -166,6 +168,7 @@ type Config struct {
 	TenantCleanupDelay                    time.Duration            `yaml:"tenant_cleanup_delay"`
 	SkipBlocksWithOutOfOrderChunksEnabled bool                     `yaml:"skip_blocks_with_out_of_order_chunks_enabled"`
 	BlockFilesConcurrency                 int                      `yaml:"block_files_concurrency"`
+	BlocksFetchConcurrency                int                      `yaml:"blocks_fetch_concurrency"`
 
 	// Whether the migration of block deletion marks to the global markers location is enabled.
 	BlockDeletionMarksMigrationEnabled bool `yaml:"block_deletion_marks_migration_enabled"`
@@ -216,6 +219,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	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.IntVar(&cfg.BlocksFetchConcurrency, "compactor.blocks-fetch-concurrency", 3, "Number of goroutines to use when fetching 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.")

diff --git a/pkg/compactor/shuffle_sharding_grouper.go b/pkg/compactor/shuffle_sharding_grouper.go
@@ -41,6 +41,7 @@ type ShuffleShardingGrouper struct {
 	limits                      Limits
 	userID                      string
 	blockFilesConcurrency       int
+	blocksFetchConcurrency      int
 
 	ring               ring.ReadRing
 	ringLifecyclerAddr string
@@ -63,6 +64,7 @@ func NewShuffleShardingGrouper(
 	limits Limits,
 	userID string,
 	blockFilesConcurrency int,
+	blocksFetchConcurrency int,
 ) *ShuffleShardingGrouper {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -100,12 +102,13 @@ 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,
-		blockFilesConcurrency: blockFilesConcurrency,
+		compactorCfg:           compactorCfg,
+		ring:                   ring,
+		ringLifecyclerAddr:     ringLifecyclerAddr,
+		limits:                 limits,
+		userID:                 userID,
+		blockFilesConcurrency:  blockFilesConcurrency,
+		blocksFetchConcurrency: blocksFetchConcurrency,
 	}
 }
 
@@ -184,6 +187,7 @@ func (g *ShuffleShardingGrouper) Groups(blocks map[ulid.ULID]*metadata.Meta) (re
 				g.blocksMarkedForNoCompact,
 				g.hashFunc,
 				g.blockFilesConcurrency,
+				g.blocksFetchConcurrency,
 			)
 			if err != nil {
 				return nil, errors.Wrap(err, "create compaction group")

diff --git a/pkg/compactor/shuffle_sharding_grouper_test.go b/pkg/compactor/shuffle_sharding_grouper_test.go
@@ -241,7 +241,8 @@ func TestShuffleShardingGrouper_Groups(t *testing.T) {
 				"test-addr",
 				overrides,
 				"",
-				10)
+				10,
+				3)
 			actual, err := g.Groups(testData.blocks)
 			require.NoError(t, err)
 			require.Len(t, actual, len(testData.expected))

diff --git a/vendor/github.com/minio/minio-go/v7/api-put-object-streaming.go b/vendor/github.com/minio/minio-go/v7/api-put-object-streaming.go
diff --git a/vendor/github.com/minio/minio-go/v7/api.go b/vendor/github.com/minio/minio-go/v7/api.go
diff --git a/vendor/github.com/minio/minio-go/v7/hook-reader.go b/vendor/github.com/minio/minio-go/v7/hook-reader.go