Skip to content

feat(output): add custom headers support to kafka #898

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
senthilsam wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat(output): add custom headers support to kafka #898

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5258504
feat(output): add custom headers support to kafka
senthilsam Jul 25, 2026
5cbef6c
updated rc
senthilsam Aug 15, 2026
0d6c937
updated rc
senthilsam Aug 15, 2026
b1ed9a9
Merge branch 'feat/add_kafka_headers' of github-gnmic:senthilsam/gnmi…
senthilsam Aug 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions docs/user_guide/outputs/kafka_output.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,14 @@ outputs:
enable-metrics: false
# list of processors to apply on the message before writing
event-processors:
# optional Kafka record headers.
# Header values are treated as Go templates and
# evaluated for every message.
# Message metadata is exposed under .Meta.
add-headers:
env: prod
sub: '{{ index .Meta "subscription-name" }}'
source: '{{ index .Meta "source" }}'
```

Currently all subscriptions updates (all targets and all subscriptions) are published to the defined topic name unless the `topic-prefix` configuration option is set.
Expand Down Expand Up @@ -259,3 +267,59 @@ When a Prometheus server is enabled, `gnmic` kafka output exposes 4 prometheus m
* `number_of_written_kafka_bytes_total`: Number of bytes written by gnmic kafka output. This Counter is labeled with the kafka producerID
* `number_of_kafka_msgs_sent_fail_total`: Number of failed msgs sent by gnmic kafka output. This Counter is labeled with the kafka producerID as well as the failure reason
* `msg_send_duration_ns`: gnmic kafka output send duration in nanoseconds. This Gauge is labeled with the kafka producerID

### Kafka headers

The `add-headers` field can be used to add Kafka record headers to produced messages.

Header values are treated as Go templates and evaluated for each message.

Plain strings are valid templates and render as-is, allowing static and dynamic header values to be configured using the same mechanism.

Header templates use the same Go template syntax and helper functions as `msg-template` and `target-template`.

The template context exposes message metadata under `.Meta`.

Example:

```yaml
outputs:
kafka-output:
type: kafka
address: localhost:9092
topic: telemetry
add-headers:
env: prod
sub: '{{ index .Meta "subscription-name" }}'
source: '{{ index .Meta "source" }}'
```

For a message with metadata:

```text
subscription-name=interfaces
source=router01
```

the resulting Kafka record headers will be:

```text
env=prod
sub=interfaces
source=router01
```

Header values may be either plain strings or template expressions:

```yaml
outputs:
kafka-output:
type: kafka
add-headers:
env: prod
region: us-east-1
sub: '{{ index .Meta "subscription-name" }}'
source: '{{ index .Meta "source" }}'
```

If a header template fails during execution, the failed header is skipped and the Kafka message is still sent.
145 changes: 111 additions & 34 deletions pkg/outputs/kafka_output/kafka_output.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
package kafka_output

import (
"bytes"
"context"
"errors"
"fmt"
Expand Down Expand Up @@ -92,40 +93,42 @@ type kafkaOutput struct {
}

type dynConfig struct {
targetTpl *template.Template
msgTpl *template.Template
evps []formatters.EventProcessor
mo *formatters.MarshalOptions
targetTpl *template.Template
msgTpl *template.Template
evps []formatters.EventProcessor
mo *formatters.MarshalOptions
headerTpls map[string]*template.Template
}

// config //
type config struct {
Address string `mapstructure:"address,omitempty"`
Topic string `mapstructure:"topic,omitempty"`
TopicPrefix string `mapstructure:"topic-prefix,omitempty"`
Name string `mapstructure:"name,omitempty"`
SASL *types.SASL `mapstructure:"sasl,omitempty"`
TLS *types.TLSConfig `mapstructure:"tls,omitempty"`
MaxRetry int `mapstructure:"max-retry,omitempty"`
Timeout time.Duration `mapstructure:"timeout,omitempty"`
RecoveryWaitTime time.Duration `mapstructure:"recovery-wait-time,omitempty"`
FlushFrequency time.Duration `mapstructure:"flush-frequency,omitempty"`
SyncProducer bool `mapstructure:"sync-producer,omitempty"`
RequiredAcks string `mapstructure:"required-acks,omitempty"`
Format string `mapstructure:"format,omitempty"`
InsertKey bool `mapstructure:"insert-key,omitempty"`
AddTarget string `mapstructure:"add-target,omitempty"`
TargetTemplate string `mapstructure:"target-template,omitempty"`
MsgTemplate string `mapstructure:"msg-template,omitempty"`
SplitEvents bool `mapstructure:"split-events,omitempty"`
NumWorkers int `mapstructure:"num-workers,omitempty"`
CompressionCodec string `mapstructure:"compression-codec,omitempty"`
KafkaVersion string `mapstructure:"kafka-version,omitempty"`
Debug bool `mapstructure:"debug,omitempty"`
BufferSize int `mapstructure:"buffer-size,omitempty"`
OverrideTimestamps bool `mapstructure:"override-timestamps,omitempty"`
EnableMetrics bool `mapstructure:"enable-metrics,omitempty"`
EventProcessors []string `mapstructure:"event-processors,omitempty"`
Address string `mapstructure:"address,omitempty"`
Topic string `mapstructure:"topic,omitempty"`
TopicPrefix string `mapstructure:"topic-prefix,omitempty"`
Name string `mapstructure:"name,omitempty"`
SASL *types.SASL `mapstructure:"sasl,omitempty"`
TLS *types.TLSConfig `mapstructure:"tls,omitempty"`
MaxRetry int `mapstructure:"max-retry,omitempty"`
Timeout time.Duration `mapstructure:"timeout,omitempty"`
RecoveryWaitTime time.Duration `mapstructure:"recovery-wait-time,omitempty"`
FlushFrequency time.Duration `mapstructure:"flush-frequency,omitempty"`
SyncProducer bool `mapstructure:"sync-producer,omitempty"`
RequiredAcks string `mapstructure:"required-acks,omitempty"`
Format string `mapstructure:"format,omitempty"`
InsertKey bool `mapstructure:"insert-key,omitempty"`
AddTarget string `mapstructure:"add-target,omitempty"`
TargetTemplate string `mapstructure:"target-template,omitempty"`
MsgTemplate string `mapstructure:"msg-template,omitempty"`
SplitEvents bool `mapstructure:"split-events,omitempty"`
NumWorkers int `mapstructure:"num-workers,omitempty"`
CompressionCodec string `mapstructure:"compression-codec,omitempty"`
KafkaVersion string `mapstructure:"kafka-version,omitempty"`
Debug bool `mapstructure:"debug,omitempty"`
BufferSize int `mapstructure:"buffer-size,omitempty"`
OverrideTimestamps bool `mapstructure:"override-timestamps,omitempty"`
EnableMetrics bool `mapstructure:"enable-metrics,omitempty"`
EventProcessors []string `mapstructure:"event-processors,omitempty"`
AddHeaders map[string]string `mapstructure:"add-headers,omitempty"`
}

func (c *config) LogValue() slog.Value {
Expand Down Expand Up @@ -159,6 +162,60 @@ func (k *kafkaOutput) buildEventProcessors(logger *slog.Logger, eventProcessors
return evps, nil
}

func buildKafkaHeaders(headers map[string]string) (map[string]*template.Template, error) {
if len(headers) == 0 {
return nil, nil
}

result := make(map[string]*template.Template, len(headers))

for k, v := range headers {
tpl, err := gtemplate.CreateTemplate(
fmt.Sprintf("header-%s", k),
v,
)
if err != nil {
return nil, err
}

result[k] = tpl.Funcs(outputs.TemplateFuncs)
}

return result, nil
}

func (k *kafkaOutput) getHeaders(dc *dynConfig, meta outputs.Meta) []sarama.RecordHeader {
if len(dc.headerTpls) == 0 {
return nil
}

headers := make([]sarama.RecordHeader, 0, len(dc.headerTpls))

data := map[string]any{
"Meta": meta,
}

for hk, tpl := range dc.headerTpls {
var buf bytes.Buffer

if err := tpl.Execute(&buf, data); err != nil {
k.logger.Warn(
"failed to execute kafka header template",
"header", hk,
"err", err,
)
continue
}

headers = append(headers, sarama.RecordHeader{
Key: []byte(hk),
Value: buf.Bytes(),
})
}

return headers
}

// Init /
func (k *kafkaOutput) Init(ctx context.Context, name string, cfg map[string]interface{}, opts ...outputs.Option) error {
k.init() // init struct fields
Expand Down Expand Up @@ -227,6 +284,11 @@ func (k *kafkaOutput) Init(ctx context.Context, name string, cfg map[string]inte
OverrideTS: newCfg.OverrideTimestamps,
}

dc.headerTpls, err = buildKafkaHeaders(newCfg.AddHeaders)
if err != nil {
return err
}

k.dynCfg.Store(dc)
config, err := k.createConfigFor(newCfg)
if err != nil {
Expand Down Expand Up @@ -261,6 +323,11 @@ func (k *kafkaOutput) Validate(cfg map[string]any) error {
if err != nil {
return err
}

_, err = buildKafkaHeaders(ncfg.AddHeaders)
if err != nil {
return err
}
return nil
}

Expand Down Expand Up @@ -314,6 +381,11 @@ func (k *kafkaOutput) Update(ctx context.Context, cfg map[string]any) error {
},
}

dc.headerTpls, err = buildKafkaHeaders(newCfg.AddHeaders)
if err != nil {
return err
}

prevDC := k.dynCfg.Load()
if rebuildProcessors {
dc.evps, err = k.buildEventProcessors(k.logger, newCfg.EventProcessors)
Expand Down Expand Up @@ -612,10 +684,12 @@ CRPROD:
}
}

headers := k.getHeaders(dc, m.GetMeta())
topic := k.selectTopic(m.GetMeta())
msg := &sarama.ProducerMessage{
Topic: topic,
Value: sarama.ByteEncoder(b),
Topic: topic,
Value: sarama.ByteEncoder(b),
Headers: headers,
}
if cfg.InsertKey {
msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta()))
Expand Down Expand Up @@ -688,10 +762,13 @@ CRPROD:
}
}

headers := k.getHeaders(dc, m.GetMeta())

topic := k.selectTopic(m.GetMeta())
msg := &sarama.ProducerMessage{
Topic: topic,
Value: sarama.ByteEncoder(b),
Topic: topic,
Value: sarama.ByteEncoder(b),
Headers: headers,
}
if cfg.InsertKey {
msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta()))
Expand Down
Loading