MQTT zu InfluxDB mit Telegraf: mqtt_consumer einrichten

Der kürzeste Weg von einem MQTT-Broker in eine Zeitreihendatenbank heißt Telegraf: Der Input mqtt_consumer abonniert deine Topics, ein Parser macht aus jedem Payload eine Metrik, und der Output influxdb_v2 schreibt sie in InfluxDB. Kein Code, eine Konfigurationsdatei – und es läuft problemlos neben der Datenbank oder auf einem Edge-Gateway.

Diese Anleitung baut eine funktionierende Pipeline mit Docker Compose, parst JSON-Payloads mit json_v2, bildet Teile des Topics auf Tags ab und zeigt, wie du jede Stufe prüfst, bevor echte Geräte darauf zeigen.

Die Pipeline im Überblick

Das Beispielgerät publiziert JSON wie dieses auf testmqtt/telegraf/sensor-01/telemetry:

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

InfluxDB und Telegraf starten

Diese Compose-Datei startet InfluxDB 2.x mit initialer Organisation, Bucket und Token sowie Telegraf mit deiner schreibgeschützt eingebundenen Konfiguration:

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:

Die Init-Variablen greifen nur beim ersten Start mit leerem Volume. In Produktion legst du für Telegraf ein eigenes Token mit reinem Schreibrecht an, statt das Admin-Token wiederzuverwenden. Für einen größeren lokalen Stack mit Broker und Dashboards siehe den MQTT-Docker-Compose-Stack.

mqtt_consumer konfigurieren

[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"

Die wichtigsten Einstellungen:

• servers – tcp://host:1883 unverschlüsselt, ssl://host:8883 für TLS. Telegraf prüft das Serverzertifikat gegen den CA-Store des Systems; für eine private CA ergänzt du tls_ca = "/path/ca.pem".
• topics – normale MQTT-Filter, + und # funktionieren also. Teste einen Filter mit dem Topic-Matcher gegen echte Topic-Namen. Hintergrund zu Filtern: Topics und Wildcards.
• qos = 1 + persistent_session = true + feste client_id – der Broker hält eine Session für Telegraf und puffert QoS-1-Nachrichten, während es neu startet. Ohne stabile Client-ID lässt sich die Session nicht fortsetzen. Siehe Persistent Sessions.
• max_undelivered_messages – wie viele QoS-1/2-Nachrichten Telegraf liest, bevor sie in den Output geschrieben sind. Das ist Telegrafs Backpressure: Wird InfluxDB langsam, bestätigt Telegraf nicht mehr, und der Broker hält den Rest zurück.
• topic_tag = "" – deaktiviert den Standard-Tag topic. Das komplette Topic als Tag zu speichern, vervielfacht die Serien-Kardinalität; topic_parsing extrahiert nur den Teil, den du brauchst.

Wie json_v2 den Payload abbildet

Mit der obigen Konfiguration wird aus dem Beispiel-Payload dieses Line Protocol:

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

Zwei Regeln halten InfluxDB gesund: Tags für alles, wonach du filterst und das wenige unterschiedliche Werte hat, Fields für Messwerte. Und sende Zahlen immer als JSON-Zahlen – "21.4" in Anführungszeichen ist ein String und wird entweder verworfen oder erzeugt einen Typkonflikt im Field. Wenn Payloads manchmal fehlerhaft sind, lies MQTT-JSON-Parse-Fehler.

Senden Geräte keinen Zeitstempel, entferne timestamp_path; Telegraf nutzt dann die Ankunftszeit. Für Live-Daten passt das, aber gepufferte Daten von Geräten, die offline waren, bekommen dann einen zu späten Zeitstempel.

Jede Stufe prüfen

Teste die Pipeline in der Reihenfolge, in der die Daten fließen – so weißt du immer, welche Stufe hakt.

1. Sind die Daten auf dem Broker? Abonniere testmqtt/telegraf/# im Online-MQTT-Client und publiziere einen Testwert:

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. Parst Telegraf ihn? Starte Telegraf im Testmodus. Es gibt Metriken auf stdout aus, statt sie zu schreiben; --test-wait hält den MQTT-Input lange genug offen, um Nachrichten zu empfangen:

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

Publiziere erneut, während es wartet. Du solltest die Zeile telemetry,device=sensor-01,... sehen. Keine Zeile heißt: Abonnement oder Parser stimmen nicht; eine Zeile mit fehlenden Feldern heißt: json_v2-Pfade oder -Typen passen nicht.

3. Ist es in InfluxDB? Starte den Stack mit docker compose up -d und frage den Bucket ab:

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

Telemetrie-Topic im Browser beobachten

Häufige Fehler

Broker-seitige Reason Codes schlägst du mit dem Reason-Code-Tool nach. Parser-Fehler werden mit dem Präfix E! [inputs.mqtt_consumer] geloggt und bedeuten meist, dass der Payload kein gültiges JSON ist oder ein deklariertes Feld einen anderen Typ hat. Starte Telegraf mit --debug, um Details zu Verbindung und Flush zu sehen.

Skalierung und Aufbewahrung

• Mehr Durchsatz: Telegraf verarbeitet pro Instanz viele Nachrichten. Brauchst du mehrere Instanzen, abonniere über eine Shared Subscription wie $share/telegraf/testmqtt/telegraf/+/telemetry, damit der Broker den Strom aufteilt, statt ihn zu duplizieren. Jede Instanz braucht ihre eigene client_id.
• Aufbewahrung: Setze eine Retention Period auf dem Bucket (in der Oberfläche oder mit influx bucket update --id <bucket-id> --retention 30d) und downsample ältere Daten mit einem InfluxDB-Task. Wie lange du Rohtelemetrie aufbewahren solltest, behandelt MQTT-Datenaufbewahrung und -Speicherung.
• Visualisierung: Verbinde Grafana mit InfluxDB für die Historie und ergänze die Grafana-MQTT-Datenquelle, wenn du zusätzlich ein Live-Panel direkt vom Broker willst.

Telegraf macht aus „MQTT nach InfluxDB“ ein Konfigurations-Review statt eines Softwareprojekts. Wähle deine Tags mit Bedacht, nutze QoS 1 mit Persistent Session auf deinem eigenen Broker, prüfe mit --test, bevor du live gehst, und halte die Konfiguration neben deinen Dashboards in der Versionsverwaltung.

Häufig gestellte Fragen