Convert MQTT messages with telegraf

Telegraf
1

Hi,
this is my first try with Telegraf and I’m struggling with MQTT data for storing on influxdb.

Subscription to the topic works but the only way i found to store the data to influxdb is:
data_format=value
data_type=string

Then the value in the measurement is the full json MQTT message like this
time host topic value
---- ---- ----- -----
1527437423530234352 Ubuntu-1604-xenial-64-minimal mw/openair/807f6940-d0f6-11e7-b63b-001999f98771 {“temp”: “32.76”,“hum”: “25.44”,“r2”: “1715”}
1527437423591364969 Ubuntu-1604-xenial-64-minimal mw/openair/807f6940-d0f6-11e7-b63b-001999f98771 {“pm10”: “0.00”,“pm25”: “0.00”,“rssi”: “-60”,“r1”: “52” }

The original MQTT message looks like
{
“pm10” : “0.00”,
“pm25” : “0.00”,
“rssi” : “-57”,
“r1” : “184”
}
which is a valid json format - BUT every single data is delivered as string data.
I have no access to the original data, so reformatting this is not an option.
I tried other data formats without any success.

Any idea?
for Reinhard - thanks

2

Unfortunately, the data is being delivered as Strings and not numeric values, so there’s not a whole lot that telegraf, itself, can do with it. There is a GitHub issue filed for this exact thing, but as far as I’m aware, no commitment has been made on implementing it. See Ability to retrieve string value via httpjson · Issue #2835 · influxdata/telegraf · GitHub if you’r like to add your vote.

The only idea I have, since you have no access to change the originating data types, is to write a small program to pull the MQTT data from the broker and transform it yourself.

3

In the upcoming 1.7 release you will be able to convert the strings to integers using the converter processor. If you want to experiment with this you can try the nightly builds.

1 Like
4

Thank you, I almost thought so too.
I did the conversion with node-red and I’ll try to create it in nodejs now.
For those who want to try it with node-red - this is the flow

[{“id”:“d0c349f5.0d4f6”,“type”:“mqtt in”,“z”:“75dda658.9a6b3”,“name”:“all_sensors”,“topic”:“mw/openair/#”,“qos”:“2”,“broker”:“9d5b42af.edb5c8”,“x”:310,“y”:1260,“wires”:[[“65247e2b.d61db”,“5b8e1f26.c9bc7”]]},{“id”:“65247e2b.d61db”,“type”:“debug”,“z”:“75dda658.9a6b3”,“name”:“”,“active”:false,“console”:“false”,“complete”:“false”,“x”:522.8833312988281,“y”:1251.9499816894531,“wires”:},{“id”:“4ef6251c.76aff4”,“type”:“debug”,“z”:“75dda658.9a6b3”,“name”:“”,“active”:false,“console”:“false”,“complete”:“true”,“x”:650,“y”:1380,“wires”:},{“id”:“5b8e1f26.c9bc7”,“type”:“json”,“z”:“75dda658.9a6b3”,“name”:“”,“pretty”:false,“x”:490,“y”:1320,“wires”:[[“4ef6251c.76aff4”,“657f1560.452664”]]},{“id”:“657f1560.452664”,“type”:“function”,“z”:“75dda658.9a6b3”,“name”:“”,“func”:“var out = ;\nvar out1 = msg.payload;\nvar feed = msg.topic;\nvar pos1 = feed.indexOf("/",3);\nfeed = feed.substr(pos1+1,feed.length-pos1);\nfor (var prop in out1){\n out1[prop]=parseFloat(out1[prop]);\n}\nout = [out1,\n {feed:feed}];\nmsg.payload=out;\nreturn msg;”,“outputs”:1,“noerr”:0,“x”:710,“y”:1300,“wires”:[[“b88d83bc.4f414”,“b609dc6b.d8a0d8”]]},{“id”:“b88d83bc.4f414”,“type”:“debug”,“z”:“75dda658.9a6b3”,“name”:“”,“active”:true,“console”:“false”,“complete”:“payload”,“x”:890,“y”:1300,“wires”:},{“id”:“b609dc6b.d8a0d8”,“type”:“influxdb out”,“z”:“75dda658.9a6b3”,“influxdb”:“53bc8665.62923”,“name”:“”,“measurement”:“all_openair”,“precision”:“”,“retentionPolicy”:“”,“x”:960,“y”:1360,“wires”:},{“id”:“9d5b42af.edb5c8”,“type”:“mqtt-broker”,“z”:“”,“broker”:“babeauf.nt.fh-koeln.de”,“port”:“2883”,“clientid”:“”,“usetls”:false,“compatmode”:true,“keepalive”:“60”,“cleansession”:true,“willTopic”:“”,“willQos”:“0”,“willPayload”:“”,“birthTopic”:“”,“birthQos”:“0”,“birthPayload”:“”},{“id”:“53bc8665.62923”,“type”:“influxdb”,“z”:“”,“hostname”:“127.0.0.1”,“port”:“8086”,“protocol”:“http”,“database”:“lanuv”,“name”:“”,“usetls”:false,“tls”:“”}]

Expects a json object like I described above. The mqtt feed cerates the tag. All values are have been converted from string to float.

1 Like