Add MinShards and configure number of shards based on that

Author: kalleep

internal/component/common/loki/client/client.go

@@ -68,12 +68,13 @@ func newClient(metrics *Metrics, cfg Config, logger log.Logger) (*client, error)
 		cancel:  cancel,
 	}
 
+	c.shards.start(cfg.Queue.MinShards)
+
 	c.wg.Go(func() { c.run() })
 	return c, nil
 }
 
 func (c *client) run() {
-	c.shards.start(1)
 
 	for {
 		select {

internal/component/common/loki/client/config.go

@@ -46,6 +46,9 @@ type QueueConfig struct {
 	// is the 1 MiB default, and a capacity of 100 MiB, the underlying buffered channel would buffer up to 100 batches.
 	Capacity int
 
+	// MinShards is the minimum amount of concurrent shards sending batches to the endpoint.
+	MinShards int
+
 	// DrainTimeout controls the maximum time that draining the send queue can take.
 	DrainTimeout time.Duration
 }

internal/component/common/loki/client/queue_client.go

@@ -81,7 +81,7 @@ func newQueueClient(metrics *Metrics, qcMetrics *QueueClientMetrics, cfg Config,
 		markerHandler: markerHandler,
 	}
 
-	c.shards.start(1)
+	c.shards.start(cfg.Queue.MinShards)
 
 	return c, nil
 }

internal/component/loki/write/types.go

@@ -70,17 +70,20 @@ func (r *EndpointOptions) Validate() error {
 	return nil
 }
 
-// QueueConfig controls how the queue logs remote write client is configured. Note that this client is only used when the
-// loki.write component has WAL support enabled.
+// QueueConfig controls how the queue logs remote write client is configured.
 type QueueConfig struct {
-	Capacity     units.Base2Bytes `alloy:"capacity,attr,optional"`
-	DrainTimeout time.Duration    `alloy:"drain_timeout,attr,optional"`
+	Capacity units.Base2Bytes `alloy:"capacity,attr,optional"`
+
+	MinShards int `alloy:"min_shards,attr,optional"`
+
+	DrainTimeout time.Duration `alloy:"drain_timeout,attr,optional"`
 }
 
 // SetToDefault implements syntax.Defaulter.
 func (q *QueueConfig) SetToDefault() {
 	*q = QueueConfig{
 		Capacity:     10 * units.MiB, // considering the default BatchSize of 1MiB, this gives us a default buffered channel of size 10
+		MinShards:    1,
 		DrainTimeout: 15 * time.Second,
 	}
 }