Skip to main content

OPC-UA Source Node

The opcua_source node subscribes to variables on an OPC-UA server and emits one record per data-change notification. It is aimed at industrial automation data — PLCs, SCADA systems, and historians that expose an OPC-UA endpoint — and reports each value together with its OPC-UA status code and source/server timestamps.

Unlike most sources, this node has no encodingType: OPC-UA is a typed protocol, so values arrive already structured and are mapped directly into a fixed record shape.

Key Features​

  • Subscription-based: the server pushes only on data change, rather than the node polling every tag on an interval
  • Per-node sampling control: publishing and sampling intervals and the server-side queue size are configurable
  • All four NodeId flavors: NUMERIC, STRING, GUID, and OPAQUE identifiers
  • Quality preserved: each record carries the OPC-UA status code and a status_good flag, so bad-quality reads are visible downstream instead of silently becoming values
  • Both timestamps: source timestamp (when the device sampled) and server timestamp (when the server observed) are emitted separately
  • Optional security: unencrypted, or Basic256Sha256 sign-and-encrypt; anonymous or username/password identity
  • Bounded buffering with accounting: a full inbound queue drops notifications and increments an opcua_dropped_count metric rather than growing without limit
  • Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration​

FieldTypeRequiredDefaultDescription
endpointUrlStringYes—OPC-UA endpoint. Must start with opc.tcp://, e.g. opc.tcp://plc-host:4840
nodeIdsList<Object>Yes—Nodes to monitor. Must be non-empty. See Monitored Nodes
securityPolicyStringNoNONENONE for an unencrypted channel, or BASIC256SHA256 to sign and encrypt
usernameStringNo—Username. Omit for an anonymous identity
passwordStringNo—Password. Only valid together with username
publishingIntervalMsdoubleNo1000.0Subscription publishing interval in milliseconds. Must be > 0
samplingIntervalMsdoubleNo1000.0Per-item sampling interval in milliseconds. Must be > 0
serverQueueSizeintNo10Server-side queue depth per monitored item. Must be > 0
inboundQueueCapacityintNo10000Bounded client-side queue between the OPC-UA callback and the pipeline loop
offerTimeoutMslongNo1000How long to wait offering to a full inbound queue before dropping the notification
shutdownGraceMslongNo5000On shutdown, how long to wait for the queue to drain before forcing exit
requestTimeoutMslongNo5000OPC-UA request timeout in milliseconds

Monitored Nodes​

nodeIds is a list of objects, each identifying one OPC-UA variable:

FieldTypeRequiredDescription
namespaceIndexintYesThe node's namespace index. Must be >= 0
identifierStringYesThe node identifier, parsed according to identifierType
identifierTypeStringYesNUMERIC, STRING, GUID, or OPAQUE

How identifier is parsed depends on identifierType:

identifierTypeExpected identifier formatExample
NUMERICAn unsigned integer2258
STRINGAny string, used verbatimMachine1.Temperature
GUIDA UUID09087e15-1e1f-4c40-b76e-4e5c8b1a0f3a
OPAQUEBase64-encoded bytesAQIDBA==

Identifiers are parsed at validation time, so a NUMERIC identifier that isn't a number, or a malformed GUID, fails when the pipeline is validated rather than mid-run.

config:
  nodeIds:
    - namespaceIndex: 2
      identifier: "Machine1.Temperature"
      identifierType: "STRING"
    - namespaceIndex: 2
      identifier: "Machine1.Pressure"
      identifierType: "STRING"
    - namespaceIndex: 0
      identifier: "2258"
      identifierType: "NUMERIC"

Record Shape​

Each data-change notification becomes one record:

{
  "node_id": "ns=2;s=Machine1.Temperature",
  "namespace_index": 2,
  "identifier": "Machine1.Temperature",
  "value": 21.4,
  "status_code": 0,
  "status_good": true,
  "source_timestamp_ms": 1753280531318,
  "server_timestamp_ms": 1753280531402,
  "received_at_ms": 1753280531455
}
FieldDescription
node_idThe full NodeId in OPC-UA's parseable form
namespace_indexNamespace index of the node
identifierThe node's identifier as a string
valueThe value read. null when the notification carried no value
status_codeRaw OPC-UA status code. null when absent
status_goodtrue only when a status code is present and good
source_timestamp_msEpoch millis when the source sampled the value. null when the server did not supply one
server_timestamp_msEpoch millis when the server observed the value. null when not supplied
received_at_msEpoch millis when this pipeline received the notification

Always check status_good before trusting value — OPC-UA reports bad-quality reads as notifications with a non-good status, and the node passes them through rather than dropping them.

Because one record is emitted per node per change, downstream nodes see a stream of individual tag readings, not a combined row per device. Use a processing node to correlate tags if you need them together.

Security​

Security policy​

NONE (the default) opens an unencrypted channel and no client certificate is involved.

BASIC256SHA256 signs and encrypts the channel. In this mode the node generates an ephemeral self-signed client certificate in memory at startup, and trusts the server's certificate without pinning it.

warning

With BASIC256SHA256, the server certificate is accepted without validation. The channel is encrypted, but the server's identity is not verified, so this does not protect against a man-in-the-middle on the path to the endpoint. Use it on trusted networks, and treat endpointUrl as the trust boundary.

Because the client certificate is generated fresh in memory on each start, servers that require a pre-registered and trusted client certificate will reject the connection.

The node selects an endpoint whose security policy matches the configured one, so the server must actually offer that policy.

Identity​

Setting username (with optional password) authenticates with a username identity; omitting it connects anonymously. Supplying a password without a username is rejected at validation.

How It Works​

  1. The node connects to endpointUrl, selecting an endpoint matching securityPolicy, and authenticates with the configured identity.
  2. It creates a subscription with publishingIntervalMs and registers each entry in nodeIds as a monitored item using samplingIntervalMs and serverQueueSize.
  3. The server pushes data-change notifications, which are offered to a bounded queue sized by inboundQueueCapacity.
  4. The pipeline loop drains the queue, converting each notification into a record.

Backpressure and drops​

OPC-UA has no consumer-driven flow control here, so the client-side queue is the buffer. When it is full, the node waits up to offerTimeoutMs for room; if none frees up, the notification is dropped and the opcua_dropped_count metric is incremented.

Sustained drops mean the pipeline cannot keep up with the tag change rate. Raise inboundQueueCapacity to absorb bursts, or raise samplingIntervalMs / publishingIntervalMs to reduce the incoming rate.

On shutdown the node waits up to shutdownGraceMs for the queue to drain before exiting.

DAG Example​

jobContext:
  otherProperties: {}
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "opcua_source"
    config:
      endpointUrl: "opc.tcp://plc-host:4840"
      securityPolicy: "NONE"
      nodeIds:
        - namespaceIndex: 2
          identifier: "Machine1.Temperature"
          identifierType: "STRING"
        - namespaceIndex: 2
          identifier: "Machine1.Pressure"
          identifierType: "STRING"
      publishingIntervalMs: 1000.0
      samplingIntervalMs: 1000.0
      serverQueueSize: 10
      inboundQueueCapacity: 10000
    outputs:
      - "good_only"

  - id: "good_only"
    commandName: "filter"
    config:
      expression: $.status_good == true
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"