Telegraf MQTT Consumer: Ingest MQTT Data into InfluxDB

The Telegraf MQTT consumer plugin is the simplest way to move MQTT messages into a time-series database without writing code. In this tutorial you will configure inputs.mqtt_consumer to subscribe to topics on broker.connectmq.com, parse JSON payloads with proper timestamps and tags, write the result to InfluxDB 2 and verify every step.

How Telegraf Fits Into an MQTT Pipeline

An MQTT broker routes messages; it does not store history. To chart last week's temperatures or alert on a trend, something has to subscribe to the topics and write each reading into a database. Telegraf is a small agent from InfluxData that does exactly that. It is configured with a single TOML file containing inputs, optional processors and outputs.

The mqtt_consumer input is a service plugin: it keeps a persistent connection to the broker, receives messages as they arrive, runs them through a parser to produce metrics, and hands them to outputs that flush in batches. From the broker's perspective, Telegraf is just another MQTT client. That is also the honest pattern for any hosted broker, including ConnectMQ: the broker carries the messages and an external tool like Telegraf takes care of storage.

Prerequisites

If you do not have InfluxDB yet, the quickest route is Docker:

docker run -d --name influxdb -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=iot \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Prefer a single command for the whole stack? The TIG stack Docker Compose guide runs Telegraf, InfluxDB and Grafana together.

Topic and Payload Design

This guide assumes the topic structure and JSON payload used across our Python tutorials:

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Pick your own random run ID, for example a1b2c3, so your topics do not collide with anyone else's on the public broker. If you want a continuous stream instead of manual test messages, run the Python MQTT device simulator, which publishes exactly this format.

A Complete Telegraf MQTT Consumer Configuration

Save the following as telegraf.conf:

[agent]
  interval = "10s"
  flush_interval = "10s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/telemetry"]
  qos = 0
  client_id = "telegraf-a1b2c3"
  connection_timeout = "30s"

  # Name of the measurement in InfluxDB (default would be "mqtt_consumer")
  name_override = "sensor"
  # Keep the full topic as a tag; set to "" to disable
  topic_tag = "topic"

  # Parse JSON payloads
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

  # Extract the production line from the topic as an extra tag
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/telemetry"
    tags = "_/_/_/line/_/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "iot"
  bucket = "sensors"

# Temporary: also print metrics to stdout while testing
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

What each setting does

• servers uses tcp:// for plain MQTT on port 1883. For TLS you switch to ssl:// and port 8883, shown below.
• topics accepts MQTT wildcards. The two + levels match any line and any device, while the fixed telemetry suffix excludes other message types.
• client_id should be set explicitly. Without it, Telegraf generates a random ID on each start, which prevents persistent sessions and makes the client harder to identify in broker logs.
• name_override sets the InfluxDB measurement name. Without it, every metric lands in a measurement called mqtt_consumer.
• tag_keys turns JSON keys into tags. Tags are indexed and suited to values you filter or group by, such as device IDs. Numeric readings become fields automatically.
• json_time_key and json_time_format use the device timestamp instead of the time Telegraf received the message. unix_ms matches our millisecond timestamps.
• topic_parsing maps topic segments to tags. Each _ skips a level and named segments become tags, so line is filled with the fourth level of the topic.

A word on the topic tag: it is convenient while debugging, but every distinct topic creates a new series in InfluxDB. With a few hundred devices that is harmless. With tens of thousands of topics, or topics that contain message IDs or timestamps, series cardinality grows quickly and slows queries. Once line and device tags carry the same information, setting topic_tag = "" keeps the database lean.

Set the token as an environment variable and run Telegraf in test mode first. Because mqtt_consumer is a service input, test mode needs a wait period to collect messages:

export INFLUX_TOKEN=my-super-secret-token
telegraf --config telegraf.conf --test --test-wait 20

While that runs, publish a test message from another terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/line1/sensor-0001/telemetry \
  -m "{\"device\":\"sensor-0001\",\"ts\":$(date +%s%3N),\"temperature\":21.7,\"humidity\":44.1}"

The date +%s%3N call produces a millisecond timestamp with GNU date on Linux; on macOS, install coreutils and use gdate, or type a fixed recent value. Telegraf should print a line of InfluxDB line protocol similar to this:

sensor,device=sensor-0001,line=line1,topic=connectmq/demo/a1b2c3/line1/sensor-0001/telemetry humidity=44.1,temperature=21.7 1760600000123000000

When that works, start Telegraf normally with telegraf --config telegraf.conf, remove the outputs.file block once you trust the pipeline, and query InfluxDB to confirm the data landed:

influx query 'from(bucket: "sensors") |> range(start: -15m) |> filter(fn: (r) => r._measurement == "sensor") |> limit(n: 5)' \
  --org iot --token my-super-secret-token

Securing the Connection for Production

The public broker is for experiments only; anything on it can be read by others. For real data, use a private broker with credentials and TLS. With a fully managed MQTT service such as ConnectMQ, the input block changes like this:

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/factory1/production/+/+/telemetry"]
  qos = 1
  client_id = "YOUR_CLIENT_ID"
  username = "${MQTT_USERNAME}"
  password = "${MQTT_PASSWORD}"
  persistent_session = true
  max_undelivered_messages = 1000

  name_override = "sensor"
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

The broker presents a Let's Encrypt certificate, which is trusted by the operating system's CA store, so no tls_ca setting is needed. persistent_session = true combined with QoS 1 and a fixed client ID asks the broker to keep Telegraf's session and queue messages during short Telegraf restarts. Keep credentials in environment variables rather than the config file, especially if the file lives in version control.

Troubleshooting the Telegraf MQTT Consumer

Run Telegraf with --debug to see connection events and parser errors as they happen. For payloads with nested objects or arrays, the legacy JSON parser quickly reaches its limits; the Telegraf json_v2 parser guide shows how to map nested structures and topic segments precisely. The full option list is in the official mqtt_consumer plugin documentation.

Where to Go Next

Once data is flowing into InfluxDB, add Grafana for dashboards and alerting. If you prefer code over configuration, the Python InfluxDB bridge does the same job in a short script with full control over mapping. And if you are still choosing client tooling for the devices themselves, see the MQTT client libraries guide.

Get Started with ConnectMQ

You can test this whole configuration against the public MQTT test broker in a few minutes. For production data, ConnectMQ gives you a private broker with TLS on port 8883, MQTT 5 and a portal with a live topic explorer to check what Telegraf will receive. Create a free Serverless broker at app.connectmq.com with no credit card, then paste the generated credentials into your Telegraf config. Read more about running an MQTT broker in the cloud if you are planning a larger deployment.

Frequently Asked Questions

Does the Telegraf MQTT consumer support MQTT 5?

Recent Telegraf versions add a protocol option to select MQTT 5. The default is MQTT 3.1.1, which is sufficient for subscribing and writing metrics in most setups.

Can one Telegraf instance subscribe to several topics?

Yes. List multiple filters in topics, or add several mqtt_consumer blocks with different parsers when topics carry different payload formats.

Why are my JSON strings not stored?

The legacy json parser ignores string values unless you list them in json_string_fields or tag_keys. The json_v2 parser handles strings explicitly.

Should Telegraf use QoS 0 or QoS 1?

QoS 0 is fine for high-rate telemetry where an occasional lost reading does not matter. Use QoS 1 with a persistent session when every reading must be stored.