feat(outputs.kafka): Option to add metric name as record header

influxdata · Aug 12, 2024 · 1616e8e · 1616e8e
1 parent 9df4800
commit 1616e8e
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/plugins/outputs/kafka/README.md b/plugins/outputs/kafka/README.md
@@ -127,6 +127,9 @@ to use them.
   ##   * now: Uses the time of write
   # producer_timestamp = metric
 
+  ## Add metric name as specified kafka header if not empty
+  # metric_name_header = ""
+
   ## Optional TLS Config
   # enable_tls = false
   # tls_ca = "/etc/telegraf/ca.pem"

diff --git a/plugins/outputs/kafka/kafka.go b/plugins/outputs/kafka/kafka.go
@@ -39,6 +39,7 @@ type Kafka struct {
 	RoutingTag        string          `toml:"routing_tag"`
 	RoutingKey        string          `toml:"routing_key"`
 	ProducerTimestamp string          `toml:"producer_timestamp"`
+	MetricNameHeader  string          `toml:"metric_name_header"`
 	Log               telegraf.Logger `toml:"-"`
 	proxy.Socks5ProxyConfig
 	kafka.WriteConfig
@@ -209,6 +210,15 @@ func (k *Kafka) Write(metrics []telegraf.Metric) error {
 			Value: sarama.ByteEncoder(buf),
 		}
 
+		if k.MetricNameHeader != "" {
+			m.Headers = []sarama.RecordHeader{
+				{
+					Key:   []byte(k.MetricNameHeader),
+					Value: []byte(metric.Name()),
+				},
+			}
+		}
+
 		// Negative timestamps are not allowed by the Kafka protocol.
 		if k.ProducerTimestamp == "metric" && !metric.Time().Before(zeroTime) {
 			m.Timestamp = metric.Time()

diff --git a/plugins/outputs/kafka/sample.conf b/plugins/outputs/kafka/sample.conf
@@ -87,6 +87,9 @@
   ##   * now: Uses the time of write
   # producer_timestamp = metric
 
+  ## Add metric name as specified kafka header if not empty
+  # metric_name_header = ""
+
   ## Optional TLS Config
   # enable_tls = false
   # tls_ca = "/etc/telegraf/ca.pem"