mqqt-scrubber

Small Go daemon for collecting MQTT topics, normalizing mostly Tasmota payloads, and writing them into InfluxDB v3.

Current status

This repository now contains:

• a milestone work plan in docs/WORKPLAN.md
• a technical design in docs/ARCHITECTURE.md
• an initial runnable scaffold for the background service

Scope for v1

• subscribe to MQTT topics, primarily tele/+/SENSOR and tele/+/STATE
• parse Tasmota JSON payloads into normalized records
• batch writes to InfluxDB v3 using the HTTP write_lp endpoint
• run as a long-lived background process with reconnect and graceful shutdown

Project layout

cmd/mqqt-scrubber/      application entrypoint
internal/config/        config loading and validation
internal/influx/        InfluxDB v3 HTTP writer
internal/model/         internal message and record types
internal/mqtt/          MQTT subscriber wrapper
internal/parser/        Tasmota payload parsing and flattening
internal/pipeline/      batching, flushing, and service orchestration
docs/                   work plan and architecture notes

Configuration

The service loads configuration from a JSON file passed with -config and then applies environment variable overrides.

Start with:

cp config.example.json config.json

Supported environment variables:

• MQTT_SCRUBBER_MQTT_BROKER
• MQTT_SCRUBBER_MQTT_USERNAME
• MQTT_SCRUBBER_MQTT_PASSWORD
• MQTT_SCRUBBER_MQTT_CLIENT_ID
• MQTT_SCRUBBER_MQTT_TOPICS
• MQTT_SCRUBBER_MQTT_QOS
• MQTT_SCRUBBER_INFLUX_URL
• MQTT_SCRUBBER_INFLUX_DATABASE
• MQTT_SCRUBBER_INFLUX_TOKEN
• MQTT_SCRUBBER_INFLUX_PRECISION
• MQTT_SCRUBBER_DEVICE_ALIASES as a JSON object such as {"kitchen_plug":"Kitchen Plug"}
• MQTT_SCRUBBER_APP_BATCH_SIZE
• MQTT_SCRUBBER_APP_BUFFER_SIZE
• MQTT_SCRUBBER_APP_FLUSH_INTERVAL
• MQTT_SCRUBBER_APP_FLUSH_TIMEOUT
• MQTT_SCRUBBER_APP_LOG_LEVEL
• MQTT_SCRUBBER_APP_HEALTH_ADDRESS

MQTT_SCRUBBER_MQTT_TOPICS expects a comma-separated list.

You can also define optional per-device aliases in config with a top-level device_aliases object. Keys are normalized like device tags, so kitchen-plug, Kitchen Plug, and kitchen_plug all resolve to kitchen_plug.

Run

go run ./cmd/mqqt-scrubber -config config.json

The process also exposes simple health endpoints when health_address is set:

• /healthz
• /readyz
• /metrics

Build

go build ./cmd/mqqt-scrubber

Docker

Build the runtime image:

docker build -t mqqt-scrubber .

Run it with your config mounted in:

docker run --rm \
	-p 8080:8080 \
	-v "$PWD/config.json:/app/config.json:ro" \
	mqqt-scrubber

The container health check uses http://127.0.0.1:8080/healthz.

Docker Compose

Bring the service up with the included compose file:

docker-compose up --build -d

It expects a local config.json next to the compose file and exposes port 8080 for health and metrics.

Grafana

A simple starter dashboard for the current Tasmota schema is included at docs/grafana/tasmota-simple-dashboard.json.

Assumptions:

• Grafana uses the built-in InfluxDB datasource plugin
• the datasource is configured for InfluxDB v3 with query language set to SQL
• the datasource points at the same database name configured in MQTT_SCRUBBER_INFLUX_DATABASE

Import the dashboard JSON and bind the DS_INFLUXDB input to your InfluxDB datasource.

The dashboard includes a Device variable for the selected-device section and a Relay Device variable that only lists devices publishing relay fields.

Use Device for the health row, time series, and recent-state table. Use Relay Device for the relay status row so non-relay devices do not clutter that picker.

The dashboard currently visualizes fields that are known to be emitted by the parser today and that exist with the default topic subscription set:

• tasmota_sensor: energy_power, energy_voltage, energy_total, analog_temperature, analog_a0
• tasmota_state: wifi_signal, uptime_sec

If you also want LWT availability panels, add tele/+/LWT to the configured MQTT topics so the tasmota_lwt measurement is created.

Relay panels use the latest power, power1, power2, power3, and power4 values from tasmota_state. Devices that do not publish a given relay field will show no value for that panel.

The dashboard is split into a fleet summary section and a selected-device section. The selected-device section also includes Last Seen, Seconds Since Last Message, and Messages In Range panels derived from both tasmota_state and tasmota_sensor timestamps.

If device_aliases is configured, the summary table will expose an alias column populated from the latest state or sensor record for each device.

Notes

• The repo name is kept as mqqt-scrubber to match the existing folder.
• The current parser is intentionally narrow and optimized for Tasmota telemetry first.
• Writes use line protocol over HTTP against InfluxDB v3 /api/v3/write_lp.
• The Docker image includes a non-root runtime user and a simple HTTP health endpoint.
• Runtime counters are exposed in Prometheus-style text format at /metrics.