Telegraf MQTT consumer : ingérer des données MQTT dans InfluxDB

Le plugin Telegraf MQTT consumer est le moyen le plus simple de faire passer des messages MQTT dans une base de données de séries temporelles sans écrire de code. Dans ce tutoriel, vous allez configurer inputs.mqtt_consumer pour s'abonner à des topics sur le serveur MQTT public gratuit broker.connectmq.com, parser des payloads JSON avec des horodatages et des tags corrects, écrire le résultat dans InfluxDB 2 et vérifier chaque étape.

La place de Telegraf dans un pipeline MQTT

Un broker MQTT route les messages ; il ne stocke pas d'historique. Pour tracer les températures de la semaine dernière ou déclencher une alerte sur une tendance, quelque chose doit s'abonner aux topics et écrire chaque mesure dans une base de données. Telegraf est un petit agent d'InfluxData qui fait exactement cela. Il se configure avec un unique fichier TOML contenant des inputs, des processors facultatifs et des outputs.

L'input mqtt_consumer est un plugin de service : il maintient une connexion persistante avec le broker, reçoit les messages à leur arrivée, les fait passer par un parseur pour produire des métriques, puis les transmet aux outputs qui les écrivent par lots. Du point de vue du broker, Telegraf n'est qu'un client MQTT de plus. C'est aussi le modèle honnête pour tout broker MQTT managé, ConnectMQ compris : le broker transporte les messages et un outil externe comme Telegraf se charge du stockage.

Prérequis

Si vous n'avez pas encore InfluxDB, le plus rapide est de passer par 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

Vous préférez une seule commande pour toute la stack ? Le guide de la stack TIG avec Docker Compose lance Telegraf, InfluxDB et Grafana ensemble.

Conception des topics et du payload

Ce guide suppose la structure de topics et le payload JSON utilisés dans nos tutoriels Python :

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

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

Choisissez votre propre run ID aléatoire, par exemple a1b2c3, pour que vos topics n'entrent pas en collision avec ceux des autres sur le broker public. Si vous voulez un flux continu plutôt que des messages de test manuels, lancez le simulateur d'appareils MQTT en Python, qui publie exactement ce format.

Une configuration Telegraf MQTT consumer complète

Enregistrez le contenu suivant sous 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"

  # Nom de la measurement dans InfluxDB (par défaut "mqtt_consumer")
  name_override = "sensor"
  # Conserver le topic complet comme tag ; mettre "" pour désactiver
  topic_tag = "topic"

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

  # Extraire la ligne de production du topic comme tag supplémentaire
  [[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"

# Temporaire : afficher aussi les métriques sur stdout pendant les tests
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Le rôle de chaque paramètre

• servers utilise tcp:// pour MQTT en clair sur le port 1883. Pour TLS, passez à ssl:// et au port 8883, comme montré plus bas.
• topics accepte les wildcards MQTT. Les deux niveaux + correspondent à n'importe quelle ligne et n'importe quel appareil, tandis que le suffixe fixe telemetry exclut les autres types de messages.
• client_id doit être défini explicitement. Sans lui, Telegraf génère un ID aléatoire à chaque démarrage, ce qui empêche les sessions persistantes et rend le client plus difficile à identifier dans les journaux du broker.
• name_override définit le nom de la measurement InfluxDB. Sans lui, toutes les métriques atterrissent dans une measurement nommée mqtt_consumer.
• tag_keys transforme des clés JSON en tags. Les tags sont indexés et adaptés aux valeurs sur lesquelles vous filtrez ou regroupez, comme les identifiants d'appareils. Les mesures numériques deviennent automatiquement des fields.
• json_time_key et json_time_format utilisent l'horodatage de l'appareil plutôt que l'heure de réception par Telegraf. unix_ms correspond à nos horodatages en millisecondes.
• topic_parsing associe des segments du topic à des tags. Chaque _ saute un niveau et les segments nommés deviennent des tags : line reçoit donc le quatrième niveau du topic.

Un mot sur le tag topic : il est pratique pour le débogage, mais chaque topic distinct crée une nouvelle série dans InfluxDB. Avec quelques centaines d'appareils, c'est sans conséquence. Avec des dizaines de milliers de topics, ou des topics contenant des identifiants de message ou des horodatages, la cardinalité des séries augmente vite et ralentit les requêtes. Dès que les tags line et device portent la même information, définir topic_tag = "" garde une base de données légère.

Définissez le token comme variable d'environnement et lancez d'abord Telegraf en mode test. Comme mqtt_consumer est un input de service, le mode test nécessite un délai d'attente pour collecter les messages :

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

Pendant ce temps, publiez un message de test depuis un autre 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}"

L'appel date +%s%3N produit un horodatage en millisecondes avec GNU date sous Linux ; sous macOS, installez coreutils et utilisez gdate, ou saisissez une valeur récente fixe. Telegraf devrait afficher une ligne de line protocol InfluxDB semblable à celle-ci :

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

Une fois que cela fonctionne, lancez Telegraf normalement avec telegraf --config telegraf.conf, supprimez le bloc outputs.file lorsque vous avez confiance dans le pipeline, et interrogez InfluxDB pour confirmer l'arrivée des données :

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

Sécuriser la connexion pour la production

Le serveur MQTT public gratuit est réservé aux expérimentations ; tout ce qui y transite peut être lu par d'autres. Pour des données réelles, utilisez un broker privé avec identifiants et TLS. Avec un broker MQTT managé comme ConnectMQ, le bloc d'input devient :

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

Le broker présente un certificat Let's Encrypt, reconnu par le magasin de CA du système d'exploitation : aucun paramètre tls_ca n'est donc nécessaire. persistent_session = true combiné à la QoS 1 et à un client ID fixe demande au broker de conserver la session de Telegraf et de mettre les messages en file pendant ses courts redémarrages. Gardez les identifiants dans des variables d'environnement plutôt que dans le fichier de configuration, surtout si celui-ci est versionné.

Dépanner le Telegraf MQTT consumer

Lancez Telegraf avec --debug pour voir les événements de connexion et les erreurs de parsing en temps réel. Pour les payloads avec objets imbriqués ou tableaux, l'ancien parseur JSON atteint vite ses limites ; le guide du parseur json_v2 de Telegraf montre comment mapper précisément les structures imbriquées et les segments de topic. La liste complète des options figure dans la documentation officielle du plugin mqtt_consumer.

Pour aller plus loin

Une fois que les données arrivent dans InfluxDB, ajoutez Grafana pour les tableaux de bord et les alertes. Si vous préférez le code à la configuration, la passerelle Python vers InfluxDB fait le même travail dans un court script, avec un contrôle total sur le mapping. Et si vous choisissez encore l'outillage client pour les appareils eux-mêmes, consultez le guide des bibliothèques clientes MQTT.

Démarrez avec ConnectMQ

Vous pouvez tester toute cette configuration sur le serveur MQTT public gratuit en quelques minutes. Pour les données de production, ConnectMQ vous fournit un broker privé avec TLS sur le port 8883, MQTT 5 et un portail doté d'un explorateur de topics en direct pour vérifier ce que Telegraf va recevoir. Créez un broker Serverless gratuit sur app.connectmq.com sans carte bancaire, puis collez les identifiants générés dans votre configuration Telegraf. Pour en savoir plus sur l'exploitation d'un broker MQTT dans le cloud, consultez notre guide si vous prévoyez un déploiement plus important.

Questions fréquentes

Le Telegraf MQTT consumer prend-il en charge MQTT 5 ?

Les versions récentes de Telegraf ajoutent une option protocol pour sélectionner MQTT 5. La valeur par défaut est MQTT 3.1.1, ce qui suffit pour s'abonner et écrire des métriques dans la plupart des configurations.

Une instance de Telegraf peut-elle s'abonner à plusieurs topics ?

Oui. Listez plusieurs filtres dans topics, ou ajoutez plusieurs blocs mqtt_consumer avec des parseurs différents lorsque les topics transportent des formats de payload différents.

Pourquoi mes chaînes JSON ne sont-elles pas stockées ?

L'ancien parseur json ignore les valeurs de type chaîne, sauf si vous les listez dans json_string_fields ou tag_keys. Le parseur json_v2 gère les chaînes explicitement.

Telegraf doit-il utiliser la QoS 0 ou la QoS 1 ?

La QoS 0 convient à la télémétrie à haut débit, où la perte occasionnelle d'une mesure n'a pas d'importance. Utilisez la QoS 1 avec une session persistante lorsque chaque mesure doit être stockée.