kafkaexporter: Opt-in to use franz-go client #40364
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| # Use this changelog template to create an entry for release notes. | ||
|
|
||
| # One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' | ||
| change_type: 'enhancement' | ||
|
|
||
| # The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver) | ||
| component: 'kafkaexporter' | ||
|
|
||
| # A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). | ||
| note: "Add an Alpha feature gate `exporter.kafkaexporter.UseFranzGoClient` to use franz-go in the Kafka exporter for better performance." | ||
|
|
||
| # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. | ||
| issues: [40364] | ||
|
|
||
| # (Optional) One or more lines of additional information to render under the primary note. | ||
| # These lines will be padded with 2 spaces and then inserted directly into the document. | ||
| # Use pipe (|) for multiline entries. | ||
| subtext: | | ||
| This change adds an experimental opt-in support to use the franz-go client in the Kafka exporter. | ||
| The franz-go client is a high-performance Kafka client that can improve the performance of the Kafka exporter. | ||
| The default client remains sarama, which is used by the Kafka receiver and other components. | ||
| The franz-go client can be enabled by setting `client_type=franz-go` in the configuration. | ||
|
|
||
| # If your change doesn't affect end users or the exported elements of any package, | ||
| # you should instead start your pull request title with [chore] or use the "Skip Changelog" label. | ||
| # Optional: The change log or logs in which this entry should be included. | ||
| # e.g. '[user]' or '[user, api]' | ||
| # Include 'user' if the change is relevant to end users. | ||
| # Include 'api' if there is a change to a library API. | ||
| # Default: '[user]' | ||
| change_logs: [user] | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| // Package kafkaclient provides implementations of Kafka producers using | ||
| // different client libraries (Sarama, Franz-go). | ||
| package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient" | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
|
|
||
| "github.com/twmb/franz-go/pkg/kgo" | ||
| ) | ||
|
|
||
| // FranzSyncProducer is a wrapper around the franz-go client that implements | ||
| // the Producer interface. Allowing us to use the franz-go client while | ||
| // maintaining compatibility with the existing Kafka exporter code. | ||
| type FranzSyncProducer struct { | ||
| client *kgo.Client | ||
| metadataKeys []string | ||
| } | ||
|
|
||
| // NewFranzSyncProducer Franz-go producer from a kgo.Client and a Messenger. | ||
| func NewFranzSyncProducer(client *kgo.Client, | ||
| metadataKeys []string, | ||
| ) FranzSyncProducer { | ||
| return FranzSyncProducer{ | ||
| client: client, | ||
| metadataKeys: metadataKeys, | ||
| } | ||
| } | ||
|
|
||
| // ExportData sends a batch of messages to Kafka | ||
| func (p FranzSyncProducer) ExportData(ctx context.Context, msgs Messages) error { | ||
marclop marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| messages := makeFranzMessages(msgs) | ||
| setMessageHeaders(ctx, messages, p.metadataKeys, | ||
| func(key string, value []byte) kgo.RecordHeader { | ||
| return kgo.RecordHeader{Key: key, Value: value} | ||
| }, | ||
| func(m *kgo.Record) []kgo.RecordHeader { return m.Headers }, | ||
| func(m *kgo.Record, h []kgo.RecordHeader) { m.Headers = h }, | ||
| ) | ||
| result := p.client.ProduceSync(ctx, messages...) | ||
| var errs []error | ||
| for _, r := range result { | ||
| if r.Err != nil { | ||
| errs = append(errs, r.Err) | ||
| } | ||
| } | ||
| return errors.Join(errs...) | ||
| } | ||
|
|
||
| // Close shuts down the producer and flushes any remaining messages. | ||
| func (p FranzSyncProducer) Close() error { | ||
| p.client.Close() | ||
| return nil | ||
| } | ||
|
|
||
| func makeFranzMessages(messages Messages) []*kgo.Record { | ||
| msgs := make([]*kgo.Record, 0, messages.Count) | ||
| for _, msg := range messages.TopicMessages { | ||
| for _, message := range msg.Messages { | ||
| msg := &kgo.Record{Topic: msg.Topic} | ||
| if message.Key != nil { | ||
| msg.Key = message.Key | ||
| } | ||
| if message.Value != nil { | ||
| msg.Value = message.Value | ||
| } | ||
| msgs = append(msgs, msg) | ||
| } | ||
| } | ||
| return msgs | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient" | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "go.opentelemetry.io/collector/client" | ||
| ) | ||
|
|
||
| // metadataToHeaders converts metadata from the context into a slice of headers using the provided header constructor. | ||
| // This function is generic and can be used for both Sarama and Franz-go header types. | ||
| func metadataToHeaders[H any](ctx context.Context, keys []string, | ||
| makeHeader func(key string, value []byte) H, | ||
| ) []H { | ||
| if len(keys) == 0 { | ||
| return nil | ||
| } | ||
| info := client.FromContext(ctx) | ||
| headers := make([]H, 0, len(keys)) | ||
| for _, key := range keys { | ||
| valueSlice := info.Metadata.Get(key) | ||
| for _, v := range valueSlice { | ||
| headers = append(headers, makeHeader(key, []byte(v))) | ||
| } | ||
| } | ||
| return headers | ||
| } | ||
|
|
||
| // setMessageHeaders is a generic helper for setting headers on a slice of messages. | ||
| // - messages: the messages to set headers on | ||
| // - ctx: context for extracting metadata | ||
| // - metadataKeys: which metadata keys to extract | ||
| // - makeHeader: constructs the header type for the target client (Sarama/Franz-go) | ||
| // - getHeaders: gets the headers from a message | ||
| // - setHeaders: sets the headers on a message | ||
| // Usage example (Sarama): | ||
| // | ||
| // setMessageHeaders(ctx, allMessages, keys, makeHeader, getHeaders, setHeaders) | ||
| func setMessageHeaders[M any, H any](ctx context.Context, | ||
| messages []M, | ||
| metadataKeys []string, | ||
| makeHeader func(key string, value []byte) H, | ||
| getHeaders func(M) []H, | ||
| setHeadersFunc func(M, []H), | ||
|
Comment on lines
+44
to
+46
There was a problem hiding this comment. Is there anyway around needing to pass callback functions here? There was a problem hiding this comment. I can refactor this once we permanently remove Sarama to not have this abstraction. Currently, this ensures the same logic is applied for both adapters. I don't love it but it seems like the best approach to apply this generic flow consistently. |
||
| ) { | ||
| setHeaders( | ||
| messages, | ||
| metadataToHeaders(ctx, metadataKeys, makeHeader), | ||
| getHeaders, | ||
| setHeadersFunc, | ||
| ) | ||
| } | ||
|
|
||
| // setHeaders sets or appends headers on each message in messages using the provided get/set functions. | ||
| func setHeaders[M any, H any](messages []M, headers []H, | ||
| getHeaders func(M) []H, | ||
| setHeaders func(M, []H), | ||
| ) { | ||
| if len(headers) == 0 || len(messages) == 0 { | ||
| return | ||
| } | ||
| for i := range messages { | ||
| h := getHeaders(messages[i]) | ||
| if len(h) == 0 { | ||
| setHeaders(messages[i], headers) | ||
| } else { | ||
| setHeaders(messages[i], append(h, headers...)) | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package kafkaclient | ||
|
|
||
| import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/marshaler" | ||
|
|
||
| // Messages is a collection of messages (with count) to be sent to Kafka. | ||
| type Messages struct { | ||
| // Count is the total number of messages across all topics. | ||
| // Populating this field allows the downstream method to preallocate | ||
| // the slice of messages, which can improve performance. | ||
| Count int | ||
| TopicMessages []TopicMessages | ||
| } | ||
|
|
||
| // TopicMessages represents a collection of messages for a specific topic. | ||
| type TopicMessages struct { | ||
| Topic string | ||
| Messages []marshaler.Message | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.