Exposed --shipper.upload-compacted flag in components that use shipper

rdowgird-rtbh · rdowgird-rtbh · commit 83499023c13b · 2025-12-22T15:15:18.000+01:00
Signed-off-by: Rafał Dowgird &lt;rafal.dowgird@rtbhouse.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -19,6 +19,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ### Added
 
 - [#](https://github.com/thanos-io/thanos/pull/8582): Sidecar: support --storage.tsdb.delay-compact-file.path Prometheus flag.
+- [#](.): Tools: add --shipper.upload-compacted flag for controlling upload concurrency in bucket upload-blocks
 
 ### Changed
 
diff --git a/cmd/thanos/config.go b/cmd/thanos/config.go
@@ -206,6 +206,7 @@ type shipperConfig struct {
 	skipCorruptedBlocks   bool
 	hashFunc              string
 	metaFileName          string
+	uploadConcurrency     int
 }
 
 func (sc *shipperConfig) registerFlag(cmd extkingpin.FlagClause) *shipperConfig {
@@ -228,6 +229,7 @@ func (sc *shipperConfig) registerFlag(cmd extkingpin.FlagClause) *shipperConfig
 	cmd.Flag("hash-func", "Specify which hash function to use when calculating the hashes of produced files. If no function has been specified, it does not happen. This permits avoiding downloading some files twice albeit at some performance cost. Possible values are: \"\", \"SHA256\".").
 		Default("").EnumVar(&sc.hashFunc, "SHA256", "")
 	cmd.Flag("shipper.meta-file-name", "the file to store shipper metadata in").Default(shipper.DefaultMetaFilename).StringVar(&sc.metaFileName)
+	cmd.Flag("shipper.upload-concurrency", "Number of goroutines to use when uploading block files to object storage.").Default("0").IntVar(&sc.uploadConcurrency)
 	return sc
 }
 
diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go
@@ -235,6 +235,8 @@ func runReceive(
 		multiTSDBOptions = append(multiTSDBOptions, receive.WithMatchersCache(cache))
 	}
 
+	multiTSDBOptions = append(multiTSDBOptions, receive.WithUploadConcurrency(conf.uploadConcurrency))
+
 	dbs := receive.NewMultiTSDB(
 		conf.dataDir,
 		logger,
@@ -916,6 +918,7 @@ type receiveConfig struct {
 	ignoreBlockSize       bool
 	allowOutOfOrderUpload bool
 	skipCorruptedBlocks   bool
+	uploadConcurrency     int
 
 	reqLogConfig      *extflag.PathOrContent
 	relabelConfigPath *extflag.PathOrContent
@@ -1093,6 +1096,8 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 			"about order.").
 		Default("false").Hidden().BoolVar(&rc.skipCorruptedBlocks)
 
+	cmd.Flag("shipper.upload-concurrency", "Number of goroutines to use when uploading block files to object storage.").Default("0").IntVar(&rc.uploadConcurrency)
+
 	cmd.Flag("matcher-cache-size", "Max number of cached matchers items. Using 0 disables caching.").Default("0").IntVar(&rc.matcherCacheSize)
 
 	rc.reqLogConfig = extkingpin.RegisterRequestLoggingFlags(cmd)
diff --git a/cmd/thanos/rule.go b/cmd/thanos/rule.go
@@ -871,6 +871,7 @@ func runRule(
 			shipper.WithLabels(func() labels.Labels { return conf.lset }),
 			shipper.WithAllowOutOfOrderUploads(conf.shipper.allowOutOfOrderUpload),
 			shipper.WithSkipCorruptedBlocks(conf.shipper.skipCorruptedBlocks),
+			shipper.WithUploadConcurrency(conf.shipper.uploadConcurrency),
 		)
 
 		ctx, cancel := context.WithCancel(context.Background())
diff --git a/cmd/thanos/sidecar.go b/cmd/thanos/sidecar.go
@@ -426,6 +426,7 @@ func runSidecar(
 				shipper.WithUploadCompacted(conf.shipper.uploadCompacted),
 				shipper.WithAllowOutOfOrderUploads(conf.shipper.allowOutOfOrderUpload),
 				shipper.WithSkipCorruptedBlocks(conf.shipper.skipCorruptedBlocks),
+				shipper.WithUploadConcurrency(conf.shipper.uploadConcurrency),
 			)
 
 			return runutil.Repeat(30*time.Second, ctx.Done(), func() error {
diff --git a/cmd/thanos/tools_bucket.go b/cmd/thanos/tools_bucket.go
@@ -166,9 +166,10 @@ type bucketMarkBlockConfig struct {
 }
 
 type bucketUploadBlocksConfig struct {
-	path            string
-	labels          []string
-	uploadCompacted bool
+	path              string
+	labels            []string
+	uploadCompacted   bool
+	uploadConcurrency int
 }
 
 func (tbc *bucketVerifyConfig) registerBucketVerifyFlag(cmd extkingpin.FlagClause) *bucketVerifyConfig {
@@ -302,6 +303,7 @@ func (tbc *bucketUploadBlocksConfig) registerBucketUploadBlocksFlag(cmd extkingp
 	cmd.Flag("path", "Path to the directory containing blocks to upload.").Default("./data").StringVar(&tbc.path)
 	cmd.Flag("label", "External labels to add to the uploaded blocks (repeated).").PlaceHolder("key=\"value\"").StringsVar(&tbc.labels)
 	cmd.Flag("shipper.upload-compacted", "If true shipper will try to upload compacted blocks as well.").Default("false").BoolVar(&tbc.uploadCompacted)
+	cmd.Flag("shipper.upload-concurrency", "Number of goroutines to use when uploading block files to object storage.").Default("5").IntVar(&tbc.uploadConcurrency)
 
 	return tbc
 }
@@ -1512,6 +1514,7 @@ func registerBucketUploadBlocks(app extkingpin.AppClause, objStoreConfig *extfla
 			shipper.WithMetaFileName(shipper.DefaultMetaFilename),
 			shipper.WithLabels(func() labels.Labels { return lset }),
 			shipper.WithUploadCompacted(tbc.uploadCompacted),
+			shipper.WithUploadConcurrency(tbc.uploadConcurrency),
 		)
 
 		ctx, cancel := context.WithCancel(context.Background())
diff --git a/docs/components/receive.md b/docs/components/receive.md
@@ -643,6 +643,9 @@ Flags:
                                  happen. This permits avoiding downloading some
                                  files twice albeit at some performance cost.
                                  Possible values are: "", "SHA256".
+      --shipper.upload-concurrency=0
+                                 Number of goroutines to use when uploading
+                                 block files to object storage.
       --matcher-cache-size=0     Max number of cached matchers items. Using 0
                                  disables caching.
       --request.logging-config-file=<file-path>
diff --git a/docs/components/rule.md b/docs/components/rule.md
@@ -362,6 +362,9 @@ Flags:
                                  Possible values are: "", "SHA256".
       --shipper.meta-file-name="thanos.shipper.json"
                                  the file to store shipper metadata in
+      --shipper.upload-concurrency=0
+                                 Number of goroutines to use when uploading
+                                 block files to object storage.
       --query=<query> ...        Addresses of statically configured query
                                  API servers (repeatable). The scheme may be
                                  prefixed with 'dns+' or 'dnssrv+' to detect
diff --git a/docs/components/sidecar.md b/docs/components/sidecar.md
@@ -220,6 +220,9 @@ Flags:
                                  Possible values are: "", "SHA256".
       --shipper.meta-file-name="thanos.shipper.json"
                                  the file to store shipper metadata in
+      --shipper.upload-concurrency=0
+                                 Number of goroutines to use when uploading
+                                 block files to object storage.
       --store.limits.request-series=0
                                  The maximum series allowed for a single Series
                                  request. The Series call fails if this limit is
diff --git a/docs/components/tools.md b/docs/components/tools.md
@@ -1054,6 +1054,9 @@ Flags:
       --[no-]shipper.upload-compacted
                                If true shipper will try to upload compacted
                                blocks as well.
+      --shipper.upload-concurrency=5
+                               Number of goroutines to use when uploading block
+                               files to object storage.
 
 ```
 
diff --git a/pkg/receive/multitsdb.go b/pkg/receive/multitsdb.go
@@ -66,6 +66,7 @@ type MultiTSDB struct {
 	allowOutOfOrderUpload bool
 	skipCorruptedBlocks   bool
 	hashFunc              metadata.HashFunc
+	uploadConcurrency     int
 	hashringConfigs       []HashringConfig
 
 	matcherCache storecache.MatchersCache
@@ -109,6 +110,12 @@ func WithMatchersCache(cache storecache.MatchersCache) MultiTSDBOption {
 	}
 }
 
+func WithUploadConcurrency(concurrency int) MultiTSDBOption {
+	return func(s *MultiTSDB) {
+		s.uploadConcurrency = concurrency
+	}
+}
+
 // NewMultiTSDB creates new MultiTSDB.
 // NOTE: Passed labels must be sorted lexicographically (alphabetically).
 func NewMultiTSDB(
@@ -143,6 +150,7 @@ func NewMultiTSDB(
 		allowOutOfOrderUpload: allowOutOfOrderUpload,
 		skipCorruptedBlocks:   skipCorruptedBlocks,
 		hashFunc:              hashFunc,
+		uploadConcurrency:     0,
 		matcherCache:          storecache.NoopMatchersCache,
 	}
 
@@ -799,6 +807,7 @@ func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant
 			shipper.WithLabels(func() labels.Labels { return lset }),
 			shipper.WithAllowOutOfOrderUploads(t.allowOutOfOrderUpload),
 			shipper.WithSkipCorruptedBlocks(t.skipCorruptedBlocks),
+			shipper.WithUploadConcurrency(t.uploadConcurrency),
 		)
 	}
 	var options []store.TSDBStoreOption
diff --git a/pkg/shipper/shipper.go b/pkg/shipper/shipper.go
@@ -85,6 +85,7 @@ type Shipper struct {
 	allowOutOfOrderUploads bool
 	skipCorruptedBlocks    bool
 	hashFunc               metadata.HashFunc
+	uploadConcurrency      int
 
 	labels func() labels.Labels
 	mtx    sync.RWMutex
@@ -104,6 +105,7 @@ type shipperOptions struct {
 	uploadCompacted        bool
 	allowOutOfOrderUploads bool
 	skipCorruptedBlocks    bool
+	uploadConcurrency      int
 }
 
 type Option func(*shipperOptions)
@@ -171,6 +173,13 @@ func WithSkipCorruptedBlocks(skip bool) Option {
 	}
 }
 
+// WithUploadConcurrency sets the number of goroutines to use when uploading block files.
+func WithUploadConcurrency(concurrency int) Option {
+	return func(o *shipperOptions) {
+		o.uploadConcurrency = concurrency
+	}
+}
+
 func applyOptions(opts []Option) *shipperOptions {
 	so := new(shipperOptions)
 	for _, o := range opts {
@@ -209,6 +218,7 @@ func New(bucket objstore.Bucket, dir string, opts ...Option) *Shipper {
 		skipCorruptedBlocks:    options.skipCorruptedBlocks,
 		uploadCompacted:        options.uploadCompacted,
 		hashFunc:               options.hashFunc,
+		uploadConcurrency:      options.uploadConcurrency,
 		metadataFilePath:       filepath.Join(dir, filepath.Clean(options.metaFileName)),
 	}
 }
@@ -463,7 +473,11 @@ func (s *Shipper) upload(ctx context.Context, meta *metadata.Meta) error {
 	if err := meta.WriteToDir(s.logger, updir); err != nil {
 		return errors.Wrap(err, "write meta file")
 	}
-	return block.Upload(ctx, s.logger, s.bucket, updir, s.hashFunc)
+	var uploadOptions []objstore.UploadOption
+	if s.uploadConcurrency > 0 {
+		uploadOptions = append(uploadOptions, objstore.WithUploadConcurrency(s.uploadConcurrency))
+	}
+	return block.Upload(ctx, s.logger, s.bucket, updir, s.hashFunc, uploadOptions...)
 }
 
 // blockMetasFromOldest returns the block meta of each block found in dir

Original file line number	Diff line number	Diff line change
`@@ -206,6 +206,7 @@ type shipperConfig struct {`
`206`	`206`	`skipCorruptedBlocks bool`
`207`	`207`	`hashFunc string`
`208`	`208`	`metaFileName string`
	`209`	`+ uploadConcurrency int`
`209`	`210`	`}`
`210`	`211`
`211`	`212`	`func (sc shipperConfig) registerFlag(cmd extkingpin.FlagClause) shipperConfig {`
`@@ -228,6 +229,7 @@ func (sc shipperConfig) registerFlag(cmd extkingpin.FlagClause) shipperConfig`
`228`	`229`	`cmd.Flag("hash-func", "Specify which hash function to use when calculating the hashes of produced files. If no function has been specified, it does not happen. This permits avoiding downloading some files twice albeit at some performance cost. Possible values are: \"\", \"SHA256\".").`
`229`	`230`	`Default("").EnumVar(&sc.hashFunc, "SHA256", "")`
`230`	`231`	`cmd.Flag("shipper.meta-file-name", "the file to store shipper metadata in").Default(shipper.DefaultMetaFilename).StringVar(&sc.metaFileName)`
	`232`	`+ cmd.Flag("shipper.upload-concurrency", "Number of goroutines to use when uploading block files to object storage.").Default("0").IntVar(&sc.uploadConcurrency)`
`231`	`233`	`return sc`
`232`	`234`	`}`
`233`	`235`
Original file line number	Diff line number	Diff line change
`@@ -871,6 +871,7 @@ func runRule(`
`871`	`871`	`shipper.WithLabels(func() labels.Labels { return conf.lset }),`
`872`	`872`	`shipper.WithAllowOutOfOrderUploads(conf.shipper.allowOutOfOrderUpload),`
`873`	`873`	`shipper.WithSkipCorruptedBlocks(conf.shipper.skipCorruptedBlocks),`
	`874`	`+ shipper.WithUploadConcurrency(conf.shipper.uploadConcurrency),`
`874`	`875`	`)`
`875`	`876`
`876`	`877`	`ctx, cancel := context.WithCancel(context.Background())`
Original file line number	Diff line number	Diff line change
`@@ -426,6 +426,7 @@ func runSidecar(`
`426`	`426`	`shipper.WithUploadCompacted(conf.shipper.uploadCompacted),`
`427`	`427`	`shipper.WithAllowOutOfOrderUploads(conf.shipper.allowOutOfOrderUpload),`
`428`	`428`	`shipper.WithSkipCorruptedBlocks(conf.shipper.skipCorruptedBlocks),`
	`429`	`+ shipper.WithUploadConcurrency(conf.shipper.uploadConcurrency),`
`429`	`430`	`)`
`430`	`431`
`431`	`432`	`return runutil.Repeat(30*time.Second, ctx.Done(), func() error {`
Original file line number	Diff line number	Diff line change
`@@ -166,9 +166,10 @@ type bucketMarkBlockConfig struct {`
`166`	`166`	`}`
`167`	`167`
`168`	`168`	`type bucketUploadBlocksConfig struct {`
`169`		`- path string`
`170`		`- labels []string`
`171`		`- uploadCompacted bool`
	`169`	`+ path string`
	`170`	`+ labels []string`
	`171`	`+ uploadCompacted bool`
	`172`	`+ uploadConcurrency int`
`172`	`173`	`}`
`173`	`174`
`174`	`175`	`func (tbc bucketVerifyConfig) registerBucketVerifyFlag(cmd extkingpin.FlagClause) bucketVerifyConfig {`
`@@ -302,6 +303,7 @@ func (tbc *bucketUploadBlocksConfig) registerBucketUploadBlocksFlag(cmd extkingp`
`302`	`303`	`cmd.Flag("path", "Path to the directory containing blocks to upload.").Default("./data").StringVar(&tbc.path)`
`303`	`304`	`cmd.Flag("label", "External labels to add to the uploaded blocks (repeated).").PlaceHolder("key=\"value\"").StringsVar(&tbc.labels)`
`304`	`305`	`cmd.Flag("shipper.upload-compacted", "If true shipper will try to upload compacted blocks as well.").Default("false").BoolVar(&tbc.uploadCompacted)`
	`306`	`+ cmd.Flag("shipper.upload-concurrency", "Number of goroutines to use when uploading block files to object storage.").Default("5").IntVar(&tbc.uploadConcurrency)`
`305`	`307`
`306`	`308`	`return tbc`
`307`	`309`	`}`
`@@ -1512,6 +1514,7 @@ func registerBucketUploadBlocks(app extkingpin.AppClause, objStoreConfig *extfla`
`1512`	`1514`	`shipper.WithMetaFileName(shipper.DefaultMetaFilename),`
`1513`	`1515`	`shipper.WithLabels(func() labels.Labels { return lset }),`
`1514`	`1516`	`shipper.WithUploadCompacted(tbc.uploadCompacted),`
	`1517`	`+ shipper.WithUploadConcurrency(tbc.uploadConcurrency),`
`1515`	`1518`	`)`
`1516`	`1519`
`1517`	`1520`	`ctx, cancel := context.WithCancel(context.Background())`