Add ability to tag metrics with topic in kafka_consumer (#5038)

plugins/inputs/kafka_consumer/README.md

@@ -14,6 +14,8 @@ and use the old zookeeper connection method.
   brokers = ["localhost:9092"]
   ## topic(s) to consume
   topics = ["telegraf"]
+  ## Add topic as tag if topic_tag is not empty
+  # topic_tag = ""
 
   ## Optional Client id
   # client_id = "Telegraf"

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -40,6 +40,8 @@ type Kafka struct {
 	Offset                 string   `toml:"offset"`
 	SASLUsername           string   `toml:"sasl_username"`
 	SASLPassword           string   `toml:"sasl_password"`
+	TopicTag               string   `toml:"topic_tag"`
+
 	tls.ClientConfig
 
 	cluster Consumer
@@ -60,6 +62,8 @@ var sampleConfig = `
   brokers = ["localhost:9092"]
   ## topic(s) to consume
   topics = ["telegraf"]
+  ## Add topic as tag if topic_tag is not empty
+  # topic_tag = ""
 
   ## Optional Client id
   # client_id = "Telegraf"
@@ -256,7 +260,11 @@ func (k *Kafka) onMessage(acc telegraf.TrackingAccumulator, msg *sarama.Consumer
 	if err != nil {
 		return err
 	}
-
+	if len(k.TopicTag) > 0 {
+		for _, metric := range metrics {
+			metric.AddTag(k.TopicTag, msg.Topic)
+		}
+	}
 	id := acc.AddTrackingMetricGroup(metrics)
 	k.messages[id] = msg
 

plugins/inputs/kafka_consumer/kafka_consumer_test.go

@@ -61,6 +61,25 @@ func newTestKafka() (*Kafka, *TestConsumer) {
 	return &k, consumer
 }
 
+func newTestKafkaWithTopicTag() (*Kafka, *TestConsumer) {
+	consumer := &TestConsumer{
+		errors:   make(chan error),
+		messages: make(chan *sarama.ConsumerMessage, 1000),
+	}
+	k := Kafka{
+		cluster:                consumer,
+		ConsumerGroup:          "test",
+		Topics:                 []string{"telegraf"},
+		Brokers:                []string{"localhost:9092"},
+		Offset:                 "oldest",
+		MaxUndeliveredMessages: defaultMaxUndeliveredMessages,
+		doNotCommitMsgs:        true,
+		messages:               make(map[telegraf.TrackingID]*sarama.ConsumerMessage),
+		TopicTag:               "topic",
+	}
+	return &k, consumer
+}
+
 // Test that the parser parses kafka messages into points
 func TestRunParser(t *testing.T) {
 	k, consumer := newTestKafka()
@@ -75,6 +94,22 @@ func TestRunParser(t *testing.T) {
 	assert.Equal(t, acc.NFields(), 1)
 }
 
+// Test that the parser parses kafka messages into points
+// and adds the topic tag
+func TestRunParserWithTopic(t *testing.T) {
+	k, consumer := newTestKafkaWithTopicTag()
+	acc := testutil.Accumulator{}
+	ctx := context.Background()
+
+	k.parser, _ = parsers.NewInfluxParser()
+	go k.receiver(ctx, &acc)
+	consumer.Inject(saramaMsgWithTopic(testMsg, "test_topic"))
+	acc.Wait(1)
+
+	assert.Equal(t, acc.NFields(), 1)
+	assert.True(t, acc.HasTag("cpu_load_short", "topic"))
+}
+
 // Test that the parser ignores invalid messages
 func TestRunParserInvalidMsg(t *testing.T) {
 	k, consumer := newTestKafka()
@@ -173,3 +208,13 @@ func saramaMsg(val string) *sarama.ConsumerMessage {
 		Partition: 0,
 	}
 }
+
+func saramaMsgWithTopic(val string, topic string) *sarama.ConsumerMessage {
+	return &sarama.ConsumerMessage{
+		Key:       nil,
+		Value:     []byte(val),
+		Offset:    0,
+		Partition: 0,
+		Topic:     topic,
+	}
+}