dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add support for lz4 compression to kafka output (#4492)

This commit is contained in:
Daniel Nelson 2018-07-31 15:09:30 -07:00 committed by Greg
parent 943dcc0c49
commit f4032fc78d
2 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -10,9 +10,15 @@ This plugin writes to a [Kafka Broker](http://kafka.apache.org/07/quickstart.htm
## Kafka topic for producer messages
topic = "telegraf"
## Optional client id
## Optional Client id
# client_id = "Telegraf"
## Set the minimal supported Kafka version. Setting this enables the use of new
## Kafka features and APIs. Of particular interested, lz4 compression
## requires at least version 0.10.0.0.
## ex: version = "1.1.0"
# version = ""
## Optional topic suffix configuration.
## If the section is omitted, no suffix is used.
## Following topic suffix methods are supported:
@ -20,7 +26,7 @@ This plugin writes to a [Kafka Broker](http://kafka.apache.org/07/quickstart.htm
## tags - suffix equals to separator + specified tags' values
## interleaved with separator
## Suffix equals to "_" + measurement's name
## Suffix equals to "_" + measurement name
# [outputs.kafka.topic_suffix]
# method = "measurement"
# separator = "_"

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

@ -38,6 +38,8 @@ type (
// MaxRetry Tag
MaxRetry int
Version string `toml:"version"`
// Legacy TLS config options
// TLS client certificate
Certificate string
@ -74,6 +76,12 @@ var sampleConfig = `
## Optional Client id
# client_id = "Telegraf"
## Set the minimal supported Kafka version. Setting this enables the use of new
## Kafka features and APIs. Of particular interested, lz4 compression
## requires at least version 0.10.0.0.
## ex: version = "1.1.0"
# version = ""
## Optional topic suffix configuration.
## If the section is omitted, no suffix is used.
## Following topic suffix methods are supported:
@ -191,6 +199,14 @@ func (k *Kafka) Connect() error {
}
config := sarama.NewConfig()
if k.Version != "" {
version, err := sarama.ParseKafkaVersion(k.Version)
if err != nil {
return err
}
config.Version = version
}
if k.ClientID != "" {
config.ClientID = k.ClientID
} else {