receive: use async remote writing

Instead of spawning new goroutines for each peer that we want to remote write to, spawn a fixed number of worker goroutines and then schedule work on top of them. This has reduced the number of goroutines in our case about 10x-20x and the 99p of forwarding dropped from ~30s to just a few hundred milliseconds. Signed-off-by: Giedrius Statkevičius <[email protected]>
thanos-io · Jan 10, 2024 · b39d96e · b39d96e
1 parent 7794d78
commit b39d96e
Show file tree

Hide file tree

Showing 3 changed files with 202 additions and 73 deletions.
diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go
@@ -261,6 +261,7 @@ func runReceive(
 		MaxBackoff:        time.Duration(*conf.maxBackoff),
 		TSDBStats:         dbs,
 		Limiter:           limiter,
+		AsyncWorkerCount:  conf.asyncWorkerCount,
 	})
 
 	grpcProbe := prober.NewGRPC()
@@ -837,6 +838,7 @@ type receiveConfig struct {
 	writeLimitsConfig       *extflag.PathOrContent
 	storeRateLimits         store.SeriesSelectLimits
 	limitsConfigReloadTimer time.Duration
+	asyncWorkerCount        uint
 }
 
 func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
@@ -894,6 +896,7 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 
 	cmd.Flag("receive.replica-header", "HTTP header specifying the replica number of a write request.").Default(receive.DefaultReplicaHeader).StringVar(&rc.replicaHeader)
 
+	cmd.Flag("receive.async-workers", "Number of concurrent workers processing incoming remote-write requests.").Default("5").UintVar(&rc.asyncWorkerCount)
 	compressionOptions := strings.Join([]string{snappy.Name, compressionNone}, ", ")
 	cmd.Flag("receive.grpc-compression", "Compression algorithm to use for gRPC requests to other receivers. Must be one of: "+compressionOptions).Default(snappy.Name).EnumVar(&rc.compression, snappy.Name, compressionNone)