MQTT Telemetry Lab

Industrial IoT lab: simulated machines publish telemetry over MQTT, a Rust service ingests it into TimescaleDB, and Grafana shows live trends, availability / downtime / energy per machine, and statistical anomaly detection with alerting. Every piece runs as its own Docker service.

Architecture

flowchart LR
    SIM[simulator\nPython] -->|MQTT publish| MQ[Mosquitto]
    MQ -->|MQTT subscribe| ING[ingestion\nRust]
    ING -->|write| DB[(TimescaleDB)]
    DB --> GRAF[Grafana\ndashboard + alerts]
    CFG[config/machines.yaml] -.-> SIM

Loading

Component	What it does
simulator/	36 machines on 4 lines, read from config/machines.yaml. Each is a running / idle / fault state machine with drifting, noisy readings (temperature, vibration, rpm, power) and occasional fault spikes.
ingestion/	Subscribes to factory/+/+/telemetry and writes to TimescaleDB. At-least-once delivery: persistent MQTT session, ack only after the insert, duplicates ignored. Invalid data is logged and dropped; clean shutdown on SIGTERM.
storage/	Hypertable (1-day chunks, compressed after a day, 30-day retention) and two SQL functions: machine_kpis() (availability, minutes per state, kWh) and anomaly_scores() (rolling z-score).
dashboard/	Grafana provisioned as code: datasource, "Factory Overview" dashboard, alert rule on detected anomalies.

Design choices

• TimescaleDB: the data is fixed-schema numeric time series queried by time range, so plain SQL with window functions fits, and time partitioning skips the chunks outside the range. The anomaly alert query reads only the last minutes: 3.6 ms on 30 days of data (3.9M rows), against 61 s for the first version, which scored the whole table. Chunks older than a day are compressed: 7x smaller (589 MB -> 84 MB for 2 days of 36 machines).
• Rust for ingestion: a single small binary with no runtime, the piece that would sit on an edge gateway next to the machines. At this data rate Python would cope too, and a config-only tool such as Telegraf could replace it; the point here is a small consumer (an 8 MB image holding only a static binary, ~2 MiB of RAM with 36 machines, not running as root) that doesn't lose data: with the database stopped for 60 s, no sample is lost. One insert per message keeps it simple and sustains ~1,400 msg/s, 80x this plant's load (see ingestion/README.md).
• Anomaly detection independent of the fault label: the z-score never looks at the simulator's fault state, it only sees the numbers. Idle samples are left out, since a stopped machine is a known operating mode, not an anomaly. The threshold is |z| > 5, not the textbook 3: on 10 hours of simulated data it caught all 83 simulated faults with 0.1 false alarms per hour, against 42 per hour at 3, because the readings drift slowly. Those faults come from the same simulator, so the detector is also measured on real, labelled sensor data: the SKAB pump testbed, 34 experiments with one anomaly each (python3 storage/eval_skab.py). It flags 10 of 34 within 60 s of the labelled start (median delay 1 s) with 3 false alarms per hour; at |z| > 3 it would flag 17 but with 109 false alarms per hour. Most SKAB anomalies are hydraulic (valves, pressure, flow) and barely move temperature or vibration, the only two signals this detector watches, and a rolling z-score sees the onset of a long anomaly, not its whole duration (recall per sample is 0.01). Watching more signals is the obvious next step.
• Availability, not full OEE: performance and quality need part counts (good / rejected), which the simulator doesn't produce.
• No hardcoded config: machines, lines and thresholds come from config/machines.yaml, credentials from .env.

Scope

A lab, not a production deployment: the data is simulated, there is no real PLC, OPC UA or fieldbus connection, and the broker runs without auth/TLS (bound to localhost only).

Run

cp .env.example .env   # then edit .env with your own values
docker compose up -d

The first start takes a few minutes: it downloads the images and compiles the Rust service (~3 min on a laptop); later starts take seconds. Data appears in Grafana within a minute.

Service	URL / Port	Notes
Grafana	http://localhost:3000	Login with GRAFANA_ADMIN_USER/GRAFANA_ADMIN_PASSWORD from .env; dashboard "Factory Overview"
TimescaleDB	localhost:5432	Connect with any Postgres client (e.g. DBeaver) using the POSTGRES_* values from .env
Mosquitto	localhost:1883	MQTT broker, topic factory/{line}/{machine_id}/telemetry

To follow logs for a single service: docker compose logs -f simulator (or ingestion, grafana, ...).

Development

• Database schema: the scripts in storage/init/ run only on an empty data volume. From 002 on they are idempotent, so a changed one can be applied to a running database:

  docker compose exec -T timescaledb sh -c 'psql -v ON_ERROR_STOP=1 -U $POSTGRES_USER -d $POSTGRES_DB' < storage/init/003_kpis.sql

  The self-checks in storage/tests/ run the same way: each inserts known samples in a transaction, asserts the exact output, rolls back and prints ... check passed.
• Grafana is provisioned read-only: edit the files in dashboard/grafana/, not the running Grafana.

Tests

CI runs clippy and the unit tests for the Rust service, ruff and pytest for the simulator, validates the compose file, and runs the SQL self-checks against a fresh TimescaleDB with the init scripts applied.