MQTT

Synopsis

Director connects to an MQTT broker as a client, subscribes to the topic filters it is given, and writes every message the broker sends it: the sensors of a plant, the telemetry of a fleet of devices, the events of a smart building, anything published to a broker. Nothing is installed on the broker or beside it.

One device, one broker, one connection. Collecting from four brokers takes four devices. A broker cluster is reached by the address that fronts it.

The device dials the broker and binds no port, so any number of MQTT devices can name port 1883, and none of them blocks a listener. One Director owns each device: a plain subscription is delivered to every subscriber, so a device run from every node of a cluster would write every message once per node.

A message is written as it came. Its payload is the record and its topic is the record's request, dated when Director received it. Turning the payload into fields is the work of the device's pipeline. See Messages.

MQTT 5, retained message handling, and last will are not part of the device. To publish to a broker instead, see MQTT target. A broker is often both a source and a destination.

note

The MQTT device needs Director 2.3.0 or later. With no Director or cluster at that version selected, the wizard shows MQTT devices are unavailable on directors or clusters below version 2.3.0 and the device cannot be created.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: mqtt
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    transport: <string>
    path: <string>
    username: <string>
    password: <string>
    topics: <string[]>
    qos: <numeric>
    persistent_session: <boolean>
    client_id: <string>
    keep_alive: <numeric>
    ping_timeout: <numeric>
    connect_timeout: <numeric>
    max_reconnect_interval: <numeric>
    max_buffer_size: <string>
    tls:
      status: <boolean>
      insecure_skip_verify: <boolean>
      server_name: <string>
      ca_name: <string>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      min_tls_version: <string>
      max_tls_version: <string>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
`id`	Y	-	Unique numeric identifier
`name`	Y	-	Device name
`description`	N	-	Optional description
`type`	Y	-	Must be `mqtt`
`tags`	N	-	Optional tags
`pipelines`	N	-	Optional pre-processor pipelines
`status`	N	`true`	Enable/disable the device

Connection

Field	Required	Default	Description
`address`	Y	-	Host name or IP address of the broker. An IPv6 address can be written as it is or in brackets. A wildcard address (`0.0.0.0`, `::`) is refused, and so is a port or a path written into the address: each is its own field.
`port`	N	By `transport` and `tls.status`	The broker's port, from `1` to `65535`. Left out, it is `1883` for `tcp`, or `8883` for `tcp` with TLS; for `websocket` it is `80`, or `443` with TLS.
`transport`	N	`tcp`	`tcp`, or `websocket` for a broker behind an HTTP proxy or a cloud service
`path`	N	`/mqtt`	The WebSocket's URL path. Refused with the `tcp` transport.
`keep_alive`	N	`60`	Seconds between the pings that detect a dead connection, from `1` to `65535`
`ping_timeout`	N	`10`	Seconds a ping waits for its answer, from `1` to `300`. A connection with no answer for `keep_alive` plus `ping_timeout` is taken for lost.
`connect_timeout`	N	`30`	Seconds a connect and a subscription wait for the broker, from `1` to `300`
`max_reconnect_interval`	N	`30`	The wait between attempts to open a lost connection again grows from 1 second to this many seconds, from `1` to `3600`

The device connects to tcp://, ssl://, ws://, or wss:// as transport and tls.status decide.

Subscription

Field	Required	Default	Description
`topics`	Y	-	The topic filters to subscribe to, as a list (a single filter can be written as text). At most 1000, each a valid MQTT 3.1.1 filter, none listed twice. `+` matches one level of the topic, and `#` matches the rest of it and is only ever the last level.
`qos`	N	`0`	`0`, `1`, or `2`: the QoS every filter is subscribed at. The broker may grant a lower one.
`persistent_session`	N	`false`	Keep the session at the broker while the device is away, and acknowledge a message only once it was delivered. See Sessions and Acknowledgements.
`client_id`	N	`vm-<device id>`	The identifier the broker knows the client by, which names its session. It must be the same at every start, and no other client may use it: a second connection with the same identifier ends the first. At most 256 characters.

A filter written as $share/<group>/<filter> is a shared subscription: the broker delivers each message to one consumer of the group, so several consumers, such as this device on another Director or an application, share a topic. The group is a name of your own. Shared subscriptions are a broker feature (MQTT 5, and a number of 3.1.1 brokers), and the device is still run from one Director.

Authentication

Field	Required	Default	Description
`username`	N	-	Plain text, `${ENV_VAR}`, or `$secret{...}`
`password`	N	-	Plain text, `${ENV_VAR}`, or `$secret{...}`. A password needs a username: MQTT sends one only with the other.

The credentials are read at every connect, not when the configuration is read: at the first connect, and at every one after a lost connection. A password rotated in the secret store is the one the next connect sends, with no restart. A variable that is not set, or a secret that cannot be read, is an Authentication error at the first connect, never a login with nothing. At a reconnect the credentials read last are sent instead, and the log says so.

TLS

Field	Required	Default	Description
`tls.status`	N	`false`	Connect over TLS: `ssl://`, or `wss://` over a WebSocket
`tls.insecure_skip_verify`	N	`false`	Skip broker certificate verification. The device logs a warning when it starts with this on.
`tls.server_name`	N	-	The name the broker certificate is checked against, when it is not the address
`tls.ca_name`	N	-	The authority of the broker certificate, as inline PEM or a file under the Director's root. When unset, the host trust store is used.
`tls.cert_name`	N*	-	Client certificate, for a broker that identifies the client by it
`tls.key_name`	N*	-	Client private key
`tls.passphrase`	N	-	Passphrase for an encrypted private key (PKCS#8, or legacy PEM encryption)
`tls.min_tls_version`	N	`tls1.2`	Minimum accepted TLS version (`tls1.0`, `tls1.1`, `tls1.2`, `tls1.3`). Any other value is refused.
`tls.max_tls_version`	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.

* tls.cert_name and tls.key_name are set together or not at all.

A client certificate and key that are files under the Director's root are read again as they are rotated: the connect that follows a change of the files uses the new pair, with no restart.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
Inline PEM content — used verbatim when the value contains -----BEGIN.
Environment variable — ${ENV_VAR}.
Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Performance

Field	Required	Default	Description
`max_buffer_size`	N	`128MB`	Ingest buffer admission cap. Also sizes this listener's ingest buffers: 3/8 of this value at start, growing to at most 3x

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

A configuration that breaks any of these rules does not start, and the device's state is Configuration error; the message names the field.

Connection

The device connects to the broker, then subscribes to its topic filters. It is Connected once the broker has answered the subscription.

A first connect that fails stops the collector, with a connection state that says why, and the device manager starts it again after a wait that grows:

The broker...	State
cannot be reached, or answers nothing within `connect_timeout`	Connection error
refuses the user: a bad user name or password, or a user that is not allowed	Authentication error
takes none of the subscriptions, because the user may not subscribe to any of them	Authentication error
refuses the client: the protocol version (the device speaks MQTT 3.1.1), or the `client_id`	Configuration error
presents a certificate the device does not trust, or one that names another host	Certificate error

A connection that is lost is opened again by the client, without restarting the collector. The wait before each attempt starts at 1 second and doubles up to max_reconnect_interval, and the subscriptions are made again on the new connection. The state is Connection error from the loss until the subscription that follows. A subscription the broker does not answer in time is sent again, after the same growing wait, for as long as the connection lasts.

Some filters refused. A broker refuses the filters the user may not subscribe to and accepts the rest. The device reads the filters it was granted and names the refused ones in its log, once per connect. It is an error only when none was granted.

On a stop the device closes its connection, waits for what is in flight, and flushes its queue one last time; only then is it stopped. A restarted device therefore never has its predecessor still writing beside it, or still holding its client identifier. A stop that takes longer than 30 seconds is logged as a warning, and the device waits for it all the same.

Log Warnings

The device logs a warning when it starts with any of these:

A username and password sent without TLS, so they cross the network as they are.
tls.insecure_skip_verify on, so the broker certificate is not checked.
persistent_session on at qos: 0, where the broker keeps nothing for the session.
qos: 2, which is handled as QoS 1.
qos of 1 or 2 without persistent_session, where a message is acknowledged before it is delivered.

Messages

Each message is written as one record, unchanged:

Part of the record	Value
`message`	The payload, as it came
`_vmetric.event.request`	The topic the message was published to
The timestamp	When Director received the message. MQTT carries no time of its own.

A message with no payload is acknowledged and not written: MQTT clears a retained message with one, and it carries nothing to collect. The QoS, retain, and duplicate flags of a message are not written.

A message the broker sends once is written once, whichever of the device's filters it matched. Whether the broker sends a message that matches two overlapping filters once or twice is up to the broker.

Records are written under definition ID 848, as raw records of the custom log type, like those of the other message broker devices.

Parsing in the Pipeline

The device's pre-processing pipeline turns a message into fields. Parse the payload with json for the JSON most devices publish, kv, csv, or xml for those formats, and grok or dissect for text of any other shape. Take the topic's path apart with dissect on _vmetric.event.request:

processors:
  # A payload of {"celsius":21.5}, on plant/line1/press/state
  - json:
      field: message
      add_to_root: true
  # The parts of the topic's path, as fields
  - dissect:
      field: _vmetric.event.request
      pattern: "plant/%{line}/%{machine}/%{topic_kind}"

Backpressure

Messages are handed to the ingest stream one at a time, in the order they arrived, and the stream admits them only while its buffer (max_buffer_size) has room. A broker that sends faster than the pipeline delivers is slowed by the connection rather than overrunning the device.

A client held up for longer than keep_alive plus ping_timeout is taken for dead by the broker, which closes the connection. The client then opens it again, and a persistent session gets what it did not acknowledge sent again.

Sessions and Acknowledgements

MQTT acknowledges QoS 1 and 2 messages to the broker, which sends again what it is not told was received. persistent_session decides when the device tells it:

`persistent_session`	The broker keeps	A message is acknowledged
`false` (default)	Nothing while the client is away: the session ends with the connection	As the device takes it, before it is written anywhere. A stop loses what was taken and not yet delivered.
`true`	The session, with its subscriptions and the QoS 1 and 2 messages the client missed	Once the sender has confirmed delivery of the payload the message was written to, after the pipeline. A stop, or a target that is down, leaves the message with the broker, which sends it again when the device returns.

persistent_session with qos: 1 is at-least-once, the choice for data that must not be lost. A clean session at qos: 0 is at-most-once, the choice for telemetry that is sent again a moment later: it holds nothing at the broker, and nothing waits for a delivery.

A persistent session has these costs:

A message can be delivered twice. A message that was delivered, but whose acknowledgement was lost with the connection, is sent again, and the broker marks it as a duplicate.
The broker's in-flight window bounds throughput. A broker sends a client only a limited number of unacknowledged messages at a time: Mosquitto's max_inflight_messages is 20 by default, and EMQX's is 32. The device acknowledges a message once its delivery is confirmed, so a busy topic is bounded by that window divided by the delivery delay. Raise the window at the broker for a topic that carries thousands of messages a second.
Subscriptions stay with the session at the broker. A filter removed from topics is still subscribed there, and its messages are still read and acknowledged, until the session is cleared or client_id changes.
QoS 2 is handled as QoS 1. A message is acknowledged once delivered, and one sent again after a lost connection is delivered again. The device does not promise exactly-once delivery.
client_id must be stable and unique. A different identifier at the next start does not find the session again, and two clients sharing one end each other's connections. A device moved to another Director keeps its identifier, and so its session.

note

A persistent session at qos: 0 keeps nothing: brokers do not hold QoS 0 messages for a client that is away.

Device State

The connection state follows the broker:

State	When
Connected	The broker took the subscriptions, on the first connect and on every one after a loss
Connection error	The broker cannot be reached, the connection was lost, or a subscription was not answered
Certificate error	The broker certificate is not trusted, or names another host
Authentication error	The broker refused the user, took none of the subscriptions, or the credentials could not be read
Configuration error	The configuration cannot run, or the broker refused the client identifier or the protocol version

Examples

Open Broker

Reading an open broker on a private network at QoS 0, as fast as it sends...

- id: 512400300
  name: factory-broker
  type: mqtt
  properties:
    address: "192.0.2.60"
    port: 1883
    topics:
      - factory/+/+/state
      - factory/alarms/#
    qos: 0

Authenticated, TLS, and Persistent Session

Connecting with a user over TLS, with a persistent session at QoS 1 so that nothing the broker holds is lost while the device is stopped or a target is down...

- id: 512400301
  name: plant-broker
  type: mqtt
  properties:
    address: "mqtt.plant.example.com"
    port: 8883
    tls:
      status: true
      server_name: mqtt.plant.example.com
      ca_name: plant-ca.pem
      min_tls_version: tls1.2
    username: director
    password: ${MQTT_PASSWORD}
    persistent_session: true
    client_id: plant-director-1
    qos: 1
    topics:
      - plant/#

Cloud Broker over WebSocket

Reaching a cloud broker over a secure WebSocket with a client certificate, sharing one topic with other consumers through a shared subscription...

- id: 512400302
  name: cloud-broker
  type: mqtt
  properties:
    address: "iot.example.com"
    transport: websocket
    path: /mqtt
    port: 443
    tls:
      status: true
      server_name: iot.example.com
      ca_name: cloud-ca.pem
      cert_name: cloud-client.pem
      key_name: cloud-client.key
      passphrase: ${MQTT_KEY_PASSPHRASE}
    persistent_session: true
    client_id: cloud-director-1
    qos: 1
    topics:
      - $share/vmetric/devices/+/telemetry
      - devices/+/alarms

Pipelines

Parsing JSON payloads and splitting the topic into fields with a pre-processing pipeline...

- id: 512400303
  name: plant-telemetry
  type: mqtt
  pipelines:
    - mqtt_plant_parser
  properties:
    address: "mqtt.plant.example.com"
    topics:
      - plant/+/+/+

note

Pipelines are processed sequentially, and can modify or drop messages before ingestion.

Synopsis​

Schema​

Configuration​

Device​

Connection​

Subscription​

Authentication​

TLS​

Performance​

Connection​

Log Warnings​

Messages​

Parsing in the Pipeline​

Backpressure​

Sessions and Acknowledgements​

Device State​

Examples​

Open Broker​

Authenticated, TLS, and Persistent Session​

Cloud Broker over WebSocket​

Pipelines​

Synopsis

Schema

Configuration

Device

Connection

Subscription

Authentication

TLS

Performance

Connection

Log Warnings

Messages

Parsing in the Pipeline

Backpressure

Sessions and Acknowledgements

Device State

Examples

Open Broker

Authenticated, TLS, and Persistent Session

Cloud Broker over WebSocket

Pipelines