Kafka extract header metada (#24367)

**Description:** Enable support to extract headers from Kafka Messages
and attach them to resource attributes. In later stages in the pipeline,
different exporters can utilize this information.

**Link to tracking Issue:**
[21729](#21729)

**Testing:** Added test cases for logs, traces, and metrics.

**Documentation:** <Describe the documentation added.>

---------

Co-authored-by: Alex Boten <aboten@lightstep.com>

.chloggen/header-extraction-kafka.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Allow users to attach kafka header metadata with the log/metric/trace record in the pipeline. Introduce a new config param, 'header_extraction' and some examples.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24367]
+
+# (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:

receiver/kafkareceiver/README.md

@@ -87,11 +87,51 @@ The following settings can be optionally configured:
   - `after`: (default = false) If true, the messages are marked after the pipeline execution
   - `on_error`: (default = false) If false, only the successfully processed messages are marked
     **Note: this can block the entire partition in case a message processing returns a permanent error**
-
+- `header_extraction`:
+  - `extract_headers` (default = false): Allows user to attach header fields to resource attributes in otel piepline
+  - `headers` (default = []): List of headers they'd like to extract from kafka record. 
+  **Note: Matching pattern will be `exact`. Regexes are not supported as of now.** 
 Example:
 
 ```yaml
 receivers:
   kafka:
     protocol_version: 2.0.0
 ```
+
+Example of header extraction:
+
+```yaml
+receivers:
+  kafka:
+    topic: test
+    header_extraction: 
+      extract_headers: true
+      headers: ["header1", "header2"]
+```
+
+- If we feed following kafka record to `test` topic and use above configs: 
+```yaml
+{
+  event: Hello,
+  headers: {
+    header1: value1,
+    header2: value2,
+  }
+}
+```
+we will get a log record in collector similar to: 
+```yaml
+{
+  ...
+  body: Hello,
+  resource: {
+    kafka.header.header1: value1,
+    kafka.header.header2: value2,
+  },
+  ...
+}
+```
+
+- Here you can see the kafka record header `header1` and `header2` being added to resource attribute.
+- Every **matching** kafka header key is prefixed with `kafka.header` string and attached to resource attributes.

receiver/kafkareceiver/config.go

@@ -31,6 +31,11 @@ type MessageMarking struct {
 	OnError bool `mapstructure:"on_error"`
 }
 
+type HeaderExtraction struct {
+	ExtractHeaders bool     `mapstructure:"extract_headers"`
+	Headers        []string `mapstructure:"headers"`
+}
+
 // Config defines configuration for Kafka receiver.
 type Config struct {
 	// The list of kafka brokers (default localhost:9092)
@@ -60,6 +65,9 @@ type Config struct {
 
 	// Controls the way the messages are marked as consumed
 	MessageMarking MessageMarking `mapstructure:"message_marking"`
+
+	// Extract headers from kafka records
+	HeaderExtraction HeaderExtraction `mapstructure:"header_extraction"`
 }
 
 const (

receiver/kafkareceiver/factory.go

@@ -111,6 +111,9 @@ func createDefaultConfig() component.Config {
 			After:   false,
 			OnError: false,
 		},
+		HeaderExtraction: HeaderExtraction{
+			ExtractHeaders: false,
+		},
 	}
 }
 

receiver/kafkareceiver/header_extraction.go

@@ -0,0 +1,94 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkareceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver"
+
+import (
+	"fmt"
+
+	"github.com/IBM/sarama"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+func getAttribute(key string) string {
+	return fmt.Sprintf("kafka.header.%s", key)
+}
+
+type HeaderExtractor interface {
+	extractHeadersTraces(ptrace.Traces, *sarama.ConsumerMessage)
+	extractHeadersMetrics(pmetric.Metrics, *sarama.ConsumerMessage)
+	extractHeadersLogs(plog.Logs, *sarama.ConsumerMessage)
+}
+
+type headerExtractor struct {
+	logger  *zap.Logger
+	headers []string
+}
+
+func (he *headerExtractor) extractHeadersTraces(traces ptrace.Traces, message *sarama.ConsumerMessage) {
+	for _, header := range he.headers {
+		value, ok := getHeaderValue(message.Headers, header)
+		if !ok {
+			he.logger.Debug("Header key not found in the trace: ", zap.String("key", header))
+			continue
+		}
+		for i := 0; i < traces.ResourceSpans().Len(); i++ {
+			rs := traces.ResourceSpans().At(i)
+			rs.Resource().Attributes().PutStr(getAttribute(header), value)
+		}
+	}
+}
+
+func (he *headerExtractor) extractHeadersLogs(logs plog.Logs, message *sarama.ConsumerMessage) {
+	for _, header := range he.headers {
+		value, ok := getHeaderValue(message.Headers, header)
+		if !ok {
+			he.logger.Debug("Header key not found in the log: ", zap.String("key", header))
+			continue
+		}
+		for i := 0; i < logs.ResourceLogs().Len(); i++ {
+			rl := logs.ResourceLogs().At(i)
+			rl.Resource().Attributes().PutStr(getAttribute(header), value)
+		}
+	}
+}
+
+func (he *headerExtractor) extractHeadersMetrics(metrics pmetric.Metrics, message *sarama.ConsumerMessage) {
+	for _, header := range he.headers {
+		value, ok := getHeaderValue(message.Headers, header)
+		if !ok {
+			he.logger.Debug("Header key not found in the metric: ", zap.String("key", header))
+			continue
+		}
+		for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+			rm := metrics.ResourceMetrics().At(i)
+			rm.Resource().Attributes().PutStr(getAttribute(header), value)
+		}
+	}
+}
+
+func getHeaderValue(headers []*sarama.RecordHeader, header string) (string, bool) {
+	for _, kafkaHeader := range headers {
+		headerKey := string(kafkaHeader.Key)
+		if headerKey == header {
+			// matching header found
+			return string(kafkaHeader.Value), true
+		}
+	}
+	// no header found matching the key, report to the user
+	return "", false
+}
+
+type nopHeaderExtractor struct{}
+
+func (he *nopHeaderExtractor) extractHeadersTraces(_ ptrace.Traces, _ *sarama.ConsumerMessage) {
+}
+
+func (he *nopHeaderExtractor) extractHeadersLogs(_ plog.Logs, _ *sarama.ConsumerMessage) {
+}
+
+func (he *nopHeaderExtractor) extractHeadersMetrics(_ pmetric.Metrics, _ *sarama.ConsumerMessage) {
+}