Telegraf: Kafka consumer input plugin not storing tags in InfluxDB

Telegraf
1

Hi,
I’m trying to use Telegraf 1.22 to read from a Kafka topic and write the data to my InfluxDB.

The data in the Kafka topic looks like this:

Kafka Data 
{
  "fields": {
    "carrier-transitions": 3,
    "in-broadcast-pkts": 2,
    "in-discards": 0,
    "in-errors": 0,
    "in-fcs-errors": 0,
    "in-multicast-pkts": 3529482,
    "in-octets": 134256000853399,
    "in-pkts": 349627644952,
    "in-unicast-pkts": 349624115468,
    "in-unknown-protos": 0,
    "last-clear": "2021-11-03T05:36:34Z",
    "out-broadcast-pkts": 0,
    "out-discards": 0,
    "out-errors": 0,
    "out-multicast-pkts": 46243286,
    "out-octets": 2539352261443639,
    "out-pkts": 1891286684419,
    "out-unicast-pkts": 1891240441133
  },
  "name": "interfaces",
  "tags": {
    "device": "router2.mgt.net",
    "host": "central-ncs-deployment-64fb6db88c-8wdkc-telegraf-agent",
    "interface-name": "TenGigE0/0/0/19/1",
    "path": "openconfig:/interfaces/interface/state"
  },
  "timestamp": 1650645988
}

The data inside the Kafka topic has both “fields” and “tags”. However, when the data gets stored inside of InfluxDB I see the following fields and tags:

Data In InfluxDB 
> show field keys from kafka_consumer
name: kafka_consumer
fieldKey                   fieldType
--------                   ---------
fields_carrier-transitions float
fields_ifindex             float
fields_in-broadcast-pkts   float
fields_in-discards         float
fields_in-errors           float
fields_in-fcs-errors       float
fields_in-multicast-pkts   float
fields_in-octets           float
fields_in-pkts             float
fields_in-unicast-pkts     float
fields_in-unknown-protos   float
fields_last-change         float
fields_mtu                 float
fields_out-broadcast-pkts  float
fields_out-discards        float
fields_out-errors          float
fields_out-multicast-pkts  float
fields_out-octets          float
fields_out-pkts            float
fields_out-unicast-pkts    float
timestamp                  float
> show tag keys from kafka_consumer
name: kafka_consumer
tagKey
------
host
> show tag keys from kafka_consumer
name: kafka_consumer
tagKey
------
host

The tags “interface-name” and “device” are not being stored inside InfluxDB.

My Telegraf Conf looks like this:

Telegraf Conf 
[global_tags]

[agent]
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "5s"
  flush_jitter = "30s"
  debug = false
  quiet = false
  hostname = "netools2"
  omit_hostname = false

[[inputs.kafka_consumer]]
  brokers = ["server01.mgt.com:9093", 
            "server02.mgt.com:9093",
            "server03.mgt.com:9093",
            "server04.mgt.com:9093",
            "server05.mgt.com:9093"]

  topics = ["hub-clean-json-interfaces"]
  client_id = "netools2-ncs-telegraf-agent"
  consumer_group= "netools2-hub-clean-json-interfaces-netools-v1"
  offset = "newest"
  data_format = "json"
  tls_ca = "/etc/telegraf/ca.pem"
  tls_cert = "/etc/telegraf/bdcde.cer"
  tls_key = "/etc/telegraf/bdcde-priv.key"

[[outputs.influxdb]]
  urls = ["http://10.63.60.32:8086"]
  database = "prod-ncs"
  write_consistency = "any"
  timeout = "5s"

 [[outputs.file]]
  files = ["stdout"]

Not sure why this is happening. I would like to get “interface-name” and “device” stored as tags inside of my InfluxDB. Any guidance would be appreciated. Thank you.

2

Actually, I think I figured it out. Had to add the following to my Telegraf config file under the [[inputs.kafka.consumer]] stanza:

tag_keys = ["tags_device","tags_interface-name"]
1 Like