Telegraf MQTT consumer: ingesta de datos MQTT en InfluxDB

El plugin Telegraf MQTT consumer es la forma más sencilla de llevar mensajes MQTT a una base de datos de series temporales sin escribir código. En este tutorial configurarás inputs.mqtt_consumer para suscribirte a topics en broker.connectmq.com, el servidor MQTT público gratuito de ConnectMQ, interpretar payloads JSON con marcas de tiempo y tags correctos, escribir el resultado en InfluxDB 2 y verificar cada paso.

Cómo encaja Telegraf en un pipeline MQTT

Un broker MQTT enruta mensajes; no guarda historial. Para representar las temperaturas de la semana pasada o alertar sobre una tendencia, algo tiene que suscribirse a los topics y escribir cada lectura en una base de datos. Telegraf es un pequeño agente de InfluxData que hace exactamente eso. Se configura con un único archivo TOML que contiene inputs, processors opcionales y outputs.

El input mqtt_consumer es un plugin de servicio: mantiene una conexión persistente con el broker, recibe los mensajes a medida que llegan, los pasa por un parser para generar métricas y los entrega a los outputs, que los vuelcan por lotes. Desde el punto de vista del broker, Telegraf es un cliente MQTT más. Ese es también el patrón honesto para cualquier broker MQTT gestionado, ConnectMQ incluido: el broker transporta los mensajes y una herramienta externa como Telegraf se encarga del almacenamiento.

Requisitos previos

Si aún no tienes InfluxDB, la vía más rápida es 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

¿Prefieres un solo comando para todo el stack? La guía del stack TIG con Docker Compose ejecuta Telegraf, InfluxDB y Grafana juntos.

Diseño de topics y payloads

Esta guía supone la estructura de topics y el payload JSON que usamos en nuestros tutoriales de Python:

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

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

Elige tu propio run ID aleatorio, por ejemplo a1b2c3, para que tus topics no choquen con los de nadie en el servidor público. Si quieres un flujo continuo en lugar de mensajes de prueba manuales, ejecuta el simulador de dispositivos MQTT en Python, que publica exactamente este formato.

Una configuración completa del Telegraf MQTT consumer

Guarda lo siguiente como 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"

  # Nombre de la measurement en InfluxDB (por defecto sería "mqtt_consumer")
  name_override = "sensor"
  # Conserva el topic completo como tag; pon "" para desactivarlo
  topic_tag = "topic"

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

  # Extraer la línea de producción del topic como tag adicional
  [[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"

# Temporal: imprimir también las métricas en stdout durante las pruebas
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Qué hace cada ajuste

• servers usa tcp:// para MQTT en claro en el puerto 1883. Para TLS cambias a ssl:// y al puerto 8883, como se muestra más abajo.
• topics acepta comodines MQTT. Los dos niveles + coinciden con cualquier línea y cualquier dispositivo, mientras que el sufijo fijo telemetry excluye otros tipos de mensaje.
• client_id debe fijarse explícitamente. Sin él, Telegraf genera un ID aleatorio en cada arranque, lo que impide las sesiones persistentes y dificulta identificar al cliente en los registros del broker.
• name_override fija el nombre de la measurement en InfluxDB. Sin él, todas las métricas acaban en una measurement llamada mqtt_consumer.
• tag_keys convierte claves JSON en tags. Los tags se indexan y sirven para valores por los que filtras o agrupas, como los IDs de dispositivo. Las lecturas numéricas se convierten en fields automáticamente.
• json_time_key y json_time_format usan la marca de tiempo del dispositivo en lugar de la hora a la que Telegraf recibió el mensaje. unix_ms coincide con nuestras marcas de tiempo en milisegundos.
• topic_parsing asigna segmentos del topic a tags. Cada _ salta un nivel y los segmentos con nombre se convierten en tags, así que line se rellena con el cuarto nivel del topic.

Una nota sobre el tag topic: es práctico al depurar, pero cada topic distinto crea una nueva serie en InfluxDB. Con unos cientos de dispositivos no pasa nada. Con decenas de miles de topics, o topics que contienen IDs de mensaje o marcas de tiempo, la cardinalidad de series crece rápido y ralentiza las consultas. Cuando los tags line y device ya llevan la misma información, fijar topic_tag = "" mantiene ligera la base de datos.

Define el token como variable de entorno y ejecuta primero Telegraf en modo de prueba. Como mqtt_consumer es un input de servicio, el modo de prueba necesita un tiempo de espera para recoger mensajes:

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

Mientras se ejecuta, publica un mensaje de prueba desde otra 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}"

La llamada date +%s%3N genera una marca de tiempo en milisegundos con GNU date en Linux; en macOS, instala coreutils y usa gdate, o escribe un valor fijo reciente. Telegraf debería imprimir una línea de line protocol de InfluxDB similar a esta:

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

Cuando funcione, arranca Telegraf normalmente con telegraf --config telegraf.conf, elimina el bloque outputs.file cuando confíes en el pipeline y consulta InfluxDB para confirmar que los datos han llegado:

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

Proteger la conexión en producción

El servidor público es solo para experimentos; cualquier cosa que haya en él la pueden leer otros. Para datos reales, usa un broker privado con credenciales y TLS. Con un broker MQTT gestionado como ConnectMQ, el bloque de input cambia así:

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

El broker presenta un certificado de Let's Encrypt, de confianza en el almacén de CA del sistema operativo, así que no hace falta el ajuste tls_ca. persistent_session = true combinado con QoS 1 y un client ID fijo pide al broker que conserve la sesión de Telegraf y encole los mensajes durante reinicios breves de Telegraf. Guarda las credenciales en variables de entorno y no en el archivo de configuración, sobre todo si el archivo está en control de versiones.

Resolución de problemas del Telegraf MQTT consumer

Ejecuta Telegraf con --debug para ver los eventos de conexión y los errores del parser en tiempo real. Con payloads que tienen objetos anidados o arrays, el parser JSON heredado llega pronto a sus límites; la guía del parser json_v2 de Telegraf muestra cómo mapear con precisión estructuras anidadas y segmentos de topic. La lista completa de opciones está en la documentación oficial del plugin mqtt_consumer.

Siguientes pasos

Cuando los datos ya fluyan hacia InfluxDB, añade Grafana para dashboards y alertas. Si prefieres código a configuración, el puente de Python a InfluxDB hace el mismo trabajo en un script corto con control total sobre el mapeo. Y si aún estás eligiendo las herramientas cliente para los propios dispositivos, consulta la guía de bibliotecas cliente MQTT.

Empieza con ConnectMQ

Puedes probar toda esta configuración contra el servidor MQTT público gratuito en pocos minutos. Para datos de producción, ConnectMQ te ofrece un broker privado con TLS en el puerto 8883, MQTT 5 y un portal con explorador de topics en vivo para comprobar qué recibirá Telegraf. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito y pega las credenciales generadas en tu configuración de Telegraf. Lee más sobre cómo ejecutar un broker MQTT en la nube si estás planificando un despliegue mayor.

Preguntas frecuentes

¿Admite MQTT 5 el Telegraf MQTT consumer?

Las versiones recientes de Telegraf añaden una opción protocol para seleccionar MQTT 5. El valor por defecto es MQTT 3.1.1, suficiente para suscribirse y escribir métricas en la mayoría de configuraciones.

¿Puede una sola instancia de Telegraf suscribirse a varios topics?

Sí. Enumera varios filtros en topics, o añade varios bloques mqtt_consumer con distintos parsers cuando los topics lleven formatos de payload diferentes.

¿Por qué no se guardan mis cadenas JSON?

El parser json heredado ignora los valores de texto salvo que los incluyas en json_string_fields o tag_keys. El parser json_v2 maneja las cadenas de forma explícita.

¿Debe Telegraf usar QoS 0 o QoS 1?

QoS 0 sirve para telemetría de alta frecuencia en la que perder alguna lectura ocasional no importa. Usa QoS 1 con una sesión persistente cuando haya que guardar todas las lecturas.