Skip to main content
Pitch In Subscribe

MQTT to InfluxDB and Grafana

Architecture Overview

The full telemetry pipeline runs: Meshtastic node → MQTT broker → Telegraf (or Python subscriber) → InfluxDB 2.x → Grafana dashboard. Each component is independently replaceable, so you can swap Telegraf for a custom Python script without touching InfluxDB or Grafana.

InfluxDB 2.x Setup

Install InfluxDB 2.x on a server or Raspberry Pi:

# Debian / Ubuntu / Raspberry Pi OS
wget https://dl.influxdata.com/influxdb/releases/influxdb2-2.7.1-amd64.deb
sudo dpkg -i influxdb2-2.7.1-amd64.deb
sudo systemctl enable --now influxdb

After installation, visit http://<host>:8086 to complete the setup wizard. Create an organisation (e.g. meshamerica), a bucket (e.g. meshtastic), and generate an API token with write access to that bucket. Store the token - you will need it for both Telegraf and the Python client.

Telegraf MQTT Consumer

Telegraf is the easiest path from MQTT to InfluxDB. Install it, then add a stanza to /etc/telegraf/telegraf.conf: 

[[inputs.mqtt_consumer]]
 servers = ["tcp://localhost:1883"]
 topics = ["msh/#"]
 qos = 0
 data_format = "json"
 json_string_fields = ["node_id", "node_name"]

[[outputs.influxdb_v2]]
 urls = ["http://localhost:8086"]
 token = "YOUR_INFLUXDB_TOKEN"
 org = "meshamerica"
 bucket = "meshtastic"

Telegraf will subscribe to the msh/# wildcard, parse each arriving JSON payload, and write every numeric field directly to InfluxDB. Restart Telegraf after any config change: sudo systemctl restart telegraf.

Python Subscriber (Alternative)

If you need to decode Meshtastic protobuf packets or apply custom logic, use a Python subscriber instead: 

pip install paho-mqtt meshtastic influxdb-client
import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import meshtastic.mesh_pb2 as mesh_pb2, json, os

INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = os.environ["INFLUX_TOKEN"]
INFLUX_ORG = "meshamerica"
INFLUX_BUCKET = "meshtastic"

client_db = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = client_db.write_api(write_options=SYNCHRONOUS)

def on_message(client, userdata, msg):
 try:
 payload = json.loads(msg.payload)
 node_id = payload.get("from", "unknown")
 node_name = payload.get("fromName", node_id)
 tel = payload.get("payload", {})
 point = (
 Point("telemetry")
 .tag("node_id", node_id)
 .tag("node_name", node_name)
 .field("temperature", tel.get("temperature"))
 .field("humidity", tel.get("relativeHumidity"))
 .field("battery_voltage", tel.get("voltage"))
 .field("snr", payload.get("rxSnr"))
 .field("rssi", payload.get("rxRssi"))
 )
 write_api.write(bucket=INFLUX_BUCKET, record=point)
 except Exception as e:
 print(f"Parse error: {e}")

mqttc = mqtt.Client()
mqttc.on_message = on_message
mqttc.connect("localhost", 1883)
mqttc.subscribe("msh/#")
mqttc.loop_forever()

InfluxDB Line Protocol

Whether you use Telegraf or Python, each telemetry reading lands in InfluxDB as a measurement with tags and fields: 

telemetry,node_id=!abc12345,node_name=ridge-repeater temperature=22.4,humidity=61.0,battery_voltage=3.91,snr=8.5,rssi=-87 1746230400000000000
  • Measurement: telemetry
  • Tags: node_id, node_name (indexed, low-cardinality)
  • Fields: temperature, humidity, battery_voltage, snr, rssi

Grafana Setup

Install Grafana on the same host or a separate machine, then add InfluxDB as a data source using the Flux query language:

  1. Configuration → Data Sources → Add data source → InfluxDB
  2. Set Query Language to Flux
  3. URL: http://localhost:8086
  4. Paste your InfluxDB org, bucket, and token
  • Temperature over time (line graph per node)
  • Battery voltage trends (multi-node time series)
  • Node count (stat panel - unique node_id values in last hour)
  • Signal quality map (table of last RSSI/SNR per node)

Example Flux query - temperature trend for last 24 hours:

from(bucket: "meshtastic")
 |> range(start: -24h)
 |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature")
 |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)

Retention Policies and Downsampling

In InfluxDB 2.x, retention is set per bucket. For raw telemetry data set the bucket retention to 90 days. Create a second bucket (e.g. meshtastic_hourly) with indefinite retention and use a Flux task to downsample: 

// Task: run every 1h - downsample telemetry to hourly averages
option task = { name: "downsample_telemetry", every: 1h }

from(bucket: "meshtastic")
 |> range(start: -2h)
 |> filter(fn: (r) => r._measurement == "telemetry")
 |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
 |> to(bucket: "meshtastic_hourly", org: "meshamerica")

Alerting

Grafana alerting fires when conditions are met. Useful rules for mesh deployments:

  • Low battery: battery_voltage < 3.5 for any node - triggers SMS or email alert.
  • Temperature exceeded: temperature > 50 - useful for enclosure health monitoring in summer.
  • Node offline: no telemetry from a given node_id for more than N hours - check the "Last seen" derived field using a Flux last() query against the current time.
Back to top