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

Kafka output fixup

This commit is contained in:
Cameron Sparr 2016-12-06 15:38:59 +00:00
parent d71a42cd1b
commit a093ec1eaa
1 changed files with 11 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -154,26 +154,23 @@ func (k *Kafka) Write(metrics []telegraf.Metric) error {
} }
for _, metric := range metrics { for _, metric := range metrics {
values, err := k.serializer.Serialize(metric) buf, err := k.serializer.Serialize(metric)
if err != nil { if err != nil {
return err return err
} }
var pubErr error m := &sarama.ProducerMessage{
for _, value := range values { Topic: k.Topic,
m := &sarama.ProducerMessage{ Value: sarama.ByteEncoder(buf),
Topic: k.Topic, }
Value: sarama.StringEncoder(value), if h, ok := metric.Tags()[k.RoutingTag]; ok {
} m.Key = sarama.StringEncoder(h)
if h, ok := metric.Tags()[k.RoutingTag]; ok {
m.Key = sarama.StringEncoder(h)
}
_, _, pubErr = k.producer.SendMessage(m)
} }
if pubErr != nil { _, _, err = k.producer.SendMessage(m)
return fmt.Errorf("FAILED to send kafka message: %s\n", pubErr)
if err != nil {
return fmt.Errorf("FAILED to send kafka message: %s\n", err)
} }
} }
return nil return nil