kafka replay speed: add bytes limit for inflight fetch requests

[dimitarvdimitrov]

Problem

OOMs happen because the concurrency * records-per-fetch product is too large. We hold too many records in memory and don't consume them as fast as we fetch them. The goal is to have a setting to control the memory consumption of the process. In the future it may be scaled with the ingester's memory request for example.

Proposal

Instead of configuring records-per-fetch we should configure the maximum memory for all fetchers. We can work backwards to devise the number of records per fetchWant from the number of concurrent fetchers.

I propose to remove these flags

-ingest-storage.kafka.ongoing-records-per-fetch=30
-ingest-storage.kafka.startup-records-per-fetch=512

and replace them with a single flag

-ingest-storage.kafka.max-buffered-bytes=100000000 # 100MB

Each fetchWant still needs to have a well defined startOffset and endOffset. The number of records of each fetchWant will be ($max_inflight_bytes / $startup_concurrency) / $bytes_per_record. We already track the average bytes_per_record of the last few fetches, but the value there can be volatile. Maybe we can also invest in a better tracking of average bytes.

What this PR does

This PR is the first step to doing the above: adding a limit for the inflight bytes.

-ingest-storage.kafka.max-buffered-bytes

The limit works by controlling the fetchWants. We don't dispatch a fetchWant if its MaxBytes would exceed the limit. The idea is that MaxBytes of the request are usually a good estimation of how much data we have to fetch.

This does have a caveat that if estimation is bad, we'd continue and still risk OOMing.

I didn't add two separate limits for startup and ongoing because we're currently setting them both to the same value and the short/medium-term plan is to only have a single config option.

Next steps

The next step would be to remove the records per fetch config options altogether and only rely on max bytes.

Note to reviewers

This is based on #9891 because that PR fixes flaky tests. The changes aren't otherwise dependant.

Checklist

• Tests updated.
• Documentation added.
• CHANGELOG.md updated - the order of entries should be [CHANGE], [FEATURE], [ENHANCEMENT], [BUGFIX].
• about-versioning.md updated with experimental features.

[tacole02]

Thanks so much, @dimitarvdimitrov !

[pracucci]

I can't see anyone starting/stopping this service.

[dimitarvdimitrov]

it's started here

mimir/pkg/storage/ingest/reader.go

Lines 219 to 228 in 314647e

	r.dependencies, err = services.NewManager(r.committer, r.offsetReader, r.consumedOffsetWatcher, startOffsetReader, r.metrics)
	if err != nil {
	return errors.Wrap(err, "creating service manager")
	}
	// Use context.Background() because we want to stop all dependencies when the PartitionReader stops
	// instead of stopping them when ctx is cancelled and while the PartitionReader is still running.
	err = services.StartManagerAndAwaitHealthy(context.Background(), r.dependencies)
	if err != nil {
	return errors.Wrap(err, "starting service manager")
	}

and then stopped here

mimir/pkg/storage/ingest/reader.go

Lines 280 to 285 in 314647e

	func (r *PartitionReader) stopDependencies() error {
	if r.dependencies != nil {
	if err := services.StopManagerAndAwaitStopped(context.Background(), r.dependencies); err != nil {
	return errors.Wrap(err, "stopping service manager")
	}
	}

[pracucci]

Nice work and nice tests! Pre-approving, but please remember to start/stop the reader metrics service.

[pracucci]

We've lost this if check (removeNextResult assumes the list is non empty). Is it a problem?

To my understanding it's not a problem because, with the new logic, we have the guarantee that if refillBufferedResult is valued that there's at least 1 item in the list (the fetch we're currently reading from, which is set to refillBufferedResult itself). Is my understanding correct?

[dimitarvdimitrov]

that's right. Now instead of keeping nextResult as state we compute it on every iteration. The invariants from before still hold