Live streams

Collector: MQTT and Sparkplug

Subscribe to an MQTT broker, a Unified Namespace or Sparkplug B edge nodes.

MQTT and Sparkplug B

Use this source when your data already travels through an MQTT broker: a Unified Namespace (UNS), Ignition with the MQTT Transmission module, Node-RED, an IIoT gateway or a protocol converter (Kepware, Litmus, Neuron) that publishes to MQTT. The collector is an MQTT subscriber: it connects out to the broker and never publishes data. It subscribes to what you configure and stores every message it receives.

sparkplug: Sparkplug B. Names, units and types come from the birth messages, so there is nothing to map.
json: plain topics with JSON payloads, with an optional mapping.
raw: one number per message, the topic is the channel.

Install the connector with pip install "kymoscope-collector[mqtt]".

Sparkplug B

collector.yaml

sources:
  - name: Customer A UNS
    type: mqtt
    broker: mqtts://broker.customer.local:8883
    ca: broker-ca.pem
    username: kymo-reader
    password: env:MQTT_PASSWORD
    mode: sparkplug
    group: CustomerA
    edge_nodes: [Line1, Line2]   # optional; all edge nodes of the group when left out

The collector subscribes to spBv1.0/<group>/+/<edge>/#. Each metric becomes a channel named edge/device/metric (without the device for node-level metrics); its unit is the engUnit property when the birth message has one. Values keep the metric timestamp, else the payload timestamp, else the arrival time. Booleans and integers are stored as numbers; text, datasets and templates are not channels and are ignored.

Aliases are resolved with the table of the last birth message.
Null metrics (is_null) are stored as a gap.
Death (NDEATH / DDEATH) marks every channel of that node or device as not connected (a Bad status) at the moment the message arrives. An NDEATH whose bdSeq differs from the node's last birth is a stale last-will from an earlier session and is ignored.
Sequence gaps are counted. A collector that starts after the edge node has already published its birth cannot resolve aliases until the node publishes a birth again. With rebirth: true the collector then publishes a Rebirth command (NCMD, Node Control/Rebirth) to that edge node, at most once every 30 s per node.

JSON topics

collector.yaml

sources:
  - name: Packaging line
    type: mqtt
    broker: mqtts://broker.customer.local:8883
    mode: json
    topics:
      # every numeric leaf of the payload becomes a channel: <topic>.<key>
      - customer/utility/#
      # a mapping: value path, timestamp path and a name built from topic segments
      - topic: uns/+/+/telemetry
        value: data
        ts: meta.t
        ts_format: ms
        tag: "{topic[1]}.{topic[2]}.{key}"
        unit: bar

Key (in a topics entry)	Meaning
`topic`	The MQTT topic filter (+ and # wildcards). The first entry that matches a message is used.
`value`	Path to the value or object to store, such as data.temp or data (dotted, numbers index arrays). Left out: every numeric leaf of the payload.
`ts`	Path to the timestamp in the payload. Left out, or when it cannot be read: the arrival time.
`ts_format`	auto (default), s, ms, us or ns for a number; ISO 8601 text is always understood.
`tag`	Channel name template: {topic}, {topic[N]} (segment N from 0) and {key} (the leaf path). Default: {topic}.{key}, or {topic} for a single value.
`unit`	Unit shown on the channels of this entry.

Raw topics

collector.yaml

sources:
  - name: Sensors
    type: mqtt
    broker: mqtts://broker.customer.local:8883
    mode: raw
    topics:
      - sensors/+/temperature
      - topic: sensors/+/pressure
        tag: "{topic[1]}.pressure"

Each message is one number as text (23.5). The channel is named after the topic unless tag says otherwise, and the timestamp is the arrival time at the collector. If the sender puts a timestamp in the payload, use json mode with value and ts instead.

Connection and security

collector.yaml

sources:
  - name: Customer A UNS
    type: mqtt
    broker: mqtts://broker.customer.local:8883
    ca: broker-ca.pem            # the CA that signed the broker certificate (system CAs when left out)
    cert: collector.pem          # client certificate, if the broker asks for one
    key: collector-key.pem
    # insecure: true             # last resort: no certificate check at all
    # broker: mqtt://10.0.0.5:1883 + allow_insecure: true   # plain text, only on a trusted network

Option	Default and meaning
`broker`	mqtts://, mqtt://, wss:// or ws://host[:port]. Ports default to 8883, 1883, 443 and 80. Plain text (mqtt://, ws://) needs allow_insecure: true.
`username`	Broker user. password takes env:NAME or file:path.
`qos`	1. The subscription QoS, 0 to 2.
`client_id`	Random per start. Set a fixed one if the broker restricts client ids.
`clean_session`	true. With false (needs a client_id) the broker keeps queued QoS 1 messages while the collector is down.
`discover_s`	10. How long kymoscope-collector discover listens.
`max_tags`	5000. The most channels a json or raw source creates. High-cardinality topics (a + over device ids, long arrays) would otherwise create channels without limit; beyond it new channels are dropped, counted in tags_dropped and logged (at most once a minute). Narrow topics or raise it.
`reconnect_max_s`	60. The longest wait between reconnect attempts after the broker connection is lost (it starts at 1 s).

Ask the broker administrator for a user that can only subscribe: read access on the topics above (spBv1.0/<group>/# for Sparkplug) and the broker's TLS port reachable from the collector host. Nothing needs to reach the collector.

Finding what is published

kymoscope-collector discover --config collector.yaml --source "Customer A UNS" --out tags.csv listens for discover_s seconds and lists the channels it saw. In Sparkplug mode only nodes that publish a birth in that window appear, so run it while an edge node starts or is rebirthed; in json and raw modes only topics that publish within the window appear. This source has no history, so backfill does not apply: MQTT does not store past values.

Troubleshooting

The health stream shows connected 0: check the address and port, the CA file, and the collector log. A refused login is logged as "broker refused the connection". A certificate error means the CA is missing or the host name is not in the broker certificate.
Connected, but no values: the broker user may lack subscribe rights (many brokers drop the subscription silently). Try the same user with MQTT Explorer or mosquitto_sub. For Sparkplug, check the group spelling (it is case sensitive).
Sparkplug values missing after a collector restart: unresolved in the health stream counts data messages with aliases the collector has no birth for. Enable rebirth or rebirth the node from the Sparkplug host.
decode_errors rising: the payload is not what the mode expects (JSON in a Sparkplug group, text in a raw topic). Compare with the broker's own view of the topic.
seq_gaps rising: messages were lost between the edge node and the collector, usually QoS 0 on a busy or flapping link.
Health values for this source: connected, reconnects, messages_per_s, values_per_s, decode_errors, births, deaths, rebirths_requested, seq_gaps, unresolved, subscribed, tags_dropped.