Skip to content

IndustryKit MQTT

An MQTT source connects to a Broker and waits for messages; it does not actively poll devices. The publisher determines the collection rate. MQTT 3.1.1, TCP/TLS, QoS 0/1, and JSON and text payloads are supported.

Configuration hierarchy

An MQTT collection instance consists of three configuration levels:

[[sources]]
  [sources.mqtt]
    [[sources.mqtt.topics]]
  • [[sources]] defines the collection instance, default measurement, and common tags.
  • [sources.mqtt] defines the Broker, session, authentication, and TLS settings.
  • Each [[sources.mqtt.topics]] defines a topic-to-metric or topic-to-logging mapping. Add multiple topic entries to subscribe to multiple topics.

The fields, tags, and topic_tags mappings under a topic may be written either as inline tables, as shown in the example, or as child tables such as [sources.mqtt.topics.fields]. A child table belongs to the immediately preceding [[sources.mqtt.topics]] entry. When type = "mqtt", only the [sources.mqtt] protocol table may be configured, and at least one topic is required. Append the complete configuration below to the main IndustryKit configuration file.

Configuration

[[sources]]
id = "mqtt-gateway"
type = "mqtt"
measurement = "mqtt_metrics"
tags = { gateway = "gw-01" }

[sources.mqtt]
brokers = ["tcp://192.0.2.30:1883"]
client_id = "industrykit-plant-a"
connect_timeout = "10s"
keep_alive = "30s"

[[sources.mqtt.topics]]
filter = "plant/+/telemetry"
qos = 1
category = "metric"
payload_format = "json"
fields = { temperature = "/values/temperature", speed = "/values/speed" }
timestamp_pointer = "/ts"
timestamp_unit = "ms"
topic_tags = { area = 1 }

[[sources.mqtt.topics]]
filter = "plant/+/events"
qos = 1
category = "logging"
payload_format = "json"
message_pointer = "/message"
status_pointer = "/status"
topic_tags = { area = 1 }

For TLS and username authentication, use a [sources.mqtt] configuration similar to the following:

[sources.mqtt]
brokers = ["ssl://192.0.2.31:8883"]
client_id = "industrykit-plant-a"
username = "<MQTT_USERNAME>"
password = "<MQTT_PASSWORD>"
ca_file = "pki/mqtt-ca.pem"
cert_file = "pki/mqtt-client.pem"
key_file = "pki/mqtt-client-key.pem"
connect_timeout = "10s"
keep_alive = "30s"

Options

[[sources]]

Field Default Description
id None Required and unique within the current configuration. Reported as the source_id tag.
type None Must be mqtt.
measurement industrial Name of the metric measurement or log source used when a topic does not override it.
interval 10s MQTT receives messages passively. This value does not control message arrival or reporting frequency.
tags {} Static tags for this source. They are merged with global and topic tags.

[sources.mqtt]

Field Default Description
brokers None Required non-empty array. Each entry must be tcp://host:port, ssl://host:port, or tls://host:port, with a valid explicit port and no path, credentials, query, or fragment.
client_id <collector.id>-<source.id> Persistent-session ID. It must be unique for the same Broker and transport; ssl and tls are treated as the same transport.
username Empty Optional Broker username.
password Empty Optional Broker password. When set, username must also be set.
ca_file Empty TLS CA PEM file. The system trust store is used when omitted.
cert_file Empty TLS client certificate. It must be configured together with key_file.
key_file Empty TLS client private key. It must be configured together with cert_file.
connect_timeout 10s Connection and subscription timeout. Must be greater than 0.
keep_alive 30s MQTT keep-alive interval. Must be greater than 0.

Relative certificate paths are resolved from the directory containing the main configuration file. The IndustryKit service account must have permission to read these files. The minimum TLS version is 1.2, and the Server name is taken from the host in the Broker URL.

[[sources.mqtt.topics]]

MQTT does not use points and therefore has no separate field setting. For a JSON metric, each key on the left side of fields is an output field name, and its value is the JSON Pointer used to extract data from the payload. For a text metric, value_field specifies the output field name.

Field Default Description
filter None Required and unique within the current source. + must occupy an entire level, and # must occupy the final level by itself.
qos 0 Subscription QoS: 0 or 1.
category metric Output category: metric or logging.
payload_format json Payload format: json or text.
measurement Inherits from the source Overrides the metric measurement or log source name for this topic.
fields {} Required non-empty mapping for metric + json: output field name = JSON Pointer. Pointers follow RFC 6901.
value_field value Output field name for metric + text. The entire payload is written to this field as a string.
message_pointer Empty Required for logging + json. Locates the log message.
status_pointer Empty Optional for logging + json. Locates the log status; when omitted, the status is info.
timestamp_pointer Empty Optional JSON Pointer to the message timestamp. The receive time is used when omitted.
timestamp_unit ms when timestamp_pointer is set Numeric timestamp unit: s, ms, us, or ns. This value is not used for RFC 3339 timestamp strings.
tags {} Static tags for this topic. collector_id, protocol, source_id, and mqtt_topic are not allowed.
topic_tags {} tag name = zero-based topic level index. The index must be non-negative; a message is invalid if the index is out of range. Fixed tags and mqtt_topic cannot be used as tag names.

Metric fields, messages, and statuses selected by JSON Pointer must be scalar values. A missing path, null, object, or array causes that mapping to fail and the message to be discarded. A payload must contain exactly one complete top-level JSON document, with no trailing second JSON value. See IndustryKit for common installation, operation, and verification instructions.

Data and limitations

  • Every data point includes the mqtt_topic tag.
  • A text metric writes the entire payload to value_field; text logging uses the entire payload as message.
  • With QoS 1, IndustryKit sends an ACK after all matching data has been written to the WAL. Retries at the boundary may produce duplicates. Invalid messages are discarded and counted. QoS 0 is best effort.
  • The maximum MQTT packet size is 512 KiB.
  • IndustryKit currently subscribes to topics only; it does not publish device-control messages.

Feedback

Is this page helpful?