loki.write: implement sharding

[kalleep]

PR Description

This PR implements queue_config for the loki.write component, enabling users to configure queue-based batching and parallel processing. The implementation introduces a new sharding architecture that distributes log entries across multiple parallel queues based on label fingerprints. This implementation is based on Prometheus rw sharding.

The shards implementation is used with both "normal" clients and "WAL" clients. So we get rid of a lot of duplicated logic.

Before this pr we had a queue_config block that was only used when WAL was enabled. It is now always used and will affect clients regardless.

Currently no automatic "resharding" is implemented. Implementing this without the WAL will most likely be pretty primitive. So for now min_shards is the only configurable value until we address this.

Ideally we would move a couple of attributes from endpoint block to queue_config block to closer match prometheus.remote_write. But we can't do that without doing a breaking change. These attributes are:

• retry_on_http_429
• max_backoff_period
• min_backoff_period
• batch_size
• batch_wait

Which issue(s) this PR fixes

Part of: #4728

Notes to the Reviewer

I moved wal writer ownership into client.Manager. No need to expose it to loki.write component.
I plan to work on resharding in followup pr

PR Checklist

• CHANGELOG.md updated
• Documentation added
• Tests updated
• Config converters updated

[github-actions]

💻 Deploy preview available (loki.write: implement sharding):

• https://deploy-preview-alloy-4882-zb444pucvq-vp.a.run.app/docs/alloy/latest/

[Copilot]

Pull Request Overview

This PR implements queue-based sharding for the loki.write component, introducing a new architecture that distributes log entries across multiple parallel queues based on label fingerprints. The implementation unifies the handling of both normal and WAL-enabled clients through a shared shards structure, eliminating significant code duplication. The queue_config block, previously WAL-only, now applies to all endpoints and controls both queue capacity and shard count.

Key changes:

• Introduces shards.go with new sharding architecture for parallel processing via multiple queues
• Refactors WAL and fanout clients to use the shared shards implementation, removing ~500 lines of duplicated code
• Adds min_shards configuration parameter to control parallelism level

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
internal/component/loki/write/types.go	Adds MinShards field to QueueConfig and updates documentation to reflect queue config is now always used
internal/component/common/loki/client/shards.go	New file implementing the core sharding logic with queue management and parallel batch sending
internal/component/common/loki/client/shards_test.go	Comprehensive test coverage for queue operations including append, drain, and flush/shutdown scenarios
internal/component/common/loki/client/consumer_wal.go	Refactored to delegate batching and sending to the shards implementation, significantly simplified
internal/component/common/loki/client/consumer_fanout.go	Refactored to use shards implementation, removing duplicated send/batch logic
internal/component/common/loki/client/config.go	Adds MinShards field definition to QueueConfig struct
docs/sources/reference/components/loki/loki.write.md	Documents the new min_shards parameter and clarifies queue config usage

[thampiotr]

I still think we need to get a coherent story with naming established and then make sure it is reflected in docs and in the code. But we're on the right track now.

[thampiotr]

Shouldn't this continue to be experimental?

[kalleep]

Yeah we could keep this as experimental. But we would always use this config after this pr.

What would be considered experimental with it would be naming and changing defaults I guess.

[thampiotr]

Suggested change

	| `capacity` | `string` | Controls the size of the underlying send queue buffer. This setting should be considered a worst-case scenario of memory consumption, in which all enqueued batches are full. | `10MiB` | no |
	| `capacity` | `string` | Controls the size of the underlying send queue buffer of each shard. Consider this setting as the worst-case scenario of memory consumption, in which all enqueued batches are full. | `10MiB` | no |

What does it even mean 'all enqueued batches are full'? Shouldn't it say that it's the total size of all the enqueued batches instead?

[kalleep]

Yeah this was there before and I did not check / alter it.

But essentially whenever the capacity is full that means that the queue of batches is full and we cannot enqueue another one so we would block here until we get more capacity

[thampiotr]

Should we consider calling it something more intuitive than shards? I feel this term is just inherited from Prometheus 🤔

[kalleep]

Discussed this in slack, naming is fine because it would match what we use in prometheus.remote_write

[thampiotr]

We call it queue_config, but we are actually configuring both sharding and queuing in one block.

Maybe this should be called something like parallelism in write.queue?

[kalleep]

Discussed this in slack, naming is fine because it would match what we use in prometheus.remote_write

[kalleep]

In ffb2bec I renamed the two clients implementation we have to endpoint and walEndpoint.

I will look into the other ways we discussed structuring this but will have this as a fallback if that don't work out

[Copilot]

Pull Request Overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated 11 comments.

Comments suppressed due to low confidence (1)

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

• sync.WaitGroup does not have a Go method. The standard library's sync.WaitGroup has Add(), Done(), and Wait() methods. This should be:

stopWG.Add(1)
go func() {
    defer stopWG.Done()
    pair.Stop(drain)
}()

		stopWG.Go(func() {
			pair.Stop(drain)
		})

[kalleep]

@thampiotr I update the pr with an attempt to share stuff between non WAL and WAL implementation.

So we now have one endpoint struct that will handle shards. This implemntation has one method enqueue that will try to forward entry to shards for queuing with retries. If endpoint is stopped any attempts will fail end return false.

We can use this directly in Fanout and we no longer need channels between fanout and endpoint.

For WAL implementation I renamed it walEndpointAdapter. This implements the necessary interface that watcher expected and will just call enqueue on endpoint, that internally handles retries etc.

Naming could be a bit off but this would be an option we can go with. Let me know what you think :)

[Copilot]

Pull Request Overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 4 comments.

Comments suppressed due to low confidence (1)

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

• sync.WaitGroup does not have a Go() method. This code will not compile. You should either:

1. Use stopWG.Add(1) and go func(p endpointWatcherPair) { defer stopWG.Done(); p.Stop(drain) }(pair) pattern (note: capture pair as parameter to avoid closure issues)
2. Or use golang.org/x/sync/errgroup which has a Go() method

		stopWG.Go(func() {
			pair.Stop(drain)
		})

[clayton-cornell]

Nothing really stands out for the small doc changes. It's good as-is.