MQTT a InfluxDB con Telegraf: guía de configuración de mqtt_consumer

El camino más corto de un broker MQTT a una base de datos de series temporales es Telegraf: su input mqtt_consumer se suscribe a tus topics, un parser convierte cada payload en una métrica y el output influxdb_v2 la escribe en InfluxDB. Sin código, con un solo archivo de configuración, y funciona sin problemas junto a la base de datos o en un gateway edge.

Esta guía monta un pipeline funcional con Docker Compose, parsea payloads JSON con json_v2, mapea partes del topic a tags y muestra cómo comprobar cada etapa antes de conectar dispositivos reales.

El pipeline de un vistazo

El dispositivo de ejemplo publica JSON como este en testmqtt/telegraf/sensor-01/telemetry:

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

Arranca InfluxDB y Telegraf

Este archivo Compose inicia InfluxDB 2.x con una organización, un bucket y un token iniciales, y Telegraf con tu configuración montada en solo lectura:

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:

Las variables de inicialización solo se aplican en el primer arranque con un volumen vacío. En producción, crea un token dedicado de solo escritura para Telegraf en lugar de reutilizar el token de administrador. Para un stack local más completo con broker y dashboards, consulta el stack MQTT con Docker Compose.

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

Los ajustes que más importan:

• servers: tcp://host:1883 sin cifrar, ssl://host:8883 para TLS. Telegraf verifica el certificado del servidor contra el almacén de CA del sistema; añade tls_ca = "/path/ca.pem" para una CA privada.
• topics: filtros MQTT estándar, así que + y # funcionan. Prueba un filtro contra nombres de topic reales con el topic matcher. Más sobre filtros: topics y wildcards.
• qos = 1 + persistent_session = true + un client_id fijo: el broker mantiene una sesión para Telegraf y encola los mensajes QoS 1 mientras se reinicia. Sin un client ID estable, la sesión no se puede reanudar. Consulta sesiones persistentes.
• max_undelivered_messages: cuántos mensajes QoS 1/2 lee Telegraf antes de que se escriban en el output. Es el backpressure de Telegraf: si InfluxDB se ralentiza, Telegraf deja de confirmar y el broker retiene el resto.
• topic_tag = "": desactiva el tag topic por defecto. Guardar el topic completo como tag multiplica la cardinalidad de series; topic_parsing extrae solo la parte que necesitas.

Cómo mapea json_v2 el payload

Con la configuración anterior, el payload de ejemplo se convierte en este line protocol:

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

Dos reglas mantienen sano InfluxDB: tags para lo que filtras y tiene pocos valores distintos, fields para las mediciones. Y envía siempre los números como números JSON: "21.4" entre comillas es un texto y se descartará o creará un conflicto de tipos en el field. Si los payloads a veces llegan mal formados, lee errores de parseo JSON en MQTT.

Si los dispositivos no envían timestamp, elimina timestamp_path y Telegraf usará la hora de llegada. Eso vale para datos en vivo, pero los datos que llegan con retraso desde dispositivos que estuvieron offline quedarán con una hora tardía.

Verifica cada etapa

Prueba el pipeline en el orden en que fluyen los datos, para saber siempre qué etapa falla.

1. ¿Están los datos en el broker? Suscríbete a testmqtt/telegraf/# con el cliente MQTT online y publica una lectura de prueba:

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. ¿Lo parsea Telegraf? Ejecuta Telegraf en modo test. Imprime las métricas en stdout en lugar de escribirlas; --test-wait mantiene el input MQTT escuchando el tiempo suficiente para recibir mensajes:

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

Vuelve a publicar mientras espera. Deberías ver la línea telemetry,device=sensor-01,.... Si no aparece ninguna línea, la suscripción o el parser están mal; si aparece con campos ausentes, las rutas o los tipos de json_v2 no coinciden.

3. ¿Está en InfluxDB? Arranca el stack con docker compose up -d y consulta el bucket:

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

Observa el topic de telemetría en tu navegador

Errores comunes

Busca los reason codes del broker con la herramienta de reason codes. Los errores del parser se registran con el prefijo E! [inputs.mqtt_consumer] y normalmente significan que el payload no es JSON válido o que un campo declarado tiene otro tipo. Ejecuta Telegraf con --debug para ver los detalles de conexión y de flush.

Escalado y retención

• Más throughput: Telegraf procesa muchos mensajes por instancia, pero si necesitas varias instancias, suscríbete con una suscripción compartida como $share/telegraf/testmqtt/telegraf/+/telemetry para que el broker reparta el flujo en lugar de duplicarlo. Cada instancia necesita su propio client_id.
• Retención: define un periodo de retención en el bucket (en la interfaz o con influx bucket update --id <bucket-id> --retention 30d) y reduce la resolución de los datos antiguos con una task de InfluxDB. Cómo planificar cuánto tiempo guardar la telemetría en bruto se explica en retención y almacenamiento de datos MQTT.
• Visualización: conecta Grafana a InfluxDB para el historial y añade el MQTT data source de Grafana si además quieres un panel en vivo directamente desde el broker.

Telegraf convierte "MQTT a InfluxDB" en una revisión de configuración en lugar de un proyecto de software. Elige bien tus tags, usa QoS 1 con sesión persistente en tu propio broker, verifica con --test antes de ponerlo en producción y guarda la configuración en control de versiones junto a tus dashboards.

Preguntas frecuentes