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 84fc7f56..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,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 { @@ -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 @@ -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 { @@ -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 } @@ -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) @@ -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())) @@ -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())) 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) + } + } + }) + } +}