kafkaexporter: Allow topics from metadata key (#40154)

#### Description
Introduces a new per-signal config option `topic_from_metadata_key` to
allow the user to produce to topics which which can be set using parts
of the request metadata. This is useful for multi-tenant use-cases.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Closes #39208

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Unit tests

<!--Describe the documentation added.-->
#### Documentation

Updated the readme with the new option.

---------

Signed-off-by: Marc Lopez Rubio <[email protected]>

Author: marclop

.chloggen/f_kafkaexporter-add-topic-from-metadata-key.yaml

@@ -0,0 +1,29 @@
+# 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: "Allow kafka exporter to produce to topics based on metadata key values"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39208]
+
+# (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: |
+  Allows the Kafka exporter to dynamically use a signal's export target topic based
+  on the value of the pipeline's metadata, allowing dynamic signal routing.
+
+# 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]

exporter/kafkaexporter/README.md

@@ -30,12 +30,15 @@ The following settings can be optionally configured:
 - `logs`
   - `topic` (default = otlp\_logs): The name of the Kafka topic to which logs will be exported.
   - `encoding` (default = otlp\_proto): The encoding for logs. See [Supported encodings](#supported-encodings).
+  - `topic_from_metadata_key` (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over `topic_from_attribute` and `topic` settings.
 - `metrics`
   - `topic` (default = otlp\_metrics): The name of the Kafka topic from which to consume metrics.
   - `encoding` (default = otlp\_proto): The encoding for metrics. See [Supported encodings](#supported-encodings).
+  - `topic_from_metadata_key` (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over `topic_from_attribute` and `topic` settings.
 - `traces`
   - `topic` (default = otlp\_spans): The name of the Kafka topic from which to consume traces.
   - `encoding` (default = otlp\_proto): The encoding for traces. See [Supported encodings](#supported-encodings).
+  - `topic_from_metadata_key` (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over `topic_from_attribute` and `topic` settings.
 - `topic` (Deprecated in v0.124.0: use `logs::topic`, `metrics::topic`, and `traces::topic`) If specified, this is used as the default topic, but will be overridden by signal-specific configuration. See [Destination Topic](#destination-topic) below for more details.
 - `topic_from_attribute` (default = ""): Specify the resource attribute whose value should be used as the message's topic. See [Destination Topic](#destination-topic) below for more details. 
 - `encoding` (Deprecated in v0.124.0: use `logs::encoding`, `metrics::encoding`, and `traces::encoding`) If specified, this is used as the default encoding, but will be overridden by signal-specific configuration. See [Supported encodings](#supported-encodings) below for more details.
@@ -139,6 +142,7 @@ exporters:
 
 The destination topic can be defined in a few different ways and takes priority in the following order:
 
-1. When `topic_from_attribute` is configured, and the corresponding attribute is found on the ingested data, the value of this attribute is used.
-2. If a prior component in the collector pipeline sets the topic on the context via the `topic.WithTopic` function (from the `github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic` package), the value set in the context is used.
-3. Finally, the `<signal>::topic` configuration is used for the signal-specific destination topic. If this is not explicitly configured, the `topic` configuration (deprecated in v0.124.0) is used as a fallback for all signals.
+1. When `<signal>.topic_from_metadata_key` is set to use a key from the request metadata, the value of this key is used as the signal specific topic.
+2. Otherwise, if `topic_from_attribute` is configured, and the corresponding attribute is found on the ingested data, the value of this attribute is used.
+3. If a prior component in the collector pipeline sets the topic on the context via the `topic.WithTopic` function (from the `github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic` package), the value set in the context is used.
+4. Finally, the `<signal>::topic` configuration is used for the signal-specific destination topic. If this is not explicitly configured, the `topic` configuration (deprecated in v0.124.0) is used as a fallback for all signals.

exporter/kafkaexporter/config.go

@@ -117,6 +117,11 @@ type SignalConfig struct {
 	//  - "otlp_logs" for logs
 	Topic string `mapstructure:"topic"`
 
+	// TopicFromMetadataKey holds the name of the metadata key to use as the
+	// topic name for this signal type. If this is set, it takes precedence
+	// over the topic name set in the topic field.
+	TopicFromMetadataKey string `mapstructure:"topic_from_metadata_key"`
+
 	// Encoding holds the encoding of messages for the signal type.
 	//
 	// Defaults to "otlp_proto".

exporter/kafkaexporter/config_test.go

@@ -87,8 +87,9 @@ func TestLoadConfig(t *testing.T) {
 				ClientConfig:    configkafka.NewDefaultClientConfig(),
 				Producer:        configkafka.NewDefaultProducerConfig(),
 				Logs: SignalConfig{
-					Topic:    "legacy_topic",
-					Encoding: "otlp_proto",
+					Topic:                "legacy_topic",
+					Encoding:             "otlp_proto",
+					TopicFromMetadataKey: "metadata_key",
 				},
 				Metrics: SignalConfig{
 					Topic:    "metrics_topic",

exporter/kafkaexporter/kafka_exporter.go

@@ -151,7 +151,7 @@ func (e *kafkaTracesMessager) marshalData(td ptrace.Traces) ([]marshaler.Message
 }
 
 func (e *kafkaTracesMessager) getTopic(ctx context.Context, td ptrace.Traces) string {
-	return getTopic(ctx, &e.config, e.config.Traces.Topic, td.ResourceSpans())
+	return getTopic(ctx, e.config.Traces, e.config.TopicFromAttribute, td.ResourceSpans())
 }
 
 func (e *kafkaTracesMessager) partitionData(td ptrace.Traces) iter.Seq2[[]byte, ptrace.Traces] {
@@ -196,7 +196,7 @@ func (e *kafkaLogsMessager) marshalData(ld plog.Logs) ([]marshaler.Message, erro
 }
 
 func (e *kafkaLogsMessager) getTopic(ctx context.Context, ld plog.Logs) string {
-	return getTopic(ctx, &e.config, e.config.Logs.Topic, ld.ResourceLogs())
+	return getTopic(ctx, e.config.Logs, e.config.TopicFromAttribute, ld.ResourceLogs())
 }
 
 func (e *kafkaLogsMessager) partitionData(ld plog.Logs) iter.Seq2[[]byte, plog.Logs] {
@@ -239,7 +239,7 @@ func (e *kafkaMetricsMessager) marshalData(md pmetric.Metrics) ([]marshaler.Mess
 }
 
 func (e *kafkaMetricsMessager) getTopic(ctx context.Context, md pmetric.Metrics) string {
-	return getTopic(ctx, &e.config, e.config.Metrics.Topic, md.ResourceMetrics())
+	return getTopic(ctx, e.config.Metrics, e.config.TopicFromAttribute, md.ResourceMetrics())
 }
 
 func (e *kafkaMetricsMessager) partitionData(md pmetric.Metrics) iter.Seq2[[]byte, pmetric.Metrics] {
@@ -270,23 +270,27 @@ type resource interface {
 
 func getTopic[T resource](
 	ctx context.Context,
-	cfg *Config,
-	defaultTopic string,
+	signalCfg SignalConfig,
+	topicFromAttribute string,
 	resources resourceSlice[T],
 ) string {
-	if cfg.TopicFromAttribute != "" {
+	if k := signalCfg.TopicFromMetadataKey; k != "" {
+		if topic := client.FromContext(ctx).Metadata.Get(k); len(topic) > 0 {
+			return topic[0]
+		}
+	}
+	if topicFromAttribute != "" {
 		for i := 0; i < resources.Len(); i++ {
-			rv, ok := resources.At(i).Resource().Attributes().Get(cfg.TopicFromAttribute)
+			rv, ok := resources.At(i).Resource().Attributes().Get(topicFromAttribute)
 			if ok && rv.Str() != "" {
 				return rv.Str()
 			}
 		}
 	}
-	contextTopic, ok := topic.FromContext(ctx)
-	if ok {
-		return contextTopic
+	if topic, ok := topic.FromContext(ctx); ok {
+		return topic
 	}
-	return defaultTopic
+	return signalCfg.Topic
 }
 
 func makeSaramaMessages(messages []marshaler.Message, topic string) []*sarama.ProducerMessage {

exporter/kafkaexporter/kafka_exporter_test.go

@@ -666,91 +666,142 @@ func TestLogsPusher_partitioning(t *testing.T) {
 
 func Test_GetTopic(t *testing.T) {
 	tests := []struct {
-		name      string
-		cfg       Config
-		ctx       context.Context
-		resource  any
-		wantTopic string
+		name               string
+		topicFromAttribute string
+		signalCfg          SignalConfig
+		ctx                context.Context
+		resource           any
+		wantTopic          string
 	}{
+		// topicFromAttribute tests.
 		{
-			name: "Valid metric attribute, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "resource-attr",
-			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
-			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
-			wantTopic: "resource-attr-val-1",
+			name:               "Valid metric attribute, return topic name",
+			topicFromAttribute: "resource-attr",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic:          "resource-attr-val-1",
 		},
 		{
-			name: "Valid trace attribute, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "resource-attr",
-			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
-			resource:  testdata.GenerateTraces(1).ResourceSpans(),
-			wantTopic: "resource-attr-val-1",
+			name:               "Valid trace attribute, return topic name",
+			topicFromAttribute: "resource-attr",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateTraces(1).ResourceSpans(),
+			wantTopic:          "resource-attr-val-1",
 		},
 		{
-			name: "Valid log attribute, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "resource-attr",
-			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
-			resource:  testdata.GenerateLogs(1).ResourceLogs(),
-			wantTopic: "resource-attr-val-1",
+			name:               "Valid log attribute, return topic name",
+			topicFromAttribute: "resource-attr",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateLogs(1).ResourceLogs(),
+			wantTopic:          "resource-attr-val-1",
 		},
 		{
-			name: "Attribute not found",
-			cfg: Config{
-				TopicFromAttribute: "nonexistent_attribute",
-			},
+			name:               "Attribute not found",
+			topicFromAttribute: "nonexistent_attribute",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                context.Background(),
+			resource:           testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic:          "defaultTopic",
+		},
+		// Nonexistent attribute tests.
+		{
+			name:               "Valid metric context, return topic name",
+			topicFromAttribute: "nonexistent_attribute",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic:          "context-topic",
+		},
+		{
+			name:               "Valid trace context, return topic name",
+			topicFromAttribute: "nonexistent_attribute",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateTraces(1).ResourceSpans(),
+			wantTopic:          "context-topic",
+		},
+		{
+			name:               "Valid log context, return topic name",
+			topicFromAttribute: "nonexistent_attribute",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                topic.WithTopic(context.Background(), "context-topic"),
+			resource:           testdata.GenerateLogs(1).ResourceLogs(),
+			wantTopic:          "context-topic",
+		},
+		// Generic known failure modes.
+		{
+			name:               "Attribute not found",
+			topicFromAttribute: "nonexistent_attribute",
+			signalCfg:          SignalConfig{Topic: "defaultTopic"},
+			ctx:                context.Background(),
+			resource:           testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic:          "defaultTopic",
+		},
+		{
+			name:      "TopicFromAttribute, return default topic",
 			ctx:       context.Background(),
+			signalCfg: SignalConfig{Topic: "defaultTopic"},
 			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
 			wantTopic: "defaultTopic",
 		},
-
+		// topicFromMetadata tests.
 		{
-			name: "Valid metric context, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "nonexistent_attribute",
+			name: "Metrics topic from metadata",
+			signalCfg: SignalConfig{
+				Topic:                "defaultTopic",
+				TopicFromMetadataKey: "metrics_topic_metadata",
 			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
+			ctx: client.NewContext(context.Background(),
+				client.Info{Metadata: client.NewMetadata(map[string][]string{
+					"metrics_topic_metadata": {"my_metrics_topic"},
+				})},
+			),
 			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
-			wantTopic: "context-topic",
-		},
-		{
-			name: "Valid trace context, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "nonexistent_attribute",
-			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
-			resource:  testdata.GenerateTraces(1).ResourceSpans(),
-			wantTopic: "context-topic",
+			wantTopic: "my_metrics_topic",
 		},
 		{
-			name: "Valid log context, return topic name",
-			cfg: Config{
-				TopicFromAttribute: "nonexistent_attribute",
+			name: "Logs topic from metadata",
+			signalCfg: SignalConfig{
+				Topic:                "defaultTopic",
+				TopicFromMetadataKey: "logs_topic_metadata",
 			},
-			ctx:       topic.WithTopic(context.Background(), "context-topic"),
+			ctx: client.NewContext(context.Background(),
+				client.Info{Metadata: client.NewMetadata(map[string][]string{
+					"logs_topic_metadata": {"my_logs_topic"},
+				})},
+			),
 			resource:  testdata.GenerateLogs(1).ResourceLogs(),
-			wantTopic: "context-topic",
+			wantTopic: "my_logs_topic",
 		},
-
 		{
-			name: "Attribute not found",
-			cfg: Config{
-				TopicFromAttribute: "nonexistent_attribute",
+			name: "Traces topic from metadata",
+			signalCfg: SignalConfig{
+				Topic:                "defaultTopic",
+				TopicFromMetadataKey: "traces_topic_metadata",
 			},
-			ctx:       context.Background(),
-			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
-			wantTopic: "defaultTopic",
+			ctx: client.NewContext(context.Background(),
+				client.Info{Metadata: client.NewMetadata(map[string][]string{
+					"traces_topic_metadata": {"my_traces_topic"},
+				})},
+			),
+			resource:  testdata.GenerateTraces(1).ResourceSpans(),
+			wantTopic: "my_traces_topic",
 		},
 		{
-			name:      "TopicFromAttribute, return default topic",
-			cfg:       Config{},
-			ctx:       context.Background(),
-			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
+			name: "metadata key not found uses default topic",
+			signalCfg: SignalConfig{
+				Topic:                "defaultTopic",
+				TopicFromMetadataKey: "key not found",
+			},
+			ctx: client.NewContext(context.Background(),
+				client.Info{Metadata: client.NewMetadata(map[string][]string{
+					"traces_topic_metadata": {"my_traces_topic"},
+				})},
+			),
+			resource:  testdata.GenerateTraces(1).ResourceSpans(),
 			wantTopic: "defaultTopic",
 		},
 	}
@@ -760,11 +811,11 @@ func Test_GetTopic(t *testing.T) {
 			topic := ""
 			switch r := tests[i].resource.(type) {
 			case pmetric.ResourceMetricsSlice:
-				topic = getTopic(tests[i].ctx, &tests[i].cfg, "defaultTopic", r)
+				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			case ptrace.ResourceSpansSlice:
-				topic = getTopic(tests[i].ctx, &tests[i].cfg, "defaultTopic", r)
+				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			case plog.ResourceLogsSlice:
-				topic = getTopic(tests[i].ctx, &tests[i].cfg, "defaultTopic", r)
+				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			}
 			assert.Equal(t, tests[i].wantTopic, topic)
 		})

exporter/kafkaexporter/testdata/config.yaml

@@ -23,6 +23,8 @@ kafka/legacy_topic:
   topic: legacy_topic
   metrics:
     topic: metrics_topic
+  logs:
+    topic_from_metadata_key: metadata_key
 kafka/legacy_encoding:
   encoding: legacy_encoding
   metrics: