OPC UA PubSub

MQTT Industrial IoT

Synopsis

Creates an OPC UA PubSub target that publishes records to an MQTT broker as one OPC UA PubSub publisher. By default, each device that sends records is a DataSet writer of that publisher. OPC UA subscribers read the values of each device from the broker.

The target follows OPC UA Part 14 (PubSub), version 1.05, in its JSON message mapping. It works like the MQTT target, with the rules of OPC UA PubSub added.

note

The target follows OPC UA Part 14.

Schema

- name: <string>
  description: <string>
  type: opcuapubsub
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    url: <string>
    publisher_id: <string>
    writer_group: <string>
    topic_prefix: <string>
    dataset_writer: <string>
    dataset_writer_missing_field: <string>
    dataset_writer_fallback: <string>
    dataset_writer_default_value: <string>
    fields: <string[]>
    client_id: <string>
    username: <string>
    password: <string>
    protocol_version: <string>
    qos: <numeric>
    timeout: <numeric>
    keep_alive: <numeric>
    clean_session: <boolean>
    auto_reconnect: <boolean>
    batch_size: <numeric>
    max_messages_per_second: <numeric>
    max_retries: <numeric>
    retry_delay: <numeric>
    disable_retries: <boolean>
    field_format: <string>
    tls:
      status: <boolean>
      verify: <boolean>
      ca_name: <string>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>

Configuration

The following fields are used to define the target:

Field	Required	Default	Description
name	Y		Target name
description	N	-	Optional description
type	Y		Must be opcuapubsub
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

Connection

Field	Required	Default	Description
url	Y	-	The address of the MQTT broker, such as tcp://broker.example.com:1883. In the console: URL
client_id	N	The Publisher ID	The MQTT client ID. OPC UA Part 14 says that a publisher uses its publisher ID as its client ID. In the console: Client ID (Optional)
protocol_version	N	3.1.1	3.1.1 or 5. See Retained Messages and MQTT 5
timeout, keep_alive, username, password	N	30, 60, -, -	As on the MQTT page, with the same labels. timeout is also the time that a stop has to clear the retained messages

Publisher

Field	Required	Default	Description
publisher_id	Y	-	The ID of the publisher, which is this Director. It is one level of every topic, and it is the default client ID. Only one Director may run the target (see One Director per Publisher). In the console: Publisher ID
writer_group	N	WriterGroup1	The name of the writer group. It is one level of the data and metadata topics. In the console: Writer group (Optional)
topic_prefix	N	opcua	The first level of every topic. In the console: Topic prefix (Optional)

publisher_id, writer_group and topic_prefix are each one level of a topic. The target refuses a value that holds /, + or #, that starts with $, or that holds a control character or a whitespace character other than the space.

DataSet Writers

Field	Required	Default	Description
dataset_writer	N	The name of the device that sent the record	The DataSet writer that each record goes to. A template with record fields in braces, such as {host.name}. See How Writers and Fields Work. In the console: DataSet writer (Optional)
dataset_writer_missing_field	N	drop	What happens to a record that has no value for a field of dataset_writer: drop, fallback or default_value. In the console: Missing DataSet writer field (Drop the record, Use a fallback DataSet writer, Use a default value)
dataset_writer_fallback	Y, with fallback	-	The writer that such a record goes to. In the console: Fallback DataSet writer
dataset_writer_default_value	N	unknown	The text that takes the place of a missing field, with default_value. In the console: Default value
fields	N	Every number and boolean	The fields to send. Each entry names a field, or an object whose fields are all sent. In the console: Fields to send (Optional)

These work as on the Sparkplug B page, with DataSet writers in place of devices.

Delivery Settings

Field	Required	Default	Description
qos	N	1	The QoS of the data messages: 0, 1 or 2. Status and metadata messages always go out at QoS 1
batch_size	N	1000	How many records the target holds before it publishes them
max_messages_per_second	N	No limit	The most messages the target publishes in a second. Every message counts: status, metadata and data. 0 or no value means no limit
max_retries, retry_delay	N	3, 1	How many times the target tries a failed record again, and the seconds it waits before each try
disable_retries	N	false	true makes one attempt for each record, with no retry and no wait
field_format	N	-	Normalizes each record to this field format before the target publishes it. See applicable Normalization section

These work as on the MQTT page, with the same labels.

TLS

tls.status turns TLS on. The other TLS settings (tls.verify, tls.server_name, tls.ca_name, tls.cert_name, tls.key_name, tls.passphrase, tls.min_tls_version and tls.max_tls_version) work as on the MQTT page. tls.verify is true by default.

Settings That OPC UA PubSub Fixes

OPC UA PubSub fixes some of the MQTT settings. Data messages are never retained, and the target gives each connection a last will of its own. So the MQTT target's topic, retained and will settings have no effect here. clean_session and auto_reconnect work as for the MQTT target, and are true by default.

Scheduling

See Scheduling and Pool Behavior for interval and cron fields shared by all targets.

The Console Form

The form has five steps: Connection settings, Publisher, DataSet writers, Delivery settings and TLS settings. It shows these messages next to a field:

• Publisher ID is required
• Publisher ID cannot contain /, + or #, start with $, or contain a tab, a line break or another control character, and the same message for Writer group and Topic prefix
• Fallback DataSet writer is required, when the fallback choice is on

The form checks the three names as Director does. It does not check the DataSet writer template, because Director makes every writer name fit a topic (see How Writers and Fields Work).

Details

One Director per Publisher

The Publisher ID is part of every topic, and it is the default client ID. So run the target on one Director only, with Director restriction in General settings.

• With the default client_id, a second Director takes over the connection of the first.
• With a client_id of its own, both stay connected, and the subscribers get the messages of both as one publisher.

Topics

• Data: <topic_prefix>/json/data/<publisher_id>/<writer_group>/<writer>, such as opcua/json/data/Director1/WriterGroup1/pump-1.
• Metadata: <topic_prefix>/json/metadata/<publisher_id>/<writer_group>/<writer>.
• Status: <topic_prefix>/json/status/<publisher_id>.

How the Target Publishes

• All workers share one MQTT connection, and one worker writes for the whole target. So the sequence numbers of each writer go out in order, and so do the records of each writer.
• The target publishes every record. Unlike the Sparkplug B target, it does not hold back a value that has not changed.
• When a message fails, the sender sends the whole batch again. A data message that failed used no sequence number, so the next try uses it. A publish that timed out can still reach the broker, so a data message can arrive twice with the same sequence number. A subscriber can drop the second one by that number.
• A record that has no writer name is dropped. It counts as dropped in the stats, and the batch still succeeds.

How Writers and Fields Work

• Without dataset_writer, a record goes to the writer that has the name of the device that sent the record. A record from a device with no name is dropped.
• With dataset_writer, the name of a record is the template with each {field.path} replaced by the value of that field in the record. A value is missing when the field is not in the record, or when it holds null, an empty string, an object or an array. drop drops the record, fallback sends it to the writer that dataset_writer_fallback names, and default_value puts dataset_writer_default_value where the value would go.
• Writer names. The name of a writer is one level of a topic. So each /, + and #, each control character, each whitespace character other than the space, a $ at the start and each run of bytes that is not UTF-8 becomes _. A name that changes this way gets - and 8 hexadecimal characters derived from the original name. So the devices Line/1 and Line_1 stay two writers, Line_1-cc6fedab and Line_1, and the name stays the same after a restart. A name that needs no change stays as it is.
• Writer IDs. Each writer has a DataSetWriterId, a number from 32768 to 65535, derived from its name, so the same name gets the same ID after a restart. When two names give the same ID, the writer that sends its first record later takes the next free ID, counting on from 65535 to 32768. After a restart, the IDs of such names can swap when they send in another order. A Director holds at most 32768 writers, and a record for a new writer beyond that is dropped.
• Fields. Each number of a record is a field of type Double, and each boolean a field of type Boolean. The name of a field is the path to its value, with / between the levels, such as gauge/bar. Strings, arrays and nulls are no fields, and neither is @timestamp.
• fields limits what the target sends. An entry is the path to a field, with a dot between the names, such as cpu.user. When the field is an object, every number, boolean and string inside it is a field. With fields, a string is a field too, of type String. The target reads objects up to 16 levels deep, and leaves out the values of an object nested deeper.
• A writer holds at most 1,000 fields. The target leaves out the value of a field that does not fit, and logs a warning that lists some of the names, at most once a minute.
• A record that has no value to send publishes nothing, and it counts as delivered.

Messages

The target sends one data message for each record, holding one key frame of the writer. For example, written over several lines:

{
  "MessageId": "6d1c7e0e-5a3b-4a62-bb1a-0d0d4e1f0b2a",
  "MessageType": "ua-data",
  "PublisherId": "Director1",
  "WriterGroupName": "WriterGroup1",
  "Messages": [
    {
      "DataSetWriterId": 37305,
      "DataSetWriterName": "pump-1",
      "SequenceNumber": 0,
      "MinorVersion": 753315200,
      "Timestamp": "2023-11-14T22:13:20.1234567Z",
      "MessageType": "ua-keyframe",
      "Payload": { "temp": 21.5, "on": true }
    }
  ]
}

• The Timestamp is the time of the record. The Payload holds every field of the writer, in the order of its metadata: the value that the record has, or null for a field that it lacks. A number that JSON cannot write is a string: "NaN", "Infinity" or "-Infinity".
• The SequenceNumber counts the data messages of a writer from 0, and starts at 0 again after 4294967295. A restart of Director starts it at 0. The MinorVersion is the version of the metadata that the message follows.

Metadata

• A writer publishes its metadata (ua-metadata), retained, before its first data message on each connection, and again before the data whenever a record changes its fields.
• The metadata lists each field with its name, its type (BuiltInType and DataType: i=1 is Boolean, i=11 is Double and i=12 is String) and its DataSetFieldId, a UUID derived from the publisher ID, the writer name and the field name, so a restart does not change it.
• The metadata holds a ConfigurationVersion with a MajorVersion and a MinorVersion. A new field raises the MinorVersion. A field that changes its type raises both. A version is the number of seconds from 2000-01-01 00:00:00 UTC to the time of the change; when that is not above the highest version so far, it is one more than the highest, so a version never goes back.
• A writer never drops a field. A field that stops appearing stays in the metadata, and its value is null.

Status and Last Will

• A status message (ua-status) looks like this, where Status is 2 for Operational and 3 for Error: {"MessageId":"...","MessageType":"ua-status","PublisherId":"Director1","IsCyclic":false,"Status":2}.
• After every connect, the publisher publishes the status Operational to the status topic, retained at QoS 1, before any data. When that status cannot be published, the target sends no data and tries the record again.
• The last will is the status Error on the same topic, retained at QoS 1. The broker publishes it when the connection ends without a goodbye.
• Every connect, the first one and each reconnect, also makes each writer publish its metadata again before its next data, because the broker may have lost the retained messages.

Retained Messages and MQTT 5

• The status and the metadata are retained, so a subscriber that connects later finds them. The data is not retained.
• When Director stops the target, by a stop or a reload, the target publishes an empty retained message at QoS 1 to the status topic and to the metadata topic of every writer, which clears them. Then it disconnects in the normal way, so the broker drops the will.
• The whole clearing has one deadline, the timeout. With a low max_messages_per_second and many writers, the target stops clearing at the deadline, logs a warning and disconnects, and the broker can keep some of the retained messages.
• Under MQTT 5, every status, metadata and data message, and the last will, has the content type application/json and the user property UAMessageType.
• Under MQTT 5, a retained status or metadata message expires after an hour, so the broker drops it when Director has gone without clearing it. To keep them alive, the target publishes the status and the metadata of each writer again every 30 minutes while it is connected.
• Under MQTT 3.1.1, the messages have no properties and no expiry, and the target does not publish them again.

What the Target Does Not Send

• It sends the JSON mapping only, not the binary UADP mapping.
• Every data message is a key frame. The target sends no delta frames, no keep-alive messages and no events.
• It sends no connection, application or endpoints discovery messages. It subscribes to nothing, so it answers no discovery request and takes no command.

Examples

Defaults

One publisher, with a DataSet writer for each device...

targets: - name: opcua_pubsub type: opcuapubsub properties: url: "tcp://broker.example.com:1883" publisher_id: "Director1"

Named Writers, Chosen Fields, MQTT 5 over TLS

Writers named after a record field, with a fallback writer, only the holding registers sent, and MQTT 5 over TLS...

targets: - name: opcua_pubsub_plant type: opcuapubsub properties: url: "ssl://broker.example.com:8883" publisher_id: "Plant1-Director" writer_group: "Line1" dataset_writer: "{host.name}" dataset_writer_missing_field: "fallback" dataset_writer_fallback: "unknown-device" fields: - "modbus.register.holding" protocol_version: "5" username: "${MQTT_USERNAME}" password: "${MQTT_PASSWORD}" tls: status: true

Troubleshooting

Subscribers See No Data

Check that the subscribers subscribe to <topic_prefix>/json/data/#, and that the Publisher ID, the Writer group and the Topic prefix match what they expect.

A Second Director Takes Over the Connection

Two Directors run the target with one Publisher ID and the default client ID. Turn on Director restriction in General settings (see One Director per Publisher).

The Status Stays Error

Director has not reconnected, or it cannot publish to the status topic. Check the rights of the MQTT user on that topic.

Fields Are Left Out

A writer already holds 1,000 fields. Director logs a warning at most once a minute, with some of the names.

A Writer Name Has a Suffix

The device name held a character that a topic level cannot hold. See the writer names under How Writers and Fields Work.