CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

This commit is contained in:
Lauri 2024-08-13 18:42:23 +03:00 committed by GitHub
parent 99b55b71b0
commit fc198910cf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 16 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
plugins/outputs/kafka/README.md
View File

@ -127,6 +127,9 @@ to use them.
## * now: Uses the time of write ## * now: Uses the time of write
# producer_timestamp = metric # producer_timestamp = metric
## Add metric name as specified kafka header if not empty
# metric_name_header = ""
## Optional TLS Config ## Optional TLS Config
# enable_tls = false # enable_tls = false
# tls_ca = "/etc/telegraf/ca.pem" # tls_ca = "/etc/telegraf/ca.pem"

10
plugins/outputs/kafka/kafka.go
View File

@ -39,6 +39,7 @@ type Kafka struct {
RoutingTag string `toml:"routing_tag"` RoutingTag string `toml:"routing_tag"`
RoutingKey string `toml:"routing_key"` RoutingKey string `toml:"routing_key"`
ProducerTimestamp string `toml:"producer_timestamp"` ProducerTimestamp string `toml:"producer_timestamp"`
MetricNameHeader string `toml:"metric_name_header"`
Log telegraf.Logger `toml:"-"` Log telegraf.Logger `toml:"-"`
proxy.Socks5ProxyConfig proxy.Socks5ProxyConfig
kafka.WriteConfig kafka.WriteConfig
@ -209,6 +210,15 @@ func (k *Kafka) Write(metrics []telegraf.Metric) error {
Value: sarama.ByteEncoder(buf), 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. // Negative timestamps are not allowed by the Kafka protocol.
if k.ProducerTimestamp == "metric" && !metric.Time().Before(zeroTime) { if k.ProducerTimestamp == "metric" && !metric.Time().Before(zeroTime) {
m.Timestamp = metric.Time() m.Timestamp = metric.Time()

3
plugins/outputs/kafka/sample.conf
View File

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