Add experimental support to write incoming data to a Kafka-compatible backend #6888

pracucci · 2023-12-11T13:19:25Z

What this PR does

We're building a prototype of an alternative Mimir architecture where the write and read path are fully decoupled, with a Kafka-compatible backend in between. This is going to be a multi-quarter effort and we would like to progressively upstream code changes while building it. The idea is that we'll do our best to keep these changes isolated from the rest of Mimir as much as possible, with few integration points.

In this PR I'm proposing to upstream a basic support to write incoming requests from distributor to a Kafka-compatible backend.

Notes:

Configuration is hidden in the auto-generated documentation to avoid confusion to final users while we build the prototype.
No CHANGELOG entry while building the prototype, for the same reason as above: to avoid confusion to final users.
The plugging in the distributor is not the final one. Currently it's a very hacky way to partition data, while @pstibrany is working on the definitive partitioning scheme.
This PR also introduces the franz-go Kafka client, which is the library we selected after testing the segmentio one and this one. Based on our tests, the franz-go library looks more evoluted, mature and stable than the segmentio one.

Which issue(s) this PR fixes or relates to

N/A

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.

dimitarvdimitrov

mostly LGTM, i have one more important remark about the timeouts, but otherwise looks good. I quite liked TestWriter_WriteSync

pkg/distributor/distributor.go

pkg/storage/ingest/config.go

pkg/storage/ingest/util.go

pkg/storage/ingest/writer.go

pkg/storage/ingest/config.go

pkg/storage/ingest/writer.go

dimitarvdimitrov · 2023-12-11T14:38:46Z

pkg/storage/ingest/writer.go

+		//   after being sent on the network. The actual timeout is increased by the configured overhead.
+		kgo.RecordRetries(math.MaxInt64),
+		kgo.RecordDeliveryTimeout(w.writerCfg.KafkaWriteTimeout),
+		kgo.ProduceRequestTimeout(w.writerCfg.KafkaWriteTimeout),


shouldn't we set this to something lower? Otherwise I can see how a network timeout would result in no retries:

try to send

wait for ProduceRequestTimeout (== w.writerCfg.KafkaWriteTimeout)

time out

try to send again

fail, it's already past RecordDeliveryTimeout

Should we set it to something like w.writerCfg.KafkaWriteTimeout / 3?

Discussed offline.

The TL;DR is that it depends on the actual failure scenario.

If the backend is slow, then an higher ProduceRequestTimeout increases the chances of a successful request (in other words, one try of 10s is better than 2x5s tries because maybe just waiting longer fixes it).

If the backend is unhealthy, then a shorter ProduceRequestTimeout is better because it will retry within the RecordDeliveryTimeout but the retry will be successful only if the meanwhile the cluster metadata has been updated and the replica owning a given partition has been moved to another broker, otherwise the client will just keep trying to connect to the previous (unhealthy) one. In a setup where these timeouts are relatively low, it may not be that common that the unhealthy replica has been actually detected as unhealthy and the replica owner for a given partition moved.

dimitarvdimitrov · 2023-12-11T14:50:15Z

pkg/storage/ingest/writer_test.go

+		received := mimirpb.WriteRequest{}
+		require.NoError(t, received.Unmarshal(fetches.Records()[0].Value))
+		require.Len(t, received.Timeseries, len(multiSeries))
+
+		for idx, expected := range multiSeries {
+			assert.Equal(t, expected.Labels, received.Timeseries[idx].Labels)
+			assert.Equal(t, expected.Samples, received.Timeseries[idx].Samples)
+		}
+


reading these tests I'm not sure whether the interface of Writer shouldn't just accept a byte slice instead of managing mimir's protocol buffers.

I don't insist on this, but thought it might give better separation and reduce scope a little bit

My take is: the byte-level interface is the Kafka client. Our Writer has a domain-level interface, which means we write domain-level data structures (so timeseries & co).

pkg/storage/ingest/writer_test.go

… backend Signed-off-by: Marco Pracucci <[email protected]>

Signed-off-by: Marco Pracucci <[email protected]>

dimitarvdimitrov

LGTM, thanks for addressing my comments

twmb · 2023-12-13T15:38:22Z

pkg/storage/ingest/writer.go

+		// By default, the Kafka client allows 1 Produce in-flight request per broker. Disabling write idempotency
+		// (which we don't need), we can increase the max number of in-flight Produce requests per broker. A higher
+		// number of in-flight requests, in addition to short buffering ("linger") in client side before firing the
+		// next Produce request allows us to reduce the end-to-end latency.
+		//
+		// The result of the multiplication of producer linger and max in-flight requests should match the maximum
+		// Produce latency expected by the Kafka backend in a steady state. For example, 50ms * 20 requests = 1s,
+		// which means the Kafka client will keep issuing a Produce request every 50ms as far as the Kafka backend
+		// doesn't take longer than 1s to process them (if it takes longer, the client will buffer data and stop
+		// issuing new Produce requests until some previous ones complete).
+		kgo.DisableIdempotentWrite(),


I think this is true in other clients, but in franz-go, disabling idempotency forces the client to only issue one request at a time -- franz-go favors not duplicating data. With idempotency, franz-go allows 5 requests per broker (technically 4 due to some internal accounting but it's close enough).

Oh wait I missed MaxProduceRequestsInflightPerBroker just two lines down, my mistake. Nice!

Thanks for looking at it and your feedback!

dimitarvdimitrov reviewed Dec 11, 2023

View reviewed changes

pracucci added 5 commits December 11, 2023 17:11

Add experimental support to write incoming data to a Kafka-compatible…

e00d8fc

… backend Signed-off-by: Marco Pracucci <[email protected]>

Use github.com/grafana/regexp

0c51407

Signed-off-by: Marco Pracucci <[email protected]>

Added missing license headers

01061e9

Signed-off-by: Marco Pracucci <[email protected]>

Fix goroutine leaks in Writer tests

31e6740

Signed-off-by: Marco Pracucci <[email protected]>

Addressed nits from code review

95b49c7

Signed-off-by: Marco Pracucci <[email protected]>

pracucci force-pushed the add-kafka-producer-support branch from 641cda1 to 95b49c7 Compare December 11, 2023 16:11

Merge WriterConfig into KafkaConfig

ab77d27

Signed-off-by: Marco Pracucci <[email protected]>

pracucci marked this pull request as ready for review December 11, 2023 16:35

pracucci requested review from grafanabot and a team as code owners December 11, 2023 16:35

dimitarvdimitrov approved these changes Dec 11, 2023

View reviewed changes

pracucci merged commit 24591ae into main Dec 12, 2023

pracucci deleted the add-kafka-producer-support branch December 12, 2023 05:04

twmb reviewed Dec 13, 2023

View reviewed changes

dimitarvdimitrov mentioned this pull request Dec 13, 2023

ingester: add experimental support for consuming records from kafka #6929

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add experimental support to write incoming data to a Kafka-compatible backend #6888

Add experimental support to write incoming data to a Kafka-compatible backend #6888

pracucci commented Dec 11, 2023 •

edited

Loading

dimitarvdimitrov left a comment

dimitarvdimitrov Dec 11, 2023

pracucci Dec 11, 2023 •

edited

Loading

dimitarvdimitrov Dec 11, 2023

pracucci Dec 11, 2023 •

edited

Loading

dimitarvdimitrov left a comment

twmb Dec 13, 2023

twmb Dec 13, 2023

pracucci Dec 13, 2023

Add experimental support to write incoming data to a Kafka-compatible backend #6888

Add experimental support to write incoming data to a Kafka-compatible backend #6888

Conversation

pracucci commented Dec 11, 2023 • edited Loading

What this PR does

Which issue(s) this PR fixes or relates to

Checklist

dimitarvdimitrov left a comment

Choose a reason for hiding this comment

dimitarvdimitrov Dec 11, 2023

Choose a reason for hiding this comment

pracucci Dec 11, 2023 • edited Loading

Choose a reason for hiding this comment

dimitarvdimitrov Dec 11, 2023

Choose a reason for hiding this comment

pracucci Dec 11, 2023 • edited Loading

Choose a reason for hiding this comment

dimitarvdimitrov left a comment

Choose a reason for hiding this comment

twmb Dec 13, 2023

Choose a reason for hiding this comment

twmb Dec 13, 2023

Choose a reason for hiding this comment

pracucci Dec 13, 2023

Choose a reason for hiding this comment

pracucci commented Dec 11, 2023 •

edited

Loading

pracucci Dec 11, 2023 •

edited

Loading

pracucci Dec 11, 2023 •

edited

Loading