Update naming and comments

Author: kalleep

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

@@ -33,11 +33,11 @@ type Config struct {
 	// prevent HOL blocking in multitenant deployments.
 	DropRateLimitedBatches bool
 
-	// Queue controls configuration parameters specific to the queue client
-	Queue QueueConfig
+	// QueueConfig controls how shards and queues are configured for endpoints.
+	QueueConfig QueueConfig
 }
 
-// QueueConfig holds configurations for the queue-based remote-write client.
+// QueueConfig controls how shards and queue are configured for client.
 type QueueConfig struct {
 	// Capacity is the worst case size in bytes desired for the send queue. This value is used to calculate the size of
 	// the buffered channel used underneath. The worst case scenario assumed is that every batch buffered in full, hence
@@ -50,6 +50,6 @@ type QueueConfig struct {
 	// MinShards is the minimum number of concurrent shards sending batches to the endpoint.
 	MinShards int
 
-	// DrainTimeout controls the maximum time that draining the send queue can take.
+	// DrainTimeout controls the maximum time that draining the queue can take.
 	DrainTimeout time.Duration
 }

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

@@ -138,7 +138,7 @@ func newEndpoint(metrics *Metrics, cfg Config, logger log.Logger) (*endpoint, er
 		cancel:  cancel,
 	}
 
-	c.shards.start(cfg.Queue.MinShards)
+	c.shards.start(cfg.QueueConfig.MinShards)
 
 	c.wg.Go(func() { c.run() })
 	return c, nil

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

@@ -628,7 +628,7 @@ func newServerAndEndpointConfig(t *testing.T) (Config, chan utils.RemoteWriteReq
 		BackoffConfig: backoff.Config{
 			MaxRetries: 0,
 		},
-		Queue: QueueConfig{
+		QueueConfig: QueueConfig{
 			Capacity:     10, // buffered channel of size 10
 			DrainTimeout: time.Second * 10,
 		},

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

@@ -164,7 +164,7 @@ func newWalEndpoint(metrics *Metrics, wcMetrics *WALEndpointMetrics, cfg Config,
 		markerHandler: markerHandler,
 	}
 
-	c.shards.start(cfg.Queue.MinShards)
+	c.shards.start(cfg.QueueConfig.MinShards)
 
 	return c, nil
 }

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

@@ -255,7 +255,7 @@ func TestWALEndpoint(t *testing.T) {
 				BackoffConfig: backoff.Config{MinBackoff: 5 * time.Second, MaxBackoff: 10 * time.Second, MaxRetries: 1},
 				Timeout:       1 * time.Second,
 				TenantID:      "",
-				Queue:         tc.queueConfig,
+				QueueConfig:   tc.queueConfig,
 			}
 
 			logger := log.NewLogfmtLogger(os.Stdout)
@@ -394,7 +394,7 @@ func runWALEndpointBenchCase(b *testing.B, bc testCase, mhFactory func(t *testin
 		BackoffConfig: backoff.Config{MinBackoff: 5 * time.Second, MaxBackoff: 10 * time.Second, MaxRetries: 1},
 		Timeout:       1 * time.Second,
 		TenantID:      "",
-		Queue: QueueConfig{
+		QueueConfig: QueueConfig{
 			Capacity:     1000, // queue size of 100
 			DrainTimeout: time.Second * 10,
 		},
@@ -485,7 +485,7 @@ func runEndpointBenchCase(b *testing.B, bc testCase) {
 		BackoffConfig: backoff.Config{MinBackoff: 5 * time.Second, MaxBackoff: 10 * time.Second, MaxRetries: 1},
 		Timeout:       1 * time.Second,
 		TenantID:      "",
-		Queue: QueueConfig{
+		QueueConfig: QueueConfig{
 			Capacity:     1000, // queue size of 100
 			DrainTimeout: time.Second * 10,
 		},

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

@@ -41,7 +41,7 @@ func newQueue(metrics *Metrics, logger log.Logger, cfg Config) *queue {
 	// the channel capacity would be calculated as: bufferChannelSize = Capacity / BatchSize.
 	// For example, assuming BatchSize is the 1 MiB default and Capacity is 100 MiB,
 	// the underlying buffered channel would buffer up to 100 batches.
-	capacity := max(cfg.Queue.Capacity/max(cfg.BatchSize, 1), 1)
+	capacity := max(cfg.QueueConfig.Capacity/max(cfg.BatchSize, 1), 1)
 
 	return &queue{
 		cfg:     cfg,
@@ -278,7 +278,7 @@ func (s *shards) stop() {
 	select {
 	case <-s.done:
 		return
-	case <-time.After(s.cfg.Queue.DrainTimeout):
+	case <-time.After(s.cfg.QueueConfig.DrainTimeout):
 	}
 
 	// Perform hard shutdown

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

@@ -24,7 +24,7 @@ func TestQueue_append(t *testing.T) {
 	// a queue with 8 bytes batches and only one batch can queued.
 	q := newQueue(NewMetrics(prometheus.NewRegistry()), log.NewNopLogger(), Config{
 		BatchSize: 8,
-		Queue: QueueConfig{
+		QueueConfig: QueueConfig{
 			Capacity: 8,
 		},
 	})
@@ -61,7 +61,7 @@ func TestQueue_drain(t *testing.T) {
 		// a queue with 8 bytes batches and only one batch can queued at any given time.
 		q := newQueue(NewMetrics(prometheus.NewRegistry()), log.NewNopLogger(), Config{
 			BatchSize: 8,
-			Queue: QueueConfig{
+			QueueConfig: QueueConfig{
 				Capacity: 8,
 			},
 		})
@@ -83,7 +83,7 @@ func TestQueue_drain(t *testing.T) {
 		q := newQueue(NewMetrics(prometheus.NewRegistry()), log.NewNopLogger(), Config{
 			BatchSize: 8,
 			BatchWait: 10 * time.Second,
-			Queue: QueueConfig{
+			QueueConfig: QueueConfig{
 				Capacity: 8,
 			},
 		})
@@ -106,7 +106,7 @@ func TestQueue_flushAndShutdown(t *testing.T) {
 		// a queue with 8 bytes batches and only one batch can queued at any given time.
 		q := newQueue(NewMetrics(prometheus.NewRegistry()), log.NewNopLogger(), Config{
 			BatchSize: 8,
-			Queue: QueueConfig{
+			QueueConfig: QueueConfig{
 				Capacity: 8,
 			},
 		})
@@ -144,7 +144,7 @@ func TestQueue_flushAndShutdown(t *testing.T) {
 		// a queue with 8 bytes batches and only one batch can queued at any given time.
 		q := newQueue(NewMetrics(prometheus.NewRegistry()), log.NewNopLogger(), Config{
 			BatchSize: 8,
-			Queue: QueueConfig{
+			QueueConfig: QueueConfig{
 				Capacity: 8,
 			},
 		})

internal/component/loki/write/types.go

@@ -70,7 +70,7 @@ func (r *EndpointOptions) Validate() error {
 	return nil
 }
 
-// QueueConfig controls how shards and queue are configured for client.
+// QueueConfig controls how shards and queue are configured for endpoint.
 type QueueConfig struct {
 	Capacity     units.Base2Bytes `alloy:"capacity,attr,optional"`
 	MinShards    int              `alloy:"min_shards,attr,optional"`
@@ -86,7 +86,7 @@ func (q *QueueConfig) SetToDefault() {
 	}
 }
 
-func (args Arguments) convertClientConfigs() []client.Config {
+func (args Arguments) convertEndpointConfigs() []client.Config {
 	var res []client.Config
 	for _, cfg := range args.Endpoints {
 		url, _ := url.Parse(cfg.URL)
@@ -106,7 +106,7 @@ func (args Arguments) convertClientConfigs() []client.Config {
 			TenantID:               cfg.TenantID,
 			MaxStreams:             args.MaxStreams,
 			DropRateLimitedBatches: !cfg.RetryOnHTTP429,
-			Queue: client.QueueConfig{
+			QueueConfig: client.QueueConfig{
 				Capacity:     int(cfg.QueueConfig.Capacity),
 				MinShards:    cfg.QueueConfig.MinShards,
 				DrainTimeout: cfg.QueueConfig.DrainTimeout,

internal/component/loki/write/write.go

@@ -164,7 +164,7 @@ func (c *Component) Update(args component.Arguments) error {
 		c.consumer.Stop()
 	}
 
-	cfgs := newArgs.convertClientConfigs()
+	cfgs := newArgs.convertEndpointConfigs()
 
 	uid := alloyseed.Get().UID
 	for i := range cfgs {