Help needed: Using telegraf to parse mqtt with JSON payload

universal-dilettant · April 2, 2023, 3:25pm

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

srebhan · April 3, 2023, 12:31pm

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)?

universal-dilettant · April 4, 2023, 5:29pm

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

srebhan · May 16, 2023, 2:22pm

@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?

universal-dilettant · May 17, 2023, 3:41pm

@srebhan TOML Tom’s Obvious Minimal Language

srebhan · May 22, 2023, 1:08pm

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

Pierrot8 · February 19, 2024, 7:09pm

I was making the same mistake, and it’s because of the last example provided by the Influxdata documentation in this post:
MQTT Topic and Payload Parsing with Telegraf | InfluxData

What that article and the OP are missing is the object part in

[[inputs.mqtt_consumer.json_v2.object.fields]]

which has the function of setting the data type of the values read with [[inputs.mqtt_consumer.json_v2.object]]

I suspect in general there are a couple of typos in that article, but being a newbie myself I am never sure about that

universal-dilettant · February 20, 2024, 4:23am

I solved the issue for me long time ago. Getting more fluent with Telegraf was helping, of course, but when you start the complexity is high and the learning curve is steep.

Mistakes or ambiguity in blogs and other documentation is a showstopper, then. I’ve got no idea how to improve it in an agile development environment, with developers not really keen to document or even write a tutorial.

jpowers · February 20, 2024, 5:14pm

Hi,

If there is a typo in a blog post, bringing the typo to our attention is extremely appreciated. I have passed the missing object in the last code example and a plural fields to our marketing folks now and the blog post will get updated.

I am not sure I follow what an agile development environment has to do with anything here.

with developers not really keen to document or even write a tutorial.

We absolutely do update documentation and write tutorials. Furthermore, we are more than happy to assist users working through issues here on these forums and on the slack. If there is an instance where you have seen folks not want to document or where something is missing, then please share.

ChrisChros · February 24, 2024, 3:24pm

Hi,
I am at the beginning of the learning curve how to configure Telegraf for my needs. Till now I have successfully configured to receive MQTT datas from my wallbox for normal values. But the box also publish some datas combined in a JSON String.
Within the MQTT Explorer the String looks like this:

The Value of the JSON-String for the config topic is as follows:

{“average_consumption”: 18000, “charge_template”: 1, “chargemode”: “pv_charging”, “current_plan”: null, “ev_template”: 1, “priority”: false, “time_charging_in_use”: false}

How can I extract for example the chargemode value?

My telegraf.conf:

[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 = false
  logtarget = "file"
  logfile = "/var/log/telegraf/telegraf.log"
  hostname = ""
  omit_hostname = false

###############################################################################
#                            OUTPUT PLUGINS                                   #
###############################################################################

[[outputs.influxdb]]
	urls = ["http://127.0.0.1:8086"]
	database = "telegraf"

###############################################################################
#                            SERVICE INPUT PLUGINS                            #
###############################################################################

# # Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
servers = ["tcp://10.10.20.12:1883"]
#
#   ## Topics that will be subscribed to.
topics = [
     "openWB/chargepoint/get/#",
	 "openWB/chargepoint/1/get/#",
	 "openWB/vehicle/2/get/#",
   ]
client_id = "TelegrafOpenWB"
data_format = "value"
# data_type = "float"
data_type = "string"
#
#   ## Enable extracting tag values from MQTT topics
#   ## _ denotes an ignored entry in the topic path
	[[inputs.mqtt_consumer.topic_parsing]]
		topic = "openWB/chargepoint/1/get/+"
   		measurement = "chargemode"
#   #   tags = ""
#   #   fields = ""
#   ## Value supported is int, float, unit
#   #   [[inputs.mqtt_consumer.topic.types]]
#   #      key = type

Which is not working for me. I receive all the topics and values but I can not separate all the informations from the JSON-Sttring.

jpowers · February 26, 2024, 7:09pm

@ChrisChros can you open a new topic please? Thanks

Topic		Replies	Views
Telegraf parse Mqtt payload Telegraf influxdb , telegraf , json	1	25	December 11, 2024
Parsing MQTT Topic Telegraf	2	99	July 10, 2024
JSON parsing help Telegraf telegraf , json	8	1027	February 26, 2022
Parsing an array from an MQTT topic with json_v2 Telegraf telegraf , mqtt , json	2	2661	September 26, 2022
Basic JSON>MQTT>Telegraf issues Telegraf telegraf	2	7344	October 1, 2017

Help needed: Using telegraf to parse mqtt with JSON payload

Related topics