Hi i try to setup telegraf to write mqtt messages to influxdb.
I got telegraf up and running but got error messages when trying to consume mqtt messages because the payload is json.
I found a blog post on how to parse json from mqtt-payloads: MQTT Topic and Payload Parsing with Telegraf | InfluxData
I tried it and of course it doesn’t work and sadly there are only config snippets instead of a complete config.
I get the following error:
plugin inputs.mqtt_consumer: line 2: configuration specified the fields ["json_v2"], but they weren't used
Why do i get this error? Why does telegraf think i try to specify the field “json_v2”
Here is my config:
[[inputs.mqtt_consumer]]
[[inputs.mqtt_consumer.json_v2]]
data_format = "json_v2"
[[inputs.mqtt_consumer.json_v2]]
[[inputs.mqtt_consumer.json_v2.field]]
path = "humidity"
type = "float"
[[inputs.mqtt_consumer.json_v2.field]]
path = "temperature"
type = "float"
## Broker URLs for the MQTT server or cluster. To connect to multiple
## clusters or standalone servers, use a separate plugin instance.
## example: servers = ["tcp://localhost:1883"]
## servers = ["ssl://localhost:1883"]
## servers = ["ws://localhost:1883"]
servers = ["tcp://192.168.178.38:1883"]
## Topics that will be subscribed to.
topics = [
"zigbee2mqtt/#",
]
These are all TOML issues that you are runnign into, specifically about ordering and what belongs under what section. Keep in mind TOML does not care about whitespace or intendation, so whatever the last defined “header” is, that is what you are working under.
These two lines need to be reversed, you declare the data format first, then provide the format’s config:
Also your sever and topics need to be above the json_v2 config.
Can you share your full and complete mqtt_consumer plugin config please? it may be that you have defined tags or a topic parsing above the data format section.
thank you for your answer. I thought i must be some kind of formating error. Because i do not know the correct format i was just using try and error until i came to the point where i went out of ideas.
So this is my complete telegraf.conf when i first started debugging:
# Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
## Broker URLs for the MQTT server or cluster. To connect to multiple
## clusters or standalone servers, use a separate plugin instance.
## example: servers = ["tcp://localhost:1883"]
## servers = ["ssl://localhost:1883"]
## servers = ["ws://localhost:1883"]
servers = ["tcp://192.168.178.38:1883"]
## Topics that will be subscribed to.
topics = [
"zigbee2mqtt/#",
]
## The message topic will be stored in a tag specified by this value. If set
## to the empty string no topic tag will be created.
# topic_tag = "topic"
## QoS policy for messages
## 0 = at most once
## 1 = at least once
## 2 = exactly once
##
## When using a QoS of 1 or 2, you should enable persistent_session to allow
## resuming unacknowledged messages.
# qos = 0
## Connection timeout for initial connection in seconds
# connection_timeout = "30s"
## Maximum messages to read from the broker that have not been written by an
## output. For best throughput set based on the number of metrics within
## each message and the size of the output's metric_batch_size.
##
## For example, if each message from the queue contains 10 metrics and the
## output metric_batch_size is 1000, setting this to 100 will ensure that a
## full batch is collected and the write is triggered immediately without
## waiting until the next flush_interval.
# max_undelivered_messages = 1000
## Persistent session disables clearing of the client session on connection.
## In order for this option to work you must also set client_id to identify
## the client. To receive messages that arrived while the client is offline,
## also set the qos option to 1 or 2 and don't forget to also set the QoS when
## publishing.
# persistent_session = false
## If unset, a random client ID will be generated.
# client_id = ""
## Username and password to connect MQTT server.
# username = "telegraf"
# password = "metricsmetricsmetricsmetrics"
## Optional TLS Config
# tls_ca = "/etc/telegraf/ca.pem"
# tls_cert = "/etc/telegraf/cert.pem"
# tls_key = "/etc/telegraf/key.pem"
## Use TLS but skip chain & host verification
# insecure_skip_verify = false
## Data format to consume.
## Each data format has its own unique set of configuration options, read
## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
#data_format = "influx"
data_format = "json_v2"
[[inputs.mqtt_consumer.json_v2]]
[[inputs.mqtt_consumer.json_v2.field]]
path = "humidity"
type = "float"
[[inputs.mqtt_consumer.json_v2.field]]
path = "temperature"
type = "float"
## Enable extracting tag values from MQTT topics
## _ denotes an ignored entry in the topic path
# [[inputs.mqtt_consumer.topic_parsing]]
# topic = ""
# measurement = ""
# tags = ""
# fields = ""
## Value supported is int, float, unit
# [[inputs.mqtt_consumer.topic.types]]
# key = type
# [global_tags]
# user = "influx"
[[outputs.influxdb]]
urls = ["http://influxdb"]
skip_database_creation = true
password = "myverysecretpassword"
This gives me an error parsing mqtt_consumer, Invalid data format: json_v2 error.
This error lead me to the blog post i mentioned earlier which lead to the error i posted here.
