El parser json_v2 de Telegraf y el topic parsing de MQTT, explicados

El parser json_v2 de Telegraf te permite decidir exactamente qué partes de un payload MQTT se convierten en measurements, tags y fields, mientras que el topic parsing extrae identificadores del propio topic. En este tutorial configurarás ambos en el input mqtt_consumer de Telegraf, partiendo de un payload anidado realista, y verificarás la salida línea a línea antes de que nada llegue a InfluxDB.

Por qué no basta con el parser JSON heredado

El parser json original de Telegraf aplana todo lo que encuentra en fields, une las claves anidadas con guiones bajos, ignora las cadenas salvo que las enumeres y ofrece un control limitado sobre los tipos. Eso es aceptable para payloads planos como {"temperature": 21.5}. Los payloads industriales e IoT rara vez son tan sencillos. Un gateway de máquina suele publicar algo así:

{
  "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": []
}

Quieres el firmware como tag, el estado como field de texto, las tres métricas como fields con el tipo correcto, el número de serie ignorado y la marca de tiempo del dispositivo como hora del punto. El parser json_v2 expresa cada una de esas decisiones de forma explícita mediante la sintaxis de rutas GJSON, así que nada acaba en la base de datos por accidente.

La configuración de ejemplo

Los mensajes llegan a un topic de estilo ISA-95:

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

Son siete niveles, dentro del límite de ocho del servidor público. El topic ya incluye el emplazamiento, la línea y la máquina, así que la configuración los tomará del topic en lugar de depender del payload.

Publica un mensaje de prueba en broker.connectmq.com, el servidor MQTT público gratuito de ConnectMQ (solo para pruebas, no para datos de producción), con tu propio ID aleatorio en lugar de a1b2c3:

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\":[]}"

Configuración del parser json_v2 de Telegraf

Guarda esto como 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"

  # El emplazamiento, la línea y la máquina vienen del 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"

Ejecútalo en modo de prueba. mqtt_consumer es un input de servicio, así que dale tiempo para recibir un mensaje y publica el mensaje de prueba mientras espera:

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

La salida es una sola línea 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

Cómo funciona el mapeo

Las rutas usan la sintaxis GJSON

Cada path es una expresión GJSON evaluada contra el payload. Un punto entra en los objetos anidados, así que metrics.temp_c lee la temperatura dentro de metrics. El modificador # devuelve la longitud de un array, y por eso alarms.# produce un recuento de alarmas sin guardar el array en sí. GJSON también admite índices de array como readings.0.value y consultas como readings.#(sensor=="t1").value cuando necesitas elegir un elemento.

Tags, fields y tipos

Declarar los tipos explícitamente evita un problema clásico de InfluxDB: un field escrito una vez como entero y más tarde como float se rechaza por conflicto de tipo. Elige un tipo por field y mantenlo para siempre.

Sin rename, una ruta anidada se convierte en un nombre de field con la ruta aplanada, lo que suele producir nombres largos y poco prácticos. Renombrar mantiene legible el esquema en las consultas y los dashboards.

El topic parsing completa la jerarquía

El bloque topic_parsing divide el topic por / y lo alinea con el patrón de tags. Cada _ salta un nivel y cada nombre crea un tag a partir de esa posición. El patrón debe tener exactamente tantos segmentos como el topic, y el filtro topic dentro del bloque decide a qué mensajes se aplica la regla, con los comodines + y # permitidos.

El topic parsing también puede fijar la measurement y crear fields. Esta es la misma regla tomando la measurement del último nivel del topic en lugar del measurement_name fijo:

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

Con esa regla, los mensajes de .../cnc1/telemetry acaban en una measurement telemetry y los de .../cnc1/energy en una measurement energy. El topic parsing se ejecuta después del parser del payload, así que la measurement del topic sustituye a la de measurement_name. Para valores numéricos incluidos en el propio topic, usa fields junto con una tabla [inputs.mqtt_consumer.topic_parsing.types] para fijar sus tipos, aunque rara vez es un buen diseño de topics.

Interpretar arrays como varias métricas

Los gateways suelen agrupar varias lecturas en un mensaje:

{
  "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 bloque object convierte cada elemento del array en su propia métrica. Ponlo en un input mqtt_consumer separado que escuche en el topic de lotes, para que nunca se ejecute sobre payloads sin un array 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"

Cada elemento genera una línea como reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000. La lista tags convierte claves en tags, timestamp_key lee la hora de cada elemento y la tabla fields fija los tipos. Observa que el tag del gateway viene del topic, lo que mantiene el bloque object centrado en el propio array.

Lista de comprobación para resolver problemas

1. Sin salida en modo de prueba. Confirma el topic con mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v y asegúrate de publicar durante la ventana de --test-wait.
2. Falta un field en la salida. Comprueba la ruta GJSON contra el payload real; una errata en una ruta no devuelve nada sin avisar. Las versiones más recientes de Telegraf aceptan optional = true en los fields que no siempre están presentes.
3. Error sobre los segmentos del topic parsing. Cuenta los niveles del filtro topic y del patrón tags; deben coincidir.
4. Marcas de tiempo en 1970. El payload usa segundos pero la configuración dice unix_ms, o al revés.
5. Errores de conflicto de tipo en InfluxDB. Una configuración anterior escribió el mismo field con otro tipo. Usa un nombre de field nuevo o un bucket nuevo al cambiar tipos.
6. Todos los valores son texto. Se fijó el type de un field en string o el dispositivo envía números entre comillas, como "41.7". Fijar type = "float" convierte los números entre comillas.
7. Demasiadas series. Se asignó a un tag un valor de alta cardinalidad, como un número de serie o un ID de mensaje. Conviértelo en field o descártalo.

Ejecutarlo en producción

Cuando el mapeo sea correcto, elimina el bloque outputs.file, añade un output [[outputs.influxdb_v2]] y cambia el input a un broker privado. Con un broker MQTT gestionado como ConnectMQ, eso significa servers = ["ssl://broker.connectmq.com:8883"], username, password y el client ID emitido en el portal. Antes de escribir reglas de interpretación para un nuevo tipo de dispositivo, el explorador de topics en vivo del portal es una forma cómoda de copiar un payload real. ConnectMQ también decodifica Sparkplug B y puede republicarlo en topics ISA-95, de modo que Telegraf puede suscribirse a topics ISA-95 limpios; inspecciona un payload republicado en el portal antes de escribir rutas para él.

Para los fundamentos del plugin de input, empieza por el tutorial del Telegraf MQTT consumer. Para ejecutar el pipeline completo con Grafana, usa la configuración del stack TIG con Docker Compose y, para elegir las bibliotecas del lado del dispositivo, consulta la guía de bibliotecas cliente MQTT. La referencia oficial del parser json_v2 enumera todas las opciones.

Empieza con ConnectMQ

Prueba tus reglas de interpretación contra el servidor MQTT público gratuito con payloads de ejemplo y pasa después a credenciales privadas para las máquinas reales. ConnectMQ ofrece un broker MQTT gestionado compatible con MQTT 5, con TLS y un portal que muestra topics y payloads en vivo. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito. Para una visión más amplia de las opciones de alojamiento, consulta la guía del broker MQTT en la nube.

Preguntas frecuentes

¿Qué diferencia hay entre los parsers json y json_v2 de Telegraf?

El parser json heredado aplana los payloads automáticamente con un control limitado. El parser json_v2 usa rutas GJSON explícitas para tags, fields, tipos y marcas de tiempo, lo que encaja mucho mejor con los payloads IoT anidados.

¿Puede Telegraf tomar tags del topic MQTT?

Sí. Un bloque topic_parsing asigna los niveles del topic a tags, fields o al nombre de la measurement. Usa _ para los niveles que quieras saltar.

¿Cómo interpreto un array JSON como métricas separadas?

Usa un bloque object con la ruta del array. Cada elemento se convierte en su propia métrica, y puedes convertir claves en tags y fijar una marca de tiempo por elemento.

¿Necesito json_v2 para payloads planos?

No. Para payloads numéricos planos el parser heredado funciona bien. Cambia a json_v2 cuando necesites cadenas, tipos explícitos, fields renombrados o estructuras anidadas.