Forwarding data to InfluxDB via MQTT and Telegraf. One of the most common things I do when collecting time-series data from sensors deployed in the field — sometimes literally — is having them populate our time-series database of choice, InfluxDB.
Pushing sensor data to InfluxDB via the API and curl works for testing but isn't a fire-and-forget solution at scale. A cleaner pipeline uses MQTT (Mosquitto) for guaranteed, fast delivery, with Telegraf's mqtt_consumer input subscribing to a topic and influxdb_v2 outputs routing each measurement into the right bucket via namepass. Install, drop in a minimal telegraf.conf, and your data flows.
Usually I’m working on something like an AWS EC2 instance running Ubuntu, so that’s the configuration I’ll assume for the rest of this article.
In the past, particularly for testing, I’ve usually pushed data via the InfluxDB API using curl from the command line. This has advantages — you can send individual lines easily — but it requires batching multiple lines for performance. Here’s an example of posting a single datapoint from a bash script (assuming $HOST, $BUCKET, and $TOKEN are set up):
curl -XPOST "http://Nyarlathotep.local:9999/api/v2/write?org=$ORG&bucket=$BUCKET" \
--header "Authorization: Token $TOKEN" \
--data-raw "name,tag1=194,tag2=5755 field1=7900.8 1593523543000000000"Batching multiple lines is as simple as supplying them in --data-raw with a carriage return (\n) between each line of Line Protocol. The downside: I’m either running a script like this in real time, or calling it periodically — and I have to worry about bash line limits. It’s fine for testing, not convenient for large amounts of data, and not really fire-and-forget.
Since I use MQTT elsewhere in our infrastructure for moving data en masse — where guaranteed delivery, speed and peace of mind matter — I thought that would be an ideal method instead of fiddling with bash and curl.
To keep things simple, especially for testing and research, I installed an MQTT server on the same machine as InfluxDB. Installing locally is as simple as:
sudo apt-get install mosquittoIf you need local testing tools to post to and watch incoming messages on topics, install the clients too — they’re very handy:
sudo apt-get install mosquitto-clientsThat’s it. A simple MQTT installation requires no further configuration, and can be stopped and started in the usual way:
sudo systemctl stop mosquitto
sudo systemctl start mosquittoGetting the data into an InfluxDB bucket is also trivial if you’re willing to set up Telegraf. The range of plugins is huge, but the bare-bones setup needed to ingest MQTT into a bucket is quite simple. Install by downloading the latest version:
wget https://dl.influxdata.com/telegraf/releases/telegraf_1.18.3-1_amd64.deb
sudo dpkg -i telegraf_1.18.3-1_amd64.debAs with MQTT, it can be stopped and started with systemctl. But you’ll need to do some configuration.
Say we have a bunch of sensors: one set producing wind data, the other water data. We’d like to separate incoming data into two buckets, created in InfluxDB’s UI, called “Wind” and “Water”. We’ll use Line Protocol for the MQTT payload. To discriminate between sensor types, I’ll use the measurement parameter of the Line Protocol. A typical incoming payload looks like:
water,id=100 level=57.6 1556813561098000000
wind,id=227 speed=33.2 1556813561112000000Let’s post to and retrieve from a topic called /test/sensordata. Here’s the first part of telegraf.conf (I turn off the hostname so it’s not auto-inserted as a tag):
[agent]
omit_hostname = true
[[inputs.mqtt_consumer]]
servers = ["tcp://localhost:1883"]
topics = [ "/test/sensordata" ]
data_format = "influx"
topic_tag = ""The format is “influx” (Line Protocol) and MQTT runs on localhost. Next, we filter incoming messages with “outputs” sections — one per bucket:
[[outputs.influxdb_v2]]
urls = ["http://localhost:8086"]
token = "PASTE YOUR TOKEN IN HERE"
organization = "ResourceKraft"
namepass = ["wind"]
bucket = "wind"
[[outputs.influxdb_v2]]
urls = ["http://localhost:8086"]
token = "PASTE YOUR TOKEN IN HERE"
organization = "ResourceKraft"
namepass = ["water"]
bucket = "water"What’s most interesting are the namepass parameters — each only lets through messages whose measurement matches the string in quotes. So namepass = ["water"] passes only “water” measurements to the “water” bucket. Same for “wind”.
From the sensor, everything with measurement “wind” ends up in the “wind” bucket, and everything with “water” ends up in the “water” bucket. And that’s pretty much it.
There’s a lot more that can be done — decisions on topics (using regular expressions), multiple strings, and much more — but I hope this gets you started, and encourages you to use Telegraf with InfluxDB and MQTT to get your data flowing quickly and easily.
Some of you may be aghast at me installing everything on the same machine and being lax with security and hardcoded values — but I’m just playing with ideas and researching future projects, not building a production system. You can, of course, run each application on separate machines, encrypt your traffic, and keep your URLs, tokens and organisation in environment variables (as you should). Do please take the time to look into this if you’re setting up in the wild rather than on local secure machines and networks.
Robust device-to-database pipelines are the unglamorous plumbing behind every monitoring system we build — connecting custom hardware through to analytics platforms like Advisor. Getting the time-series fundamentals right (see also calculating usage from instantaneous data) is what makes the rest trustworthy. If you’re moving sensor data at scale, we can help.
Robust device-to-database pipelines — MQTT, Telegraf, time-series databases — are the plumbing behind every monitoring system we build. Let's get your data flowing.