MQTT to InfluxDB with Telegraf: mqtt_consumer Setup Guide

The shortest path from an MQTT broker to a time-series database is Telegraf: its mqtt_consumer input subscribes to your topics, a parser turns each payload into a metric, and the influxdb_v2 output writes it to InfluxDB. No code, one config file, and it runs happily next to the database or on an edge gateway.

This guide builds a working pipeline with Docker Compose, parses JSON payloads with json_v2, maps parts of the topic to tags, and shows how to prove every stage works before you point real devices at it.

The pipeline at a glance

The example device publishes JSON like this to testmqtt/telegraf/sensor-01/telemetry:

{"ts": 1760700000000, "site": "lab", "temperature": 21.4, "humidity": 40.2, "status": "ok"}

Start InfluxDB and Telegraf

This Compose file starts InfluxDB 2.x with an initial org, bucket and token, plus Telegraf with your config mounted read-only:

services:
  influxdb:
    image: influxdb:2.7
    ports: ["8086:8086"]
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: change-me-please
      DOCKER_INFLUXDB_INIT_ORG: testmqtt
      DOCKER_INFLUXDB_INIT_BUCKET: mqtt
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: dev-token-change-me
    volumes:
      - influx-data:/var/lib/influxdb2

  telegraf:
    image: telegraf:1.32
    depends_on: [influxdb]
    environment:
      MQTT_USER: ${MQTT_USER}
      MQTT_PASS: ${MQTT_PASS}
      INFLUX_TOKEN: dev-token-change-me
    volumes:
      - ./telegraf.conf:/etc/telegraf/telegraf.conf:ro

volumes:
  influx-data:

The init variables only apply on the first start with an empty volume. In production, create a dedicated write-only token for Telegraf instead of reusing the admin token. For a bigger local stack with a broker and dashboards, see the MQTT Docker Compose stack.

Configure mqtt_consumer

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

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["testmqtt/telegraf/+/telemetry"]
  qos = 1
  client_id = "telegraf-ingest-01"
  persistent_session = true
  username = "${MQTT_USER}"
  password = "${MQTT_PASS}"
  max_undelivered_messages = 1000
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "testmqtt/telegraf/+/telemetry"
    tags = "_/_/device/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "telemetry"
    timestamp_path = "ts"
    timestamp_format = "unix_ms"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "site"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temperature"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "testmqtt"
  bucket = "mqtt"

The settings that matter most:

• servers — tcp://host:1883 for plaintext, ssl://host:8883 for TLS. Telegraf verifies the server certificate against the system CA store; add tls_ca = "/path/ca.pem" for a private CA.
• topics — standard MQTT filters, so + and # work. Test a filter against real topic names with the topic matcher. Background on filters: topics and wildcards.
• qos = 1 + persistent_session = true + a fixed client_id — the broker keeps a session for Telegraf and queues QoS 1 messages while it restarts. Without a stable client ID the session cannot be resumed. See persistent sessions.
• max_undelivered_messages — how many QoS 1/2 messages Telegraf reads before they are written to the output. It is Telegraf's backpressure: if InfluxDB slows down, Telegraf stops acknowledging and the broker holds the rest.
• topic_tag = "" — disables the default topic tag. Storing the full topic as a tag multiplies series cardinality; topic_parsing extracts only the part you need.

How json_v2 maps the payload

With the config above, the example payload becomes this line protocol:

telemetry,device=sensor-01,site=lab temperature=21.4,humidity=40.2,status="ok" 1760700000000000000

Two rules keep InfluxDB healthy: tags for things you filter by and that have few distinct values, fields for measurements. And always send numbers as JSON numbers — "21.4" in quotes is a string and will either be dropped or create a type conflict on the field. If payloads are sometimes malformed, read MQTT JSON parse errors.

If devices do not send a timestamp, remove timestamp_path and Telegraf uses the arrival time. That is fine for live data, but buffered data from offline devices will then be stamped late.

Verify each stage

Test the pipeline in the order data flows, so you always know which stage is failing.

1. Is the data on the broker? Subscribe to testmqtt/telegraf/# with the online MQTT client and publish a test reading:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/telegraf/sensor-01/telemetry \
  -m "{\"ts\":$(date +%s000),\"site\":\"lab\",\"temperature\":21.4,\"humidity\":40.2,\"status\":\"ok\"}"

2. Does Telegraf parse it? Run Telegraf in test mode. It prints metrics to stdout instead of writing them; --test-wait keeps the MQTT input listening long enough to receive messages:

docker compose run --rm telegraf telegraf --config /etc/telegraf/telegraf.conf --test --test-wait 15

Publish again while it waits. You should see the telemetry,device=sensor-01,... line. No line means the subscription or parser is wrong; a line with missing fields means the json_v2 paths or types do not match.

3. Is it in InfluxDB? Start the stack with docker compose up -d and query the bucket:

docker compose exec influxdb influx query --org testmqtt \
  'from(bucket: "mqtt") |> range(start: -10m) |> filter(fn: (r) => r._measurement == "telemetry") |> last()'

Watch the telemetry topic in your browser

Common errors

Look up broker-side reason codes with the reason code tool. Parser errors are logged with the E! [inputs.mqtt_consumer] prefix and usually mean the payload is not valid JSON or a declared field has a different type. Run Telegraf with --debug to see connection and flush details.

Scaling and retention

• More throughput: Telegraf handles a lot of messages per instance, but if you need several instances, subscribe with a shared subscription such as $share/telegraf/testmqtt/telegraf/+/telemetry so the broker splits the stream instead of duplicating it. Each instance needs its own client_id.
• Retention: set a retention period on the bucket (in the UI or with influx bucket update --id <bucket-id> --retention 30d) and downsample older data with an InfluxDB task. Planning how long to keep raw telemetry is covered in MQTT data retention and storage.
• Visualization: point Grafana at InfluxDB for history, and add the Grafana MQTT data source if you also want a live panel straight from the broker.

Telegraf turns "MQTT into InfluxDB" into a config review instead of a software project. Decide your tags carefully, use QoS 1 with a persistent session on your own broker, verify with --test before going live, and keep the config in version control next to your dashboards.

Frequently asked questions