Help needed: Using telegraf to parse mqtt with JSON payload

Telegraf
, ,
1

Hi,
I’m new to telegraf. That being said,
I’m trying to parse some mqtt topics with telegraf, with the (later) goal to import them into influxDB. The mqtt topics are like this:

mqtt_consumer,host=xxxx,topic=mtr/hzg/temp value="24.8" 1680447906430751221
mqtt_consumer,host=xxxx,topic=mtr/pwr/evu/SENSOR value="{\"Time\":\"2023-04-02T17:05:07\",\"EHZ541\":{\"cnt_imp\":3648.8,\"cnt_exp\":213492.8,\"mtr_cur\":-554}}" 1680447908136350043

They are either simple topic / value pairs (like the first one) or contain some JSON as payload.
The following telegraf.conf created above snippet.

[global_tags]

[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = "0s"
  hostname = ""
  omit_hostname = false

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

[[inputs.mqtt_consumer]]
  servers = ["tcp://127.0.0.1:1883"]
  topics = [
    "mtr/hzg/#",
    "mtr/pwr/evu/SENSOR",
  ]
  client_id = "telegraf"
  username = "UUUUUUU"
  password = "PPPPPPPP"
  data_format = "value"
  data_type = "string"

By replacing data_type with float, all good for the topic “mtr/hzg/#”.

Next, trying to parse JSON with the addition of the following config line (just appended to above config file)

[[inputs.mqtt_consumer]]
  servers = ["tcp://127.0.0.1:1883"]
  topics = [
    "mtr/pwr/evu/SENSOR",
  ]
  client_id = "telegraf"
  username = "UUUUUUUU"
  password = "PPPPPPPPP"

  data_format = "json_v2"

  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.object]]
      tags = [ "cnt_imp" ]
    [[inputs.mqtt_consumer.json_v2.fields]]
      cnt_imp = "float"
      cnt_exp = "float"
      mtr_cur = "float"

It produces the following result

2023-04-02T15:17:30Z I! Loading config file: /etc/telegraf/telegraf.conf
2023-04-02T15:17:30Z E! error loading config file /etc/telegraf/telegraf.conf: plugin inputs.mqtt_consumer: line 28: configuration specified the fields ["fields"], but they weren't used

What am I missing??

Thanks
Joe

2

Well you specify inputs.mqtt_consumer.json_v2.fields and this does not exist (that’s what the error message says). Did you mean inputs.mqtt_consumer.json_v2.field (without s)?

3

Thank you for your reply. Maybe I meant that.
However, I think it not to be good practice to have denominators like “field” and “fields”, when designing a language. The chance to err, either as newbie or as an old hand, is simply to big.

Regardless, here is the working solution I came up with. It may help others.

[global_tags]

[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = "0s"
  debug = true
  hostname = ""
  omit_hostname = false


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

#####################
# JSON mqtt messages
# i.e.
# mtr/pwr/evu/SENSOR {"Time":"2023-04-04T19:06:38","EHZ541":{"cnt_imp":3993.0,"cnt_exp":308939.6,"mtr_cur":-1393}}
#####################

[[inputs.mqtt_consumer]]
  alias = "mqtt_consumer_jsonv2"
  client_id = "telegraf_jsonv2"
  servers = ["tcp://127.0.0.1:1883"]
  username = "UUUUUUUU"
  password = "PPPPPPPP"
  name_override = "logdata"     # measurement
  topic_tag = "addr"            # tag, instead of "topic"
  topics = [
    "mtr/pwr/evu/SENSOR",
  ]
  data_format = "json_v2"

  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "@this"
      tags = [ "addr" ]


#####################
# simple mqtt message
# i.e.
# mtr/hzg/temp_VL 25.8
#####################

[[inputs.mqtt_consumer]]
  alias = "mqtt_consumer_value"
  client_id = "telegraf_value"
  servers = ["tcp://127.0.0.1:1883"]
  username = "UUUUUUUU"
  password = "PPPPPPPP"
  name_override = "logdata"     # measurement
  topic_tag = "addr"            # tag, instead of "topic"
  topics = [
    "mtr/hzg/#",
    "knx/#",
  ]
  data_format = "value"
  data_type = "string"

[[processors.enum]]             # map true/false etc to 1/0
  namepass = [ "logdata" ]
  [[processors.enum.mapping]]
    field = "value"
    [processors.enum.mapping.value_mappings]
      on = 1
      On = 1
      true = 1
      True = 1
      off = 0
      Off= 0
      false = 0
      False = 0
4

@universal-dilettant well that is not a language but a configuration file! What do you want a field to be called that does not have singular vs. plural issues? :wink:

5

@srebhan TOML Tom’s Obvious Minimal Language :shushing_face:

6

@universal-dilettant doesn’t answer my query for suggestions. :wink: