MQTT vers InfluxDB avec Telegraf : configurer mqtt_consumer

Le chemin le plus court entre un broker MQTT et une base de séries temporelles s'appelle Telegraf : son input mqtt_consumer s'abonne à vos topics, un parser transforme chaque payload en métrique et l'output influxdb_v2 l'écrit dans InfluxDB. Pas de code, un seul fichier de configuration, et il tourne sans souci à côté de la base ou sur une passerelle edge.

Ce guide monte un pipeline fonctionnel avec Docker Compose, parse les payloads JSON avec json_v2, transforme des parties du topic en tags et montre comment valider chaque étape avant d'y brancher de vrais appareils.

Le pipeline en un coup d'œil

L'appareil d'exemple publie un JSON de ce type sur testmqtt/telegraf/sensor-01/telemetry :

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

Démarrer InfluxDB et Telegraf

Ce fichier Compose démarre InfluxDB 2.x avec une organisation, un bucket et un token initiaux, ainsi que Telegraf avec votre configuration montée en lecture seule :

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:

Les variables d'initialisation ne s'appliquent qu'au premier démarrage avec un volume vide. En production, créez pour Telegraf un token dédié en écriture seule au lieu de réutiliser le token admin. Pour une stack locale plus complète avec broker et dashboards, voir la stack MQTT Docker Compose.

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

Les réglages qui comptent le plus :

• servers : tcp://host:1883 en clair, ssl://host:8883 pour TLS. Telegraf vérifie le certificat du serveur avec le magasin de CA du système ; ajoutez tls_ca = "/path/ca.pem" pour une CA privée.
• topics : des filtres MQTT standard, donc + et # fonctionnent. Testez un filtre sur de vrais noms de topics avec le topic matcher. Rappels sur les filtres : topics et wildcards.
• qos = 1 + persistent_session = true + un client_id fixe : le broker conserve une session pour Telegraf et met en file les messages QoS 1 pendant son redémarrage. Sans client ID stable, la session ne peut pas être reprise. Voir sessions persistantes.
• max_undelivered_messages : le nombre de messages QoS 1/2 que Telegraf lit avant qu'ils soient écrits dans l'output. C'est le backpressure de Telegraf : si InfluxDB ralentit, Telegraf cesse d'acquitter et le broker garde le reste.
• topic_tag = "" : désactive le tag topic par défaut. Stocker le topic complet en tag multiplie la cardinalité des séries ; topic_parsing n'extrait que la partie utile.

Comment json_v2 transforme le payload

Avec la configuration ci-dessus, le payload d'exemple devient ce line protocol :

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

Deux règles gardent InfluxDB en bonne santé : des tags pour ce que vous filtrez et qui a peu de valeurs distinctes, des fields pour les mesures. Et envoyez toujours les nombres comme des nombres JSON : "21.4" entre guillemets est une chaîne, qui sera ignorée ou créera un conflit de type sur le field. Si vos payloads sont parfois malformés, lisez les erreurs de parsing JSON MQTT.

Si les appareils n'envoient pas d'horodatage, supprimez timestamp_path : Telegraf utilisera l'heure d'arrivée. C'est suffisant pour du temps réel, mais les données bufferisées par des appareils hors ligne seront alors horodatées trop tard.

Vérifier chaque étape

Testez le pipeline dans l'ordre où circulent les données : vous saurez toujours quelle étape échoue.

1. Les données sont-elles sur le broker ? Abonnez-vous à testmqtt/telegraf/# avec le client MQTT en ligne et publiez une mesure de test :

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. Telegraf le parse-t-il ? Lancez Telegraf en mode test. Il affiche les métriques sur stdout au lieu de les écrire ; --test-wait garde l'input MQTT à l'écoute assez longtemps pour recevoir des messages :

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

Publiez à nouveau pendant l'attente. Vous devez voir la ligne telemetry,device=sensor-01,.... Aucune ligne signifie que l'abonnement ou le parser est faux ; une ligne avec des champs manquants signifie que les chemins ou les types json_v2 ne correspondent pas.

3. Est-ce dans InfluxDB ? Démarrez la stack avec docker compose up -d et interrogez le bucket :

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

Suivre le topic de télémétrie dans le navigateur

Erreurs fréquentes

Recherchez les reason codes côté broker avec l'outil reason codes. Les erreurs du parser sont journalisées avec le préfixe E! [inputs.mqtt_consumer] et signifient généralement que le payload n'est pas un JSON valide ou qu'un champ déclaré a un autre type. Lancez Telegraf avec --debug pour voir les détails de connexion et de flush.

Montée en charge et rétention

• Plus de débit : Telegraf traite beaucoup de messages par instance, mais si vous avez besoin de plusieurs instances, abonnez-vous via un abonnement partagé comme $share/telegraf/testmqtt/telegraf/+/telemetry afin que le broker répartisse le flux au lieu de le dupliquer. Chaque instance a besoin de son propre client_id.
• Rétention : définissez une durée de rétention sur le bucket (dans l'interface ou avec influx bucket update --id <bucket-id> --retention 30d) et sous-échantillonnez les anciennes données avec une task InfluxDB. La question de la durée de conservation de la télémétrie brute est traitée dans rétention et stockage des données MQTT.
• Visualisation : connectez Grafana à InfluxDB pour l'historique, et ajoutez la source de données MQTT de Grafana si vous voulez aussi un panneau en direct alimenté par le broker.

Avec Telegraf, « MQTT vers InfluxDB » devient une relecture de configuration plutôt qu'un projet logiciel. Choisissez vos tags avec soin, utilisez QoS 1 avec une session persistante sur votre propre broker, validez avec --test avant la mise en production et versionnez la configuration à côté de vos dashboards.

Questions fréquentes