Recoger datos IoT por MQTT con Telegraf e InfluxDB 2

InfluxDB es una base de datos de series temporales pensada para guardar lecturas con marca de tiempo, como la temperatura o la humedad que envían los sensores IoT. Telegraf es el agente de recogida de InfluxData: se suscribe a un broker MQTT, interpreta los mensajes y los escribe en InfluxDB. En este tutorial montarás en Ubuntu 24.04 un broker Mosquitto, InfluxDB 2 y Telegraf, enviarás lecturas de prueba por MQTT y configurarás la retención y una tarea que resume los datos por horas.

Requisitos previos

Para seguir esta guía necesitas:

Un servidor con Ubuntu 24.04 LTS, por ejemplo un VPS de CubePath, con al menos 2 GB de RAM y 20 GB libres en disco.
Un usuario no root con privilegios sudo.
UFW activo con SSH permitido.
Sensores o un script capaz de publicar mensajes MQTT en formato JSON. En la guía se simulan con mosquitto_pub.

Nota

esta guía usa InfluxDB 2.x (paquete influxdb2), que sigue disponible en el repositorio oficial y usa el lenguaje Flux. InfluxDB 3 es un producto distinto con otra CLI y otra forma de configuración.

Los ejemplos usan la organización iot, el bucket sensores para datos en bruto y sensores-horario para los resúmenes. Cámbialos si lo prefieres, pero hazlo en todos los pasos.

Paso 1: Instalar y proteger Mosquitto

Telegraf necesita un broker MQTT del que leer. Instala Mosquitto y sus clientes desde los repositorios de Ubuntu:

sudo apt update
sudo apt install mosquitto mosquitto-clients

Mosquitto 2 solo acepta conexiones locales si no se configura un listener. Crea un usuario para los sensores y otro para Telegraf. La opción -c crea el fichero, así que úsala solo la primera vez:

sudo mosquitto_passwd -c /etc/mosquitto/passwd sensores
sudo mosquitto_passwd /etc/mosquitto/passwd telegraf
sudo chown mosquitto:mosquitto /etc/mosquitto/passwd
sudo chmod 600 /etc/mosquitto/passwd

Cada comando te pedirá una contraseña. Anota la del usuario telegraf, la usarás en el paso 5.

Crea un fichero de configuración para el listener con autenticación:

sudo nano /etc/mosquitto/conf.d/iot.conf

listener 1883
allow_anonymous false
password_file /etc/mosquitto/passwd

Reinicia el servicio y comprueba que escucha en el puerto 1883:

sudo systemctl restart mosquitto
sudo ss -tlnp | grep 1883

LISTEN 0      100          0.0.0.0:1883      0.0.0.0:*    users:(("mosquitto",pid=2143,fd=5))
LISTEN 0      100             [::]:1883         [::]:*    users:(("mosquitto",pid=2143,fd=6))

Si los sensores están fuera del servidor, abre el puerto solo para su red. Sustituye 203.0.113.0/24 por el rango de tus dispositivos:

sudo ufw allow from 203.0.113.0/24 to any port 1883 proto tcp

Advertencia

MQTT en el puerto 1883 viaja sin cifrar. Si los sensores envían datos por Internet, usa una VPN o configura un listener TLS en el puerto 8883.

Paso 2: Añadir el repositorio de InfluxData

InfluxDB 2 y Telegraf se instalan desde el repositorio oficial de InfluxData. Descarga su clave, conviértela al formato binario y guárdala en /etc/apt/keyrings:

sudo install -m 0755 -d /etc/apt/keyrings
curl -fsSL https://repos.influxdata.com/influxdata-archive.key | sudo gpg --dearmor -o /etc/apt/keyrings/influxdata-archive.gpg

Añade el repositorio:

echo "deb [signed-by=/etc/apt/keyrings/influxdata-archive.gpg] https://repos.influxdata.com/debian stable main" | sudo tee /etc/apt/sources.list.d/influxdata.list

Actualiza el índice y comprueba que los paquetes están disponibles:

sudo apt update
apt-cache policy influxdb2 telegraf

Ambos paquetes deben mostrar un Candidate procedente de repos.influxdata.com.

Paso 3: Instalar e inicializar InfluxDB

Instala el servidor y la CLI influx, que en InfluxDB 2 van en paquetes separados:

sudo apt install influxdb2 influxdb2-cli
sudo systemctl enable --now influxdb

Comprueba que responde en el puerto 8086:

curl -s http://localhost:8086/health

{"name":"influxdb", "message":"ready for queries and writes", "status":"pass", "checks":[], "version": "v2.7.12", "commit": "..."}

Inicializa la instancia con un usuario administrador, la organización y el bucket de datos en bruto con 30 días de retención. Sustituye your_strong_password por una contraseña segura de al menos 8 caracteres:

influx setup \
  --username admin \
  --password your_strong_password \
  --org iot \
  --bucket sensores \
  --retention 30d \
  --force

User	Organization	Bucket
admin	iot		sensores

influx setup guarda un perfil de configuración en ~/.influxdbv2/configs con el token de administrador, así que los siguientes comandos influx no necesitan --org ni --token.

Crea el bucket para los datos resumidos, con un año de retención:

influx bucket create --name sensores-horario --retention 365d

InfluxDB borra automáticamente los puntos más antiguos que la retención de cada bucket. Compruébalo:

influx bucket list

ID			Name			Retention	Shard group duration	Organization ID		Schema Type
3f1a2b4c5d6e7f80	sensores		720h0m0s	24h0m0s			9a8b7c6d5e4f3a21	implicit
7b6a5c4d3e2f1a09	sensores-horario	8760h0m0s	168h0m0s		9a8b7c6d5e4f3a21	implicit
...

Paso 4: Crear un token de escritura para Telegraf

Telegraf no debe usar el token de administrador. Crea uno que solo pueda escribir en el bucket sensores. La opción --write-bucket necesita el ID del bucket, no el nombre, así que cópialo de la salida anterior:

influx auth create \
  --description "telegraf-sensores" \
  --write-bucket 3f1a2b4c5d6e7f80

La salida incluye una columna Token con una cadena larga que termina en ==. Guárdala para el paso siguiente.

Paso 5: Instalar y configurar Telegraf

Instala Telegraf desde el mismo repositorio:

sudo apt install telegraf

El paquete trae un telegraf.conf de ejemplo con muchos plugins activos, entre ellos una salida para InfluxDB 1.x que no sirve aquí. Guárdalo como referencia y crea uno mínimo:

sudo mv /etc/telegraf/telegraf.conf /etc/telegraf/telegraf.conf.orig
sudo nano /etc/telegraf/telegraf.conf

[agent]
  interval = "10s"
  flush_interval = "10s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  omit_hostname = true

Los secretos se guardan fuera de la configuración. El servicio de Telegraf lee variables de entorno de /etc/default/telegraf, y Telegraf sustituye ${VARIABLE} en sus ficheros de configuración:

sudo nano /etc/default/telegraf

INFLUX_TOKEN=el_token_del_paso_4
MQTT_PASSWORD=la_contraseña_del_usuario_telegraf

sudo chmod 600 /etc/default/telegraf

Configura la salida hacia InfluxDB en su propio fichero dentro de /etc/telegraf/telegraf.d/, que el servicio carga automáticamente:

sudo nano /etc/telegraf/telegraf.d/salida-influxdb.conf

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "iot"
  bucket = "sensores"

Ahora la entrada MQTT. Se asume que cada sensor publica un JSON en un topic con la forma sensores/<ubicación>/<tipo>, por ejemplo sensores/salon/ambiente con {"temperatura": 22.5, "humedad": 48}:

sudo nano /etc/telegraf/telegraf.d/entrada-mqtt.conf

[[inputs.mqtt_consumer]]
  servers = ["tcp://127.0.0.1:1883"]
  topics = ["sensores/#"]
  username = "telegraf"
  password = "${MQTT_PASSWORD}"
  client_id = "telegraf-iot"
  qos = 1
  data_format = "json"
  name_override = "lectura_sensor"
  topic_tag = ""

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "sensores/+/+"
    tags = "_/ubicacion/tipo"

Qué hace cada parte:

data_format = "json" convierte cada clave numérica del JSON en un campo (temperatura, humedad).
name_override fija el nombre de la medida en InfluxDB, que por defecto sería mqtt_consumer.
topic_parsing extrae las etiquetas ubicacion y tipo de los niveles del topic; _ descarta el primer nivel. Con topic_tag = "" no se guarda además el topic completo.

Reinicia Telegraf y revisa el registro:

sudo systemctl restart telegraf
sudo journalctl -u telegraf -n 20 --no-pager

I! Loaded inputs: mqtt_consumer
I! Loaded outputs: influxdb_v2
I! [inputs.mqtt_consumer] Connected [tcp://127.0.0.1:1883]

Paso 6: Enviar lecturas de prueba y consultarlas

Publica unas lecturas simulando dos sensores. Sustituye your_sensor_password por la contraseña del usuario sensores:

mosquitto_pub -h 127.0.0.1 -u sensores -P your_sensor_password -t sensores/salon/ambiente -m '{"temperatura": 22.5, "humedad": 48}'
mosquitto_pub -h 127.0.0.1 -u sensores -P your_sensor_password -t sensores/cocina/ambiente -m '{"temperatura": 24.1, "humedad": 55}'

Espera unos 10 segundos (el flush_interval de Telegraf) y consulta los datos con Flux:

influx query 'from(bucket: "sensores")
  |> range(start: -10m)
  |> filter(fn: (r) => r._measurement == "lectura_sensor")'

Result: _result
Table: keys: [_start, _stop, _field, _measurement, tipo, ubicacion]
...  _field       _measurement    tipo      ubicacion  _value
...  temperatura  lectura_sensor  ambiente  cocina     24.1
...

Cada combinación de campo y etiquetas aparece como una tabla. Si ves tus valores, la cadena MQTT, Telegraf e InfluxDB funciona.

También puedes explorar los datos en la interfaz web de InfluxDB. El puerto 8086 no está abierto en el cortafuegos, así que accede con un túnel SSH desde tu equipo y abre http://localhost:8086:

ssh -L 8086:127.0.0.1:8086 your_user@your_server_ip

Paso 7: Resumir los datos por horas con una tarea

Con lecturas cada pocos segundos, el bucket sensores crece rápido; por eso tiene 30 días de retención. Para conservar el histórico a largo plazo, una tarea de InfluxDB calcula cada hora la media de la última hora y la escribe en sensores-horario.

Crea el fichero de la tarea:

nano ~/resumen-horario.flux

option task = {name: "resumen-horario-sensores", every: 1h}

from(bucket: "sensores")
  |> range(start: -task.every)
  |> filter(fn: (r) => r._measurement == "lectura_sensor")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> to(bucket: "sensores-horario")

Regístrala en InfluxDB:

influx task create --file ~/resumen-horario.flux

Comprueba que está activa:

influx task list

ID			Name				Organization ID		Organization	Status	Every	Cron
0c1d2e3f4a5b6c7d	resumen-horario-sensores	9a8b7c6d5e4f3a21	iot		active	1h

Tras la primera ejecución, consulta el historial con el ID de la tarea. Una ejecución correcta aparece con estado success:

influx task run list --task-id 0c1d2e3f4a5b6c7d

Solución de problemas

Telegraf muestra not Authorized al conectar con MQTT. El usuario o la contraseña no coinciden. Pruébalos directamente con el cliente:

mosquitto_sub -h 127.0.0.1 -u telegraf -P la_contraseña_del_usuario_telegraf -t 'sensores/#' -v

Si funciona aquí, revisa que MQTT_PASSWORD en /etc/default/telegraf no tenga espacios ni comillas sobrantes.

Telegraf registra 401 Unauthorized al escribir. El token es incorrecto o no tiene permiso sobre el bucket. Lista los tokens y sus permisos con influx auth list y comprueba que el de Telegraf incluye write:orgs/.../buckets/<ID de sensores>.

Telegraf registra errores de parseo JSON. El mensaje no es JSON válido o contiene solo texto. Para valores sueltos (por ejemplo 22.5) usa data_format = "value" con data_type = "float" en una entrada mqtt_consumer separada para esos topics.

La tarea falla. influx task log list --task-id <ID> muestra el error de Flux de cada ejecución, normalmente un nombre de bucket mal escrito.

Conclusión

Ya tienes una cadena completa de recogida de datos IoT: los sensores publican en Mosquitto con autenticación, Telegraf transforma los mensajes JSON en puntos etiquetados y InfluxDB los guarda con retención de 30 días y un resumen horario de un año. Como siguientes pasos, puedes conectar Grafana a InfluxDB para crear paneles, definir alertas de umbral desde la sección Alerts de la interfaz web o activar TLS en Mosquitto para los sensores remotos.

On this page