Le parser json_v2 de Telegraf et le topic parsing MQTT expliqués

Le parser json_v2 de Telegraf vous permet de décider précisément quelles parties d'un payload MQTT deviennent des measurements, des tags et des fields, tandis que le topic parsing extrait des identifiants du topic lui-même. Dans ce tutoriel, vous allez configurer les deux sur l'input mqtt_consumer de Telegraf, à partir d'un payload imbriqué réaliste, et vérifier la sortie ligne par ligne avant que quoi que ce soit n'atteigne InfluxDB.

Pourquoi l'ancien parser JSON ne suffit pas

Le parser json d'origine de Telegraf aplatit tout ce qu'il trouve en fields, joint les clés imbriquées par des tirets bas, ignore les chaînes sauf si vous les listez et n'offre qu'un contrôle limité sur les types. C'est acceptable pour des payloads plats comme {"temperature": 21.5}. Les payloads industriels et IoT sont rarement aussi simples. Une passerelle machine publie typiquement quelque chose comme ceci :

{
  "ts": 1760600000123,
  "device": { "id": "cnc1", "fw": "2.4.1", "serial": "SN-88231" },
  "status": "running",
  "metrics": { "spindle_rpm": 11800, "temp_c": 41.7, "vibration_mm_s": 2.3 },
  "alarms": []
}

Vous voulez le firmware comme tag, le statut comme field de type chaîne, les trois métriques comme fields correctement typés, le numéro de série ignoré et l'horodatage de l'appareil utilisé comme instant du point. Le parser json_v2 exprime explicitement chacune de ces décisions grâce à la syntaxe de chemins GJSON : rien n'atterrit dans la base de données par accident.

La configuration d'exemple

Les messages arrivent sur un topic de style ISA-95 :

connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry

Cela fait sept niveaux, dans la limite de huit du serveur MQTT public gratuit. Le topic porte déjà le site, la ligne et la machine : la configuration les prendra donc dans le topic plutôt que de s'appuyer sur le payload.

Publiez un message de test sur broker.connectmq.com en remplaçant a1b2c3 par votre propre identifiant aléatoire :

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry \
  -m "{\"ts\":$(date +%s%3N),\"device\":{\"id\":\"cnc1\",\"fw\":\"2.4.1\",\"serial\":\"SN-88231\"},\"status\":\"running\",\"metrics\":{\"spindle_rpm\":11800,\"temp_c\":41.7,\"vibration_mm_s\":2.3},\"alarms\":[]}"

Configuration du parser json_v2 de Telegraf

Enregistrez ceci sous json_v2.conf :

[agent]
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/telemetry"]
  client_id = "telegraf-json-a1b2c3"
  qos = 0
  topic_tag = ""
  data_format = "json_v2"

  # Le site, la ligne et la machine proviennent du topic
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/telemetry"
    tags = "_/_/_/site/line/asset/_"

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

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "device.fw"
      rename = "firmware"

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

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.spindle_rpm"
      rename = "spindle_rpm"
      type = "int"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.temp_c"
      rename = "temp_c"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.vibration_mm_s"
      rename = "vibration_mm_s"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "alarms.#"
      rename = "alarm_count"
      type = "int"

[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Lancez-le en mode test. mqtt_consumer est un input de service : laissez-lui le temps de recevoir un message, et publiez le message de test pendant qu'il attend :

telegraf --config json_v2.conf --test --test-wait 20

La sortie est une seule ligne de line protocol :

machine,asset=cnc1,firmware=2.4.1,line=line1,site=factory1 status="running",spindle_rpm=11800i,temp_c=41.7,vibration_mm_s=2.3,alarm_count=0i 1760600000123000000

Comment fonctionne le mapping

Les chemins utilisent la syntaxe GJSON

Chaque path est une expression GJSON évaluée sur le payload. Un point descend dans les objets imbriqués : metrics.temp_c lit donc la température à l'intérieur de metrics. Le modificateur # renvoie la longueur d'un tableau, ce qui explique pourquoi alarms.# produit un nombre d'alarmes sans stocker le tableau lui-même. GJSON prend aussi en charge les index de tableau comme readings.0.value et des requêtes comme readings.#(sensor=="t1").value lorsque vous devez sélectionner un élément.

Tags, fields et types

Déclarer explicitement les types évite un problème classique d'InfluxDB : un field écrit une fois comme entier puis plus tard comme flottant est rejeté pour conflit de type. Choisissez un type par field et conservez-le pour toujours.

Sans rename, un chemin imbriqué devient un nom de field correspondant au chemin aplati, ce qui tend à produire des noms longs et maladroits. Le renommage garde un schéma lisible dans les requêtes et les tableaux de bord.

Le topic parsing complète la hiérarchie

Le bloc topic_parsing découpe le topic sur / et l'aligne sur le motif de tags. Chaque _ saute un niveau et chaque nom crée un tag à partir de cette position. Le motif doit comporter exactement autant de segments que le topic, et le filtre topic à l'intérieur du bloc détermine à quels messages la règle s'applique, les wildcards + et # étant autorisés.

Le topic parsing peut aussi définir la measurement et créer des fields. Voici la même règle qui prend la measurement dans le dernier niveau du topic au lieu du measurement_name fixe :

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/+"
    measurement = "_/_/_/_/_/_/measurement"
    tags = "_/_/_/site/line/asset/_"

Avec cette règle, les messages sur .../cnc1/telemetry arrivent dans une measurement telemetry et ceux sur .../cnc1/energy dans une measurement energy. Le topic parsing s'exécute après le parser de payload : la measurement issue du topic remplace donc celle de measurement_name. Pour des valeurs numériques transportées dans le topic lui-même, utilisez fields avec une table [inputs.mqtt_consumer.topic_parsing.types] pour définir leurs types, mais c'est rarement une bonne conception de topics.

Parser des tableaux en plusieurs métriques

Les passerelles regroupent souvent plusieurs mesures dans un seul message :

{
  "gateway": "gw-07",
  "readings": [
    { "sensor": "t1", "ts": 1760600000100, "value": 21.3 },
    { "sensor": "t2", "ts": 1760600000105, "value": 22.0 },
    { "sensor": "h1", "ts": 1760600000110, "value": 48.5 }
  ]
}

Un bloc object transforme chaque élément du tableau en métrique distincte. Placez-le dans un input mqtt_consumer séparé qui écoute le topic des lots, afin qu'il ne s'exécute jamais sur des payloads dépourvus de tableau readings :

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/batch"]
  client_id = "telegraf-batch-a1b2c3"
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/batch"
    tags = "_/_/_/site/line/gateway/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "reading"

    [[inputs.mqtt_consumer.json_v2.object]]
      path = "readings"
      tags = ["sensor"]
      timestamp_key = "ts"
      timestamp_format = "unix_ms"
      [inputs.mqtt_consumer.json_v2.object.fields]
        value = "float"

Chaque élément produit une ligne comme reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000. La liste tags promeut des clés en tags, timestamp_key lit l'horodatage de chaque élément et la table fields fixe les types. Notez que le tag gateway provient du topic, ce qui permet au bloc object de se concentrer sur le tableau lui-même.

Checklist de dépannage

1. Aucune sortie en mode test. Vérifiez le topic avec mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v et assurez-vous de publier pendant la fenêtre --test-wait.
2. Un field manque dans la sortie. Vérifiez le chemin GJSON par rapport au payload réel ; une faute de frappe dans un chemin ne produit silencieusement rien. Les versions récentes de Telegraf acceptent optional = true sur les fields qui ne sont pas toujours présents.
3. Erreur sur les segments du topic parsing. Comptez les niveaux du filtre topic et du motif tags ; ils doivent correspondre.
4. Horodatages en 1970. Le payload utilise des secondes alors que la configuration indique unix_ms, ou l'inverse.
5. Erreurs de conflit de type dans InfluxDB. Une configuration antérieure a écrit le même field avec un autre type. Utilisez un nouveau nom de field ou un nouveau bucket lorsque vous changez de type.
6. Toutes les valeurs sont des chaînes. Un type de field a été défini à string, ou l'appareil envoie des nombres entre guillemets, comme "41.7". Définir type = "float" convertit les nombres entre guillemets.
7. Trop de séries. Une valeur à forte cardinalité, comme un numéro de série ou un identifiant de message, a été mappée en tag. Faites-en un field ou supprimez-la.

Passer en production

Une fois le mapping correct, supprimez le bloc outputs.file, ajoutez un output [[outputs.influxdb_v2]] et basculez l'input vers un broker privé. Avec un broker MQTT managé comme ConnectMQ, cela signifie servers = ["ssl://broker.connectmq.com:8883"], username, password et le client ID délivré dans le portail. Avant d'écrire des règles de parsing pour un nouveau type d'appareil, l'explorateur de topics en direct du portail est un moyen pratique de copier un payload réel. ConnectMQ décode aussi Sparkplug B et peut le republier dans des topics ISA-95, si bien que Telegraf peut s'abonner à des topics ISA-95 propres ; inspectez un payload republié dans le portail avant d'écrire les chemins correspondants.

Pour les bases du plugin d'input, commencez par le tutoriel Telegraf MQTT consumer. Pour exécuter le pipeline complet avec Grafana, utilisez la configuration de la stack TIG avec Docker Compose, et pour choisir les bibliothèques côté appareil, consultez le guide des bibliothèques clientes MQTT. La référence officielle du parser json_v2 liste toutes les options.

Démarrez avec ConnectMQ

Testez vos règles de parsing sur le serveur MQTT public gratuit avec des payloads d'exemple, puis passez à des identifiants privés pour les vraies machines. ConnectMQ exploite un broker MQTT managé compatible MQTT 5, avec TLS et un portail qui affiche les topics et les payloads en direct. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire. Pour une vue plus large des options d'hébergement, consultez le guide du broker MQTT cloud.

Questions fréquentes

Quelle est la différence entre les parsers json et json_v2 de Telegraf ?

L'ancien parser json aplatit automatiquement les payloads avec un contrôle limité. Le parser json_v2 utilise des chemins GJSON explicites pour les tags, les fields, les types et les horodatages, ce qui convient bien mieux aux payloads IoT imbriqués.

Telegraf peut-il prendre des tags dans le topic MQTT ?

Oui. Un bloc topic_parsing associe les niveaux du topic à des tags, des fields ou au nom de la measurement. Utilisez _ pour les niveaux à ignorer.

Comment parser un tableau JSON en métriques distinctes ?

Utilisez un bloc object avec le chemin du tableau. Chaque élément devient sa propre métrique, et vous pouvez promouvoir des clés en tags et définir un horodatage par élément.

Ai-je besoin de json_v2 pour des payloads plats ?

Non. Pour des payloads numériques plats, l'ancien parser fonctionne très bien. Passez à json_v2 lorsque vous avez besoin de chaînes, de types explicites, de fields renommés ou de structures imbriquées.