Parsing mqtt messages in telegraf

1

Hello,

I have a device that is publishing a pressure every second using MQTT in the format:
topic: device/sim_knauer_p2.1/1/data/pressure msg:56 type:<class 'str'>

The client is running running python phao.mqtt.client and the broker is the mosquitto-eclipse-mqtt. The msg is a string with a single value which is a float.

my telegraf.conf looks like this:

[[inputs.mqtt_consumer]]

  servers = ["tcp://172.19.0.5:1883"]

  topics = ["device/sim_knauer_p2.1/1/data/pressure",]

I seem to get the following error:
2022-07-05T20:50:43Z E! [inputs.mqtt_consumer] Error in plugin: metric parse error: expected tag at 1:3: "78"

I’ve read through the documentation, MQTT Topic and Payload Parsing with Telegraf | InfluxData but I’m a bit confused about how to parse this and what topic, measurements and tag are doing?

I’ve tried adding:

[[inputs.mqtt_consumer.topic_parsing]]
   topic = "+/+/+/+/+"

But this causes telegraf to crash on restart?

The documentation talks about LP but i don’t know what this means?

Are the topic, measurement; and tag tags reformatting the raw MQTT message so telegraf can then parse it?

Any help would be fantastic!

2

LP = Line Protocol

Have you looked at the GitHub MQTT Plug-in Docs? There are a few examples there.

3

Thanks Phil, i’ve tried to replicate using the GitHub example telegraf/plugins/inputs/mqtt_consumer at master · influxdata/telegraf · GitHub but telegraf throws an error…

telegraf.conf:

[[inputs.mqtt_consumer]]

  servers = ["tcp://172.19.0.5:1883"]

  topics = "/device/#"
  
  #topics = [
  #  "/device/#",
  #  "device/sim_knauer_p2.1/1/data/pressure",
  #]

  #data_format = "value"
  #data_type = "float"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "device/sim_knauer_p2.1/1/data/pressure"
    measurement = "measurement/_/_/_/_"
    tags = "_/equipment/device_number/type/field"

  [[processors.pivot]]
    tag_key = "field"
    value_key = "value"

In theory, this should parse to give the following using line protocol:

device, equipment=sim_knauer_p2.1, device_number=1, type=data, field=pressure, value=39

Error:

06T16:22:30Z E! [telegraf] Error running agent: Error loading config file /etc/telegraf/telegraf.conf: error parsing mqtt_consumer, line 17061: (mqtt_consumer.MQTTConsumer.Topics) cannot unmarshal TOML string into []string

Am I missing something really obvious?

4

Looks like an indentation problem. The [[processors.pivot]] declaration signifies a new plug-in (a processor in this case). So that section should be outdented to the same level as [[inputs.mqtt_consumer]].

5

Thanks for the suggestion! I’ve tried a couple of different indentations styles but still seem to hit the same problem - is there anything wrong with the syntax I’m using? I’m still getting my head around line protocol.

I did managed to configure telegraf to listen to an OPC UA server instead, but this is proving to be slower than I hoped. I believe telegraf polls rather than listen for subscriptions?

6

Square brackets for topics in the [[inputs.mqtt_consumer]] section but none in the [[inputs.mqtt_consumer.topic_parsing]] section.

In your first post looks like there’s an extra comma before the square bracket
topics = ["device/sim_knauer_p2.1/1/data/pressure",]

In the last post, missing square brackets topics = "/device/#"

It’s tricky.

7

Thanks Phill! I’ve deleted the last part and bracketed the topic - these seem to have done the trick :slight_smile:

telegraf.conf

[[inputs.mqtt_consumer]]

  servers = ["tcp://mosquitto:1883"]
  topics = [
    "device/#"
  ]
  data_format = "value"
  data_type = "float"

Thanks again!