Introduce min_shards for remote write to set minimum number of shards. (#4924)

Signed-off-by: Ryota Arai <ryota.arai@gmail.com>

config/config.go

@@ -111,6 +111,7 @@ var (
 		// With a maximum of 1000 shards, assuming an average of 100ms remote write
 		// time and 100 samples per batch, we will be able to push 1M samples/s.
 		MaxShards:         1000,
+		MinShards:         1,
 		MaxSamplesPerSend: 100,
 
 		// By default, buffer 100 batches, which at 100ms per batch is 10s. At
@@ -706,6 +707,9 @@ type QueueConfig struct {
 	// Max number of shards, i.e. amount of concurrency.
 	MaxShards int `yaml:"max_shards,omitempty"`
 
+	// Min number of shards, i.e. amount of concurrency.
+	MinShards int `yaml:"min_shards,omitempty"`
+
 	// Maximum number of samples per send.
 	MaxSamplesPerSend int `yaml:"max_samples_per_send,omitempty"`
 

docs/configuration/configuration.md

@@ -1244,6 +1244,8 @@ queue_config:
   [ capacity: <int> | default = 10000 ]
   # Maximum number of shards, i.e. amount of concurrency.
   [ max_shards: <int> | default = 1000 ]
+  # Minimum number of shards, i.e. amount of concurrency.
+  [ min_shards: <int> | default = 1 ]
   # Maximum number of samples per send.
   [ max_samples_per_send: <int> | default = 100]
   # Maximum time a sample will wait in buffer.

storage/remote/queue_manager.go

@@ -177,7 +177,7 @@ func NewQueueManager(logger log.Logger, cfg config.QueueConfig, externalLabels m
 		queueName:      client.Name(),
 
 		logLimiter:  rate.NewLimiter(logRateLimit, logBurst),
-		numShards:   1,
+		numShards:   cfg.MinShards,
 		reshardChan: make(chan int),
 		quit:        make(chan struct{}),
 
@@ -326,8 +326,8 @@ func (t *QueueManager) calculateDesiredShards() {
 	numShards := int(math.Ceil(desiredShards))
 	if numShards > t.cfg.MaxShards {
 		numShards = t.cfg.MaxShards
-	} else if numShards < 1 {
-		numShards = 1
+	} else if numShards < t.cfg.MinShards {
+		numShards = t.cfg.MinShards
 	}
 	if numShards == t.numShards {
 		return