Mqtt sparkplug publisher plugin

Note: This plugin requires InfluxDB 3 Core.8.2 or later, paho-mqtt>=2.1,<3, protobuf>=5 and influxdata-plugin-utils>=0.5.0.

The MQTT Sparkplug B Publisher turns InfluxDB 3 Core into a Sparkplug B Edge Node. As rows are written, it publishes them to an MQTT broker as a compliant spBv1.0 session — Protocol Buffers payloads, birth and death certificates, sequence numbers and aliases — so a SCADA system, Ignition, or any Sparkplug host application consumes the data natively.

• A real Edge Node, not a passthrough: NBIRTH announces the metric set, NDATA/DDATA carry the values, NDEATH/DDEATH report loss, bdSeq and seq keep the session ordered, and aliases keep each message compact
• Devices from your data: publish at node level, one device per source table, or one device per distinct tag value
• Store-and-forward: a broker outage buffers rows in memory under a byte budget and a drop policy, then replays them with is_historical so old data never overwrites the host’s live value
• Survives across writes: one background sender owns the connection, the sequence numbers and the buffer, so a slow or down broker never blocks writes
• Rebirth, STATE and takeover handled: answers a host’s rebirth NCMD, gates on a primary host’s STATE, and detects an orphan takeover

Contents

• How it works — the Edge Node session and the background sender
• Deployment: one session per node — read this first
• Devices and metrics — how rows become Sparkplug metrics
• Store-and-forward — the buffer, historical data and draining
• Rebirth, STATE and takeover
• Setup · Examples · Parameters
• Where settings come from · Installation
• Observability · Troubleshooting

How it works

The Edge Node session

The plugin owns one Sparkplug Edge Node, identified by sparkplug_group_id and sparkplug_edge_node_id. Every topic is spBv1.0/<group>/<type>/<edge_node>, with a device appended for device-level messages:

Each connect carries a fresh bdSeq (matching the will’s) and starts NBIRTH at seq=0; every message after it increments seq, wrapping at 255. Each metric is declared once in a birth with a name and an alias, and after that every NDATA/DDATA carries only the alias, which keeps messages small.

The background sender

Publishing does not happen on the write path. A write validates the configuration, turns the batch into compact buffered entries, and returns immediately. A background sender owns the MQTT connection, the sequence numbers and the buffer: it connects, publishes the births, drains the buffer, answers rebirths and runs the watchdog. A broker that is slow or down therefore never blocks writes into InfluxDB.

If no write arrives for idle_disconnect, the sender publishes NDEATH and shuts the connection down, so a deleted or idle trigger cannot leave an orphan session alive until the server restarts.

Deployment: one session per node

A Sparkplug Edge Node is a single identity; two live sessions for one group/node pair fight over the broker. Deploy accordingly:

• Create the trigger WITHOUT --run-asynchronous. A synchronous trigger ensures exactly one sender is ever built.
• Run exactly one process node for this trigger. On a multi-node cluster the node specification picks only the owner, but runs are placed round-robin across every process node, so two process nodes would stand up two sessions. A single --mode all node (the Docker default) or one standalone --mode process node is the supported shape.
• Keep client_id stable across trigger recreations (the default already is, derived from the group and node). The broker then evicts an orphan of the same identity by takeover.

Devices and metrics

sparkplug_device_from decides how a row maps to a Sparkplug device:

payload_fields is the declared metric set of the birth certificate and is mandatory. Each entry is a column, optionally name:type to override the datatype inferred from the schema; supported types are int8/16/32/64, uint8/16/32/64, float, double, boolean, string and datetime. Signed integers are encoded as two’s-complement in Sparkplug’s unsigned fields.

Sparkplug datetime is milliseconds since the epoch, but a source column may hold another unit, so a datetime entry may name it: name:datetime:unit with unit one of s, ms (default), us or ns — the value is converted to milliseconds. For example event_ts:datetime:ns reads a nanosecond column.

A column written but not in payload_fields is not published (and is logged once at startup). A tag that is neither the device dimension nor pinned to a single value in allow_values is refused at startup, because its series would otherwise collapse onto one metric name.

Store-and-forward

Nothing is handed to a disconnected client. While the broker is unreachable, entries accumulate in the buffer up to max_queue_bytes, measured by their in-memory size. At the cap, drop_policy decides what goes: drop_oldest (default) keeps the freshest data, drop_newest the earliest.

Historical data. A row whose own timestamp is older than now − live_window is published with is_historical = true rather than as the node’s current value, so a backfill, late data or a WAL replay after a restart never rolls the host’s live value back or re-fires an alarm. sparkplug_replay_historical = false discards those rows instead of replaying them. Rows merely delayed by back-pressure are seconds old and stay live. live_window = 0 turns the test off.

Drain order. fifo (default) drains strictly in order, so after a long outage live data waits behind the backlog. live_first publishes recent rows first and drains the historical ones with the leftover rate budget, so current values stay prompt.

Report by exception. With report_by_exception = true a metric is published only when its value changed from the last published value — the Sparkplug-recommended mode for periodically polled sources. deadband suppresses a numeric change smaller than itself.

After a reconnect the buffered data is replayed on the new session with correct, continuous sequence numbers.

Rebirth, STATE and takeover

• Rebirth. A host publishes Node Control/Rebirth on NCMD; the sender answers with the full NBIRTH + DBIRTH sequence, coalescing a burst of requests into one. A rebirth is held while a configured primary host is offline, just like the first birth.
• Primary host STATE. Set sparkplug_primary_host_id to subscribe to spBv1.0/STATE/<id> and hold the first NBIRTH until the host is online. On offline the node publishes NDEATH, stops cleanly, and the next write rebuilds it — reconnecting and again waiting for the host to come back online.
• Takeover. With MQTT 5 the 0x8E reason code is exact; on 3.1.1 a run of quick post-connect closes (takeover_attempts) is read as another client taking the identity, after which this client stops and cools down. paho-mqtt 2.1 leaves a reason-only DISCONNECT undecoded (mosquitto sends e0 01 8e), so the plugin reads the reason byte itself; if it cannot, MQTT 5 falls back to the 3.1.1 heuristic.

Setup

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/mqtt_sparkplug_publisher/mqtt_sparkplug_publisher.py \
  --trigger-spec "all_tables" \
  --trigger-arguments 'broker_url=mqtt://broker:1883,sparkplug_group_id=plant1,sparkplug_edge_node_id=influxdb3,source_measurements=press_line,payload_fields=temp pressure' \
  spb_press
influxdb3 enable trigger --database mydb spb_press

Batches for any table other than source_measurements are ignored, so all_tables is a safe trigger specification and keeps a sparse source table from idling the client out.

Examples

Source row used throughout:

press_line,machine=press7 temp=21.5,pressure=4.2 1791019630123456789

Node-level metrics. No device; metric names are table/field:

--trigger-arguments 'broker_url=mqtt://broker:1883,sparkplug_group_id=plant1,sparkplug_edge_node_id=influxdb3,source_measurements=press_line,payload_fields=temp pressure'

spBv1.0/plant1/NBIRTH/influxdb3   bdSeq, Node Control/Rebirth, press_line/temp, press_line/pressure
spBv1.0/plant1/NDATA/influxdb3    press_line/temp=21.5, press_line/pressure=4.2   (is_historical if old)

A device per machine. One source table, device from the machine tag; metric names are the bare field names:

--trigger-arguments 'broker_url=mqtt://broker:1883,sparkplug_group_id=plant1,sparkplug_edge_node_id=influxdb3,source_measurements=press_line,payload_fields=temp pressure,sparkplug_device_from=tag:machine'

spBv1.0/plant1/DBIRTH/influxdb3/press7   temp, pressure
spBv1.0/plant1/DDATA/influxdb3/press7    temp=21.5, pressure=4.2

Publish only changed values, with a deadband. Typed metric and filters contain commas, so this one goes in a TOML file:

broker_url = "mqtts://broker:8883"
sparkplug_group_id = "plant1"
sparkplug_edge_node_id = "influxdb3"
source_measurements = ["press_line"]
payload_fields = ["temp", "pressure", "count:int64"]
report_by_exception = true
deadband = 0.5
filters = [["temp", ">=", 185]]
tls_ca_cert = "certs/ca.pem"

--trigger-arguments 'config_file_path=mqtt_sparkplug_publisher_config.toml'

The trigger CLI splits on commas, so any value containing a comma — a typed metric list, filters, allow_values — belongs in the TOML file, which is also the right place for credentials.

Parameters

Required

Connection

Sparkplug

TLS (for mqtts:// and wss://; a relative path resolves under PLUGIN_DIR)

WebSocket (for ws:// and wss://)

Row selection

Delivery and buffering

Common

Where settings come from

Every setting may come from three layers, applied in order, each winning over the one before:

1. Environment — INFLUXDB3_MQTT_SPARKPLUG_PUBLISHER_<SETTING> (uppercase).
2. Trigger arguments — --trigger-arguments 'key=value,key=value'.
3. TOML file — config_file_path, read from PLUGIN_DIR.

A list is space-separated in an argument (payload_fields=temp pressure) and a native array in TOML (payload_fields = ["temp", "pressure"]). Because the CLI splits arguments on commas, put any value containing a comma — and all credentials — in the TOML file. See mqtt_sparkplug_publisher_config.toml for a documented template.

Installation

• InfluxDB 3 Core: 3.8.2 or later, with the Processing Engine enabled
• Python packages: influxdata-plugin-utils>=0.5.0, paho-mqtt>=2.1,<3, protobuf>=5

One virtual environment serves every plugin on a node, so a package pinned by one plugin affects the others. This plugin requires paho-mqtt 2.x and runs alongside the other MQTT plugins without conflict. Its Sparkplug codec is built from a descriptor embedded in the plugin rather than a generated module or a library such as pysparkplug, so the plugin ships as a single file and pins no protobuf version — it will not downgrade packages the other plugins rely on.

1. Start InfluxDB 3 Core with the Processing Engine enabled:

   influxdb3 serve \
     --node-id node0 \
     --object-store file \
     --data-dir ~/.influxdb3 \
     --plugin-dir ~/.plugins

2. Install the packages:

   influxdb3 install package influxdata-plugin-utils
   influxdb3 install package paho-mqtt
   influxdb3 install package protobuf

Observability

Counters reach the log as a summary line and the mqtt_sparkplug_publisher_stats table in the trigger’s own database, at most once per stats_interval.

Rows are tagged with broker (host:port), group and edge_node.

“Published” is not a delivery count. At QoS 0 it means the message reached the socket, nothing about the broker having received it. A low keep_alive shortens the window in which a silently dropped link counts QoS 0 data as published while it is actually lost.

Troubleshooting

The host sees no data, but the counters stay green

A broker ACL may be dropping the QoS 0 publishes silently. Subscribe to the node’s own birth topic — mosquitto_sub -t 'spBv1.0/<group>/NBIRTH/<node>' — to confirm delivery. MQTT 5 surfaces an authorization refusal as a reason code; on 3.1.1 it is silent.

Two sessions keep evicting each other

Two live senders share one group/node identity. Make sure the trigger runs on exactly one process node and was created without --run-asynchronous (see Deployment). A stable client_id lets the broker evict the orphan cleanly rather than flapping.

Old data overwrites the host’s live value

live_window is too short or 0. A backfill or a WAL replay carries old timestamps; with a live_window that covers your backfill horizon those rows go out is_historical = true and the host keeps its current value. Set sparkplug_replay_historical = false to drop them entirely instead.

A device never disappears after it stops reporting

sparkplug_device_timeout defaults to 0, which never publishes DDEATH because an idle device cannot be told from a removed one. Set it to a duration longer than the device’s normal reporting gap.

The client stops and will not reconnect for a while

A refused birth (packet too large or not authorized) cools down for five minutes; a detected takeover cools down for ten seconds, before the next invocation rebuilds the client. The log line names which. Fix the underlying cause (payload size, ACL, or the duplicate session) rather than recreating the trigger in a loop.

A trigger created before its source table exists

That is expected. The schema cannot be read until the table is written, so the plugin logs that the table is not found yet and starts without its metrics; they join the birth automatically on the first write to that table, with a rebirth. Other source tables that do exist are unaffected.

The host still shows an old node after I renamed it

Changing sparkplug_group_id, sparkplug_edge_node_id or broker_url publishes NDEATH on the old topics before switching, so the host marks the old node dead. Changing only a non-identity setting (password, TLS, keep-alive) keeps the same identity and corrects the host with a fresh NBIRTH instead.

A low keep_alive costs more PINGREQ

That is the trade-off: a lower keep_alive shortens the window in which a silently dropped link loses QoS 0 data counted as published, at the cost of more keep-alive traffic. 15–30s is a reasonable floor on an unreliable link.

Report an issue

For plugin issues, see the Plugins repository issues page.

Logging

Logs are stored in the _internal database (or the database where the trigger is created) in the system.processing_engine_logs table. To view logs:

influxdb3 query --database _internal "SELECT * FROM system.processing_engine_logs WHERE trigger_name = 'your_trigger_name'"

Log columns:

• event_time: Timestamp of the log event
• trigger_name: Name of the trigger that generated the log
• log_level: Severity level (INFO, WARN, ERROR)
• log_text: Message describing the action or error

Report an issue

For plugin issues, see the Plugins repository issues page.

Find support for InfluxDB 3 Core

The InfluxDB Discord server is the best place to find support for InfluxDB 3 Core and InfluxDB 3 Enterprise. For other InfluxDB versions, see the Support and feedback options.