From 525850440000a7eefbce553284b518fc048f603b Mon Sep 17 00:00:00 2001 From: "Soundararajan, Senthilkumar" Date: Sat, 25 Jul 2026 12:57:07 +0530 Subject: [PATCH 1/3] feat(output): add custom headers support to kafka --- pkg/outputs/kafka_output/kafka_output.go | 91 ++++++++++++++++-------- 1 file changed, 61 insertions(+), 30 deletions(-) diff --git a/pkg/outputs/kafka_output/kafka_output.go b/pkg/outputs/kafka_output/kafka_output.go index 84fc7f56..117a55d0 100644 --- a/pkg/outputs/kafka_output/kafka_output.go +++ b/pkg/outputs/kafka_output/kafka_output.go @@ -100,32 +100,33 @@ type dynConfig struct { // 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 { @@ -612,10 +613,25 @@ CRPROD: } } + var headers []sarama.RecordHeader + for k, v := range cfg.AddHeaders { + headers = append(headers, sarama.RecordHeader{ + Key: []byte(k), + Value: []byte(v), + }) + } + + headers = append(headers, sarama.RecordHeader{ + Key: []byte("sub"), + Value: []byte(m.GetMeta()["subscription-name"]), + }) + topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, + Timestamp: time.Now(), } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) @@ -688,10 +704,25 @@ CRPROD: } } + var headers []sarama.RecordHeader + for k, v := range cfg.AddHeaders { + headers = append(headers, sarama.RecordHeader{ + Key: []byte(k), + Value: []byte(v), + }) + } + + headers = append(headers, sarama.RecordHeader{ + Key: []byte("sub"), + Value: []byte(m.GetMeta()["subscription-name"]), + }) + topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, + Timestamp: time.Now(), } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) From 5cbef6c9b0c3b2a9959f674d6de4743b04b91b52 Mon Sep 17 00:00:00 2001 From: "Soundararajan, Senthilkumar" Date: Sat, 15 Aug 2026 10:19:43 +0530 Subject: [PATCH 2/3] updated rc --- docs/user_guide/outputs/kafka_output.md | 63 ++++ pkg/outputs/kafka_output/kafka_output.go | 131 ++++++--- pkg/outputs/kafka_output/kafka_output_test.go | 278 ++++++++++++++++++ 3 files changed, 435 insertions(+), 37 deletions(-) diff --git a/docs/user_guide/outputs/kafka_output.md b/docs/user_guide/outputs/kafka_output.md index f30233ea..9920814c 100644 --- a/docs/user_guide/outputs/kafka_output.md +++ b/docs/user_guide/outputs/kafka_output.md @@ -103,6 +103,15 @@ outputs: enable-metrics: false # list of processors to apply on the message before writing event-processors: + # optional Kafka record headers. + # Static values are added as-is. + # Values containing Go templates are compiled once + # and evaluated for every message. + # The template context exposes message metadata 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. @@ -259,3 +268,57 @@ 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 can be either static strings or Go templates. Static values are computed once when the output is initialized or updated. Template values are compiled once and evaluated for each message. + +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 +``` + +Static and dynamic headers can be combined: + +```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. diff --git a/pkg/outputs/kafka_output/kafka_output.go b/pkg/outputs/kafka_output/kafka_output.go index 117a55d0..469f912a 100644 --- a/pkg/outputs/kafka_output/kafka_output.go +++ b/pkg/outputs/kafka_output/kafka_output.go @@ -9,6 +9,7 @@ package kafka_output import ( + "bytes" "context" "errors" "fmt" @@ -92,10 +93,12 @@ 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 + headers []sarama.RecordHeader + headerTpls map[string]*template.Template } // config // @@ -160,6 +163,70 @@ func (k *kafkaOutput) buildEventProcessors(logger *slog.Logger, eventProcessors return evps, nil } +func buildKafkaHeaders(headers map[string]string) ([]sarama.RecordHeader, map[string]*template.Template, error) { + if len(headers) == 0 { + return nil, nil, nil + } + + staticHeaders := make([]sarama.RecordHeader, 0, len(headers)) + dynamicHeaders := make(map[string]*template.Template) + + for k, v := range headers { + if strings.Contains(v, "{{") { + tpl, err := gtemplate.CreateTemplate(fmt.Sprintf("header-%s", k), v) + if err != nil { + return nil, nil, err + } + dynamicHeaders[k] = tpl.Funcs(outputs.TemplateFuncs) + continue + } + + staticHeaders = append(staticHeaders, sarama.RecordHeader{ + Key: []byte(k), + Value: []byte(v), + }) + } + + if len(dynamicHeaders) == 0 { + dynamicHeaders = nil + } + + return staticHeaders, dynamicHeaders, nil +} + +func (k *kafkaOutput) getHeaders(dc *dynConfig, meta outputs.Meta) []sarama.RecordHeader { + headers := slices.Clone(dc.headers) + + if len(dc.headerTpls) == 0 { + return headers + } + + 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 @@ -228,6 +295,11 @@ func (k *kafkaOutput) Init(ctx context.Context, name string, cfg map[string]inte OverrideTS: newCfg.OverrideTimestamps, } + dc.headers, dc.headerTpls, err = buildKafkaHeaders(newCfg.AddHeaders) + if err != nil { + return err + } + k.dynCfg.Store(dc) config, err := k.createConfigFor(newCfg) if err != nil { @@ -262,6 +334,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 } @@ -315,6 +392,11 @@ func (k *kafkaOutput) Update(ctx context.Context, cfg map[string]any) error { }, } + dc.headers, 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) @@ -613,25 +695,12 @@ CRPROD: } } - var headers []sarama.RecordHeader - for k, v := range cfg.AddHeaders { - headers = append(headers, sarama.RecordHeader{ - Key: []byte(k), - Value: []byte(v), - }) - } - - headers = append(headers, sarama.RecordHeader{ - Key: []byte("sub"), - Value: []byte(m.GetMeta()["subscription-name"]), - }) - + headers := k.getHeaders(dc, m.GetMeta()) topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), - Headers: headers, - Timestamp: time.Now(), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) @@ -704,25 +773,13 @@ CRPROD: } } - var headers []sarama.RecordHeader - for k, v := range cfg.AddHeaders { - headers = append(headers, sarama.RecordHeader{ - Key: []byte(k), - Value: []byte(v), - }) - } - - headers = append(headers, sarama.RecordHeader{ - Key: []byte("sub"), - Value: []byte(m.GetMeta()["subscription-name"]), - }) + headers := k.getHeaders(dc, m.GetMeta()) topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), - Headers: headers, - Timestamp: time.Now(), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) diff --git a/pkg/outputs/kafka_output/kafka_output_test.go b/pkg/outputs/kafka_output/kafka_output_test.go index 361b6955..72a43a1d 100644 --- a/pkg/outputs/kafka_output/kafka_output_test.go +++ b/pkg/outputs/kafka_output/kafka_output_test.go @@ -13,8 +13,10 @@ import ( "errors" "strings" "testing" + "text/template" "time" + "github.com/openconfig/gnmic/pkg/logging" "github.com/openconfig/gnmic/pkg/outputs" "github.com/zestor-dev/zestor/store" "github.com/zestor-dev/zestor/store/gomap" @@ -42,8 +44,48 @@ func TestKafkaOutput_Validate(t *testing.T) { }, {name: "bad target-template", cfg: map[string]any{"target-template": "{{"}, wantErr: true}, {name: "bad msg-template", cfg: map[string]any{"msg-template": "{{"}, wantErr: true}, + + { + name: "static headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "env": "prod", + }, + }, + wantErr: false, + }, + { + name: "templated headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + }, + wantErr: false, + }, + { + name: "mixed headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + }, + wantErr: false, + }, + { + name: "invalid header template", + cfg: map[string]any{ + "add-headers": map[string]any{ + "sub": "{{", + }, + }, + wantErr: true, + }, + {name: "valid event format", cfg: map[string]any{"format": "event"}, wantErr: false}, } + k := &kafkaOutput{} for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -126,3 +168,239 @@ func TestKafkaOutput_InitUpdateClose(t *testing.T) { t.Fatal("Close timed out") } } + +func TestBuildKafkaHeaders(t *testing.T) { + tests := []struct { + name string + headers map[string]string + wantStaticCount int + wantDynamicCount int + wantErr bool + }{ + { + name: "empty headers", + headers: nil, + wantStaticCount: 0, + wantDynamicCount: 0, + wantErr: false, + }, + { + name: "static headers only", + headers: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + wantStaticCount: 2, + wantDynamicCount: 0, + wantErr: false, + }, + { + name: "dynamic headers only", + headers: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + "source": `{{ index .Meta "source" }}`, + }, + wantStaticCount: 0, + wantDynamicCount: 2, + wantErr: false, + }, + { + name: "mixed static and dynamic headers", + headers: map[string]string{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + wantStaticCount: 1, + wantDynamicCount: 1, + wantErr: false, + }, + { + name: "invalid dynamic header template", + headers: map[string]string{ + "sub": "{{", + }, + wantStaticCount: 0, + wantDynamicCount: 0, + wantErr: true, + }, + { + name: "literal header containing no template", + headers: map[string]string{ + "literal": "subscription-name", + }, + wantStaticCount: 1, + wantDynamicCount: 0, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + staticHeaders, dynamicHeaders, err := buildKafkaHeaders(tt.headers) + + if tt.wantErr { + if err == nil { + t.Fatal("expected error") + } + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(staticHeaders) != tt.wantStaticCount { + t.Fatalf("expected %d static headers, got %d", tt.wantStaticCount, len(staticHeaders)) + } + + if len(dynamicHeaders) != tt.wantDynamicCount { + t.Fatalf("expected %d dynamic headers, got %d", tt.wantDynamicCount, len(dynamicHeaders)) + } + }) + } +} + +func TestGetHeaders(t *testing.T) { + tests := []struct { + name string + addHeaders map[string]string + meta outputs.Meta + wantHeaders map[string]string + runtimeBadTpl bool + }{ + { + name: "empty headers", + addHeaders: nil, + meta: outputs.Meta{ + "subscription-name": "interfaces", + "source": "router01", + }, + wantHeaders: map[string]string{}, + }, + { + name: "static headers only", + addHeaders: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + }, + { + name: "dynamic headers only", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + "source": `{{ index .Meta "source" }}`, + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + "source": "router01", + }, + wantHeaders: map[string]string{ + "sub": "interfaces", + "source": "router01", + }, + }, + { + name: "mixed static and dynamic headers", + addHeaders: map[string]string{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + "sub": "interfaces", + }, + }, + { + name: "missing metadata renders empty value", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{}, + wantHeaders: map[string]string{ + "sub": "", + }, + }, + { + name: "missing key renders empty value", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{"source": "router01"}, + wantHeaders: map[string]string{ + "sub": "", + }, + }, + { + name: "runtime template error skips dynamic header", + addHeaders: map[string]string{ + "env": "prod", + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + }, + runtimeBadTpl: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + k := &kafkaOutput{ + logger: logging.DiscardLogger(), + } + + staticHeaders, dynamicHeaders, err := buildKafkaHeaders(tt.addHeaders) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if tt.runtimeBadTpl { + if dynamicHeaders == nil { + dynamicHeaders = make(map[string]*template.Template) + } + + // This parses successfully but fails at execution because .Meta is not a function. + dynamicHeaders["bad"] = template.Must(template.New("bad").Parse(`{{ call .Meta }}`)) + } + + dc := &dynConfig{ + headers: staticHeaders, + headerTpls: dynamicHeaders, + } + + gotHeaders := k.getHeaders(dc, tt.meta) + + if len(gotHeaders) != len(tt.wantHeaders) { + t.Fatalf("expected %d headers, got %d: %#v", len(tt.wantHeaders), len(gotHeaders), gotHeaders) + } + + got := make(map[string]string, len(gotHeaders)) + for _, h := range gotHeaders { + got[string(h.Key)] = string(h.Value) + } + + for key, wantValue := range tt.wantHeaders { + gotValue, ok := got[key] + if !ok { + t.Fatalf("expected header %q to exist, got headers %#v", key, got) + } + + if gotValue != wantValue { + t.Fatalf("header %q: expected value %q, got %q", key, wantValue, gotValue) + } + } + }) + } +} From 0d6c9377a3344c37c31df777a6697216ee9bab50 Mon Sep 17 00:00:00 2001 From: "Soundararajan, Senthilkumar" Date: Sat, 15 Aug 2026 10:19:43 +0530 Subject: [PATCH 3/3] updated rc --- docs/user_guide/outputs/kafka_output.md | 64 +++++ pkg/outputs/kafka_output/kafka_output.go | 120 +++++--- pkg/outputs/kafka_output/kafka_output_test.go | 270 ++++++++++++++++++ 3 files changed, 417 insertions(+), 37 deletions(-) diff --git a/docs/user_guide/outputs/kafka_output.md b/docs/user_guide/outputs/kafka_output.md index f30233ea..fa8a44f3 100644 --- a/docs/user_guide/outputs/kafka_output.md +++ b/docs/user_guide/outputs/kafka_output.md @@ -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. @@ -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. diff --git a/pkg/outputs/kafka_output/kafka_output.go b/pkg/outputs/kafka_output/kafka_output.go index 117a55d0..38a4d4ac 100644 --- a/pkg/outputs/kafka_output/kafka_output.go +++ b/pkg/outputs/kafka_output/kafka_output.go @@ -9,6 +9,7 @@ package kafka_output import ( + "bytes" "context" "errors" "fmt" @@ -92,10 +93,11 @@ 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 // @@ -160,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 @@ -228,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 { @@ -262,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 } @@ -315,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) @@ -613,25 +684,12 @@ CRPROD: } } - var headers []sarama.RecordHeader - for k, v := range cfg.AddHeaders { - headers = append(headers, sarama.RecordHeader{ - Key: []byte(k), - Value: []byte(v), - }) - } - - headers = append(headers, sarama.RecordHeader{ - Key: []byte("sub"), - Value: []byte(m.GetMeta()["subscription-name"]), - }) - + headers := k.getHeaders(dc, m.GetMeta()) topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), - Headers: headers, - Timestamp: time.Now(), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) @@ -704,25 +762,13 @@ CRPROD: } } - var headers []sarama.RecordHeader - for k, v := range cfg.AddHeaders { - headers = append(headers, sarama.RecordHeader{ - Key: []byte(k), - Value: []byte(v), - }) - } - - headers = append(headers, sarama.RecordHeader{ - Key: []byte("sub"), - Value: []byte(m.GetMeta()["subscription-name"]), - }) + headers := k.getHeaders(dc, m.GetMeta()) topic := k.selectTopic(m.GetMeta()) msg := &sarama.ProducerMessage{ - Topic: topic, - Value: sarama.ByteEncoder(b), - Headers: headers, - Timestamp: time.Now(), + Topic: topic, + Value: sarama.ByteEncoder(b), + Headers: headers, } if cfg.InsertKey { msg.Key = sarama.ByteEncoder(k.partitionKey(m.GetMeta())) diff --git a/pkg/outputs/kafka_output/kafka_output_test.go b/pkg/outputs/kafka_output/kafka_output_test.go index 361b6955..0c204fcc 100644 --- a/pkg/outputs/kafka_output/kafka_output_test.go +++ b/pkg/outputs/kafka_output/kafka_output_test.go @@ -13,8 +13,10 @@ import ( "errors" "strings" "testing" + "text/template" "time" + "github.com/openconfig/gnmic/pkg/logging" "github.com/openconfig/gnmic/pkg/outputs" "github.com/zestor-dev/zestor/store" "github.com/zestor-dev/zestor/store/gomap" @@ -42,8 +44,48 @@ func TestKafkaOutput_Validate(t *testing.T) { }, {name: "bad target-template", cfg: map[string]any{"target-template": "{{"}, wantErr: true}, {name: "bad msg-template", cfg: map[string]any{"msg-template": "{{"}, wantErr: true}, + + { + name: "static headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "env": "prod", + }, + }, + wantErr: false, + }, + { + name: "templated headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + }, + wantErr: false, + }, + { + name: "mixed headers", + cfg: map[string]any{ + "add-headers": map[string]any{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + }, + wantErr: false, + }, + { + name: "invalid header template", + cfg: map[string]any{ + "add-headers": map[string]any{ + "sub": "{{", + }, + }, + wantErr: true, + }, + {name: "valid event format", cfg: map[string]any{"format": "event"}, wantErr: false}, } + k := &kafkaOutput{} for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -126,3 +168,231 @@ func TestKafkaOutput_InitUpdateClose(t *testing.T) { t.Fatal("Close timed out") } } + +func TestBuildKafkaHeaders(t *testing.T) { + tests := []struct { + name string + headers map[string]string + wantTplCount int + wantErr bool + }{ + { + name: "empty headers", + headers: nil, + wantTplCount: 0, + wantErr: false, + }, + { + name: "static headers only", + headers: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + wantTplCount: 2, + wantErr: false, + }, + { + name: "templated headers only", + headers: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + "source": `{{ index .Meta "source" }}`, + }, + wantTplCount: 2, + wantErr: false, + }, + { + name: "mixed static and templated headers", + headers: map[string]string{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + wantTplCount: 2, + wantErr: false, + }, + { + name: "invalid header template", + headers: map[string]string{ + "sub": "{{", + }, + wantTplCount: 0, + wantErr: true, + }, + { + name: "literal header value", + headers: map[string]string{ + "literal": "subscription-name", + }, + wantTplCount: 1, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + headerTpls, err := buildKafkaHeaders(tt.headers) + + if tt.wantErr { + if err == nil { + t.Fatal("expected error") + } + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(headerTpls) != tt.wantTplCount { + t.Fatalf("expected %d header templates, got %d", tt.wantTplCount, len(headerTpls)) + } + }) + } +} + +func TestGetHeaders(t *testing.T) { + tests := []struct { + name string + addHeaders map[string]string + meta outputs.Meta + wantHeaders map[string]string + runtimeBadTpl bool + }{ + { + name: "empty headers", + addHeaders: nil, + meta: outputs.Meta{ + "subscription-name": "interfaces", + "source": "router01", + }, + wantHeaders: map[string]string{}, + }, + { + name: "static headers only", + addHeaders: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + "region": "us-east-1", + }, + }, + { + name: "templated headers only", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + "source": `{{ index .Meta "source" }}`, + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + "source": "router01", + }, + wantHeaders: map[string]string{ + "sub": "interfaces", + "source": "router01", + }, + }, + { + name: "mixed static and templated headers", + addHeaders: map[string]string{ + "env": "prod", + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + "sub": "interfaces", + }, + }, + { + name: "missing metadata renders empty value", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{}, + wantHeaders: map[string]string{ + "sub": "", + }, + }, + { + name: "missing key renders empty value", + addHeaders: map[string]string{ + "sub": `{{ index .Meta "subscription-name" }}`, + }, + meta: outputs.Meta{ + "source": "router01", + }, + wantHeaders: map[string]string{ + "sub": "", + }, + }, + { + name: "runtime template error skips failed header", + addHeaders: map[string]string{ + "env": "prod", + }, + meta: outputs.Meta{ + "subscription-name": "interfaces", + }, + wantHeaders: map[string]string{ + "env": "prod", + }, + runtimeBadTpl: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + k := &kafkaOutput{ + logger: logging.DiscardLogger(), + } + + headerTpls, err := buildKafkaHeaders(tt.addHeaders) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if tt.runtimeBadTpl { + if headerTpls == nil { + headerTpls = make(map[string]*template.Template) + } + + // Parses successfully, but fails during execution because .Meta is not callable. + headerTpls["bad"] = template.Must( + template.New("bad").Parse(`{{ call .Meta }}`), + ) + } + + dc := &dynConfig{ + headerTpls: headerTpls, + } + + gotHeaders := k.getHeaders(dc, tt.meta) + + if len(gotHeaders) != len(tt.wantHeaders) { + t.Fatalf("expected %d headers, got %d: %#v", len(tt.wantHeaders), len(gotHeaders), gotHeaders) + } + + got := make(map[string]string, len(gotHeaders)) + for _, h := range gotHeaders { + got[string(h.Key)] = string(h.Value) + } + + for key, wantValue := range tt.wantHeaders { + gotValue, ok := got[key] + if !ok { + t.Fatalf("expected header %q to exist, got headers %#v", key, got) + } + + if gotValue != wantValue { + t.Fatalf("header %q: expected value %q, got %q", key, wantValue, gotValue) + } + } + }) + } +}