Event Streams

Last updated

An event stream is a pipeline: events come from one source, pass through stages that filter and reshape them, are grouped by a buffer into batches, and each batch goes to one or more handlers. Build them on the Event Streams page. Streams run on the appliance, keep their order, retry what fails and keep what they cannot deliver in a dead-letter list.

Event Streams is included and off by default. An administrator turns it on in Settings → Feature Toggles (event_streams). While it is off, the page is hidden, its API answers 403, and saved streams show Feature off and do not run.

Sources

Every event has source and timestamp, plus the fields of its source:

Source	An event when	Fields
Tag changes	A watched tag changes value or quality. Watch a list of tags, a path pattern (* within one level, ** across levels, ? one character; on one device or any), or one member (or all) of every instance of a UDT	deviceId, deviceName, tagPath, value, previousValue, dataType, quality; UDT sources add instance and member
Alarm events	An alarm fires, clears or is acknowledged (pick which)	action, historyId, label, severity, message, deviceId, tagPath, value, firedAt; acknowledgements add userId and comment
Messages	A script's messages.send or a workflow's Send Message node sends a message with this name	name, payload, and fromPeer for a message from a paired appliance
HTTP endpoint	Another system posts to the stream (below)	payload, remoteAddress
MQTT topics	A topic matching the filter (+ one level, # the rest) receives a new payload	deviceId, topic, payload (parsed when it is JSON or a number), quality
Timer	Every interval (at least 100 ms)	count

Tag sources pass good quality readings only, unless Include bad and uncertain quality readings is on. A value is a number when the tag's value is numeric, true/false for a boolean, otherwise text. Only changes reach a stream: a tag that keeps the same value sends nothing.

An MQTT source sees the topics that are tags on the MQTT device (add them on the device's tags, wildcards included in the device's subscription), and like any tag only a changed payload makes an event. A stream listening for a message counts as a recipient of messages.send, but never answers messages.sendRequest.

UDT sources pick up new instances within 15 seconds.

Stages

Stages run in order; each can be switched off without removing it.

• Filter keeps the events for which its expression is true.
• Map fields sets fields from expressions. With Keep the incoming fields on, the event is changed; off, the result has only the mapped fields. Dots make nested fields (payload.line). A field set to a lone reference such as event.payload copies it whole, objects included.
• Script runs JavaScript in the contained script host, never in the appliance's API process: no network, no Scada.Ink API, and 2 seconds for each batch of up to 200 events. The code sees event (and index). Change event and return nothing to keep it, return a new object to replace it, return null to drop it, or return an array of objects to split it into several events.

Filters, map fields and the handlers' value, column and key settings use the expression language of Expression Tags, with the event's fields as event.<field>:

event.value > 100 && event.quality == "good"
event.payload.count * 2
upper(event.deviceName) + ":" + event.tagPath
event.value > {3:Line1/Limit}

A live tag still reads as {<deviceId>:<tagPath>}. A missing field reads as empty text; true/false read as 1/0. Text in quotes is never treated as a field.

Test the stages (beside the pipeline) runs one sample event through the stages as they are on screen, saved or not, and shows what each stage produced. No handler runs.

When a stage fails on an event (a filter reading a tag that has no value, a script that throws), that event goes to Dead letters and the others carry on.

Buffer

A batch goes to the handlers when it holds Batch size events, or when its first event has waited Longest wait milliseconds. Size 1 sends each event on its own. A wait of 0 sends whatever has arrived without waiting, so batches form only under load.

Handlers

Handler	Sends	Kept until delivered
Write tag	Each event's Value expression to one tag, in order. Local tags are written directly; a device tag needs writes enabled on the device	No
Data Table / remote table	One row per event, each column from its expression. A batch goes into a Data Table all at once or not at all. A remote table uses the remote database's own store & forward when the database is unreachable	Yes
HTTP request	POST, PUT or PATCH, one request per batch (a JSON array) or one per event. Header values may use ${secret:NAME}; in one-per-event mode the URL may use {{field}}	Yes
MQTT publish	Each event, or the batch as one array, as JSON to a topic on an MQTT device with writes enabled. {{field}} in the topic inserts an event field	Yes
Kafka	Records through the Kafka connector picked in Data connector (only Kafka connectors are listed), to Topic, keyed by Key when set, one record per event or one per batch (Send)	Yes
Data connector	Records through any data connector picked in Data connector (Kafka, AWS IoT Core, Azure IoT Hub, Google Pub/Sub, MongoDB), with the same Topic, Key and Send fields. An empty topic uses the connector's own	Yes
Run workflow	One run per batch. The run's trigger data has event (the first event), events (the batch), count, streamId and streamName	No
Send message	Each event as a message to scripts and workflows listening for the name, with stream added	No
Stream log	Each event as a line in the stream's log (Status tab) and the appliance log	No

The HTTP handler follows the same network rules as scripts. It reaches the plant network only with Plant network on and the internet only with Internet on, the appliance-wide script network switch (Settings → Scripts) applies, and local services on the appliance are refused except its own API port. Kafka and data connector handlers need the Data Connectors feature and a connector set up there. The Data connector list shows the connectors by name and type (marked disabled when switched off); while the feature is off or no connector exists it is empty and says so. A handler whose connector was deleted shows it as (no longer exists), and its batches go to dead letters at once with Data connector "<name>" does not exist. While the connector is disabled, or the feature is off, batches are retried and then dead-lettered (Data connector "<name>" is disabled, Data connectors are switched off on this appliance). The handler's job ends when the batch is in the connector's queue; the connector then delivers it with its own store & forward.

On the standby of a redundant pair no stream runs: enabled streams show Standby there, and start within a second when that appliance becomes the active one. Their queued batches and dead letters are copied to the standby with the rest of the configuration, so after a takeover delivery carries on from where the other appliance left off (a batch may arrive twice).

Delivery, retries and dead letters

Each handler delivers its batches strictly in order. When a delivery fails, the same batch is tried again after First wait, doubling each time up to Longest wait (with a little jitter), and that handler's later batches wait behind it. The other handlers carry on.

A batch moves to Dead letters after Attempts tries, or at once when it can never succeed: an HTTP 4xx other than 408 and 429, a refusal by the network rules, a value a column or tag refuses, a device without writes enabled, a workflow that no longer exists, or a message loop past the hop limit.

Handlers marked kept until delivered write each batch to the appliance's database before trying it and remove it only once delivered, so a batch survives a restart or a power cut and is delivered at least once. A batch that failed partway (event 3 of 5 in one-request-per-event mode) is sent again from the start, so the receiver may see some events twice. Write tag, Run workflow, Send message and Stream log are not kept across a restart on purpose: a tag written minutes late, after the plant has moved on, can do more harm than a write that never happened. Events that are still in the stages or the buffer when the power fails are lost.

When a handler already has Most batches waiting, new batches for it are dropped and counted. When more than 10,000 events wait to enter the stages, new events are dropped and counted (Dropped (intake full)). An event larger than 256 KB is dropped.

On the Dead letters tab, each entry shows where it failed (a handler or a stage), when, the attempts and the error, and its events. Replay puts a handler failure back in that handler's queue and runs a stage failure through the pipeline from the start; Delete removes it. Replaying to a handler that is not kept until delivered, or replaying a stage failure, needs the stream to be running. The newest 10,000 dead letters per stream are kept. Removing a handler from a stream moves its waiting batches to dead letters.

Status

The Status tab, refreshed every two seconds, shows events in and out, events filtered out, stage errors, the in and out rates over the last 10 seconds, events still being processed, and events dropped. For each handler: events delivered, failed attempts, batches given up on, batches waiting, the next retry time and last error, and the latency from an event arriving to the handler accepting it (average and 95th percentile over the last 500 batches). It also lists the latest events out of the stages and the stream's log. Counters start again when the stream restarts; saving a running stream restarts it.

The HTTP source

A stream with an HTTP source receives events at POST /api/event-streams/ingest/<stream id> (the page shows the full address). Save the stream, then Generate key; the key is shown once. Send it in an X-Stream-Key header (or Authorization: Bearer <key>). The body is one JSON object, or an array of up to 1,000; each becomes one event whose payload is that object.

curl -X POST https://<appliance>/api/event-streams/ingest/<stream id> \
  -H "X-Stream-Key: <key>" -H "Content-Type: application/json" \
  -d '[{"machine":"press-3","state":"running"}]'

Answer	Meaning
202 with { accepted, dropped }	Taken in
401 Invalid stream key	No key, the wrong key, or no such HTTP stream (the same answer for each)
403	Event Streams is switched off
409 This stream is not running	Start the stream
413	More than 1,000 events in one request
503	The stream's intake is full; send again shortly

Generating a new key stops the old one at once.

Who can do what

Anyone who can open the page sees streams, their status and dead letters. Creating, changing, starting, stopping and deleting streams, generating keys, testing stages and replaying or deleting dead letters need the admin or editor role. Everything except a test is recorded in the audit log.

Troubleshooting

Symptom	Cause	Fix
Stage 1 (filter): Unknown name "value" at position 1 when saving	A field written without event.	Write event.value
A stream shows Feature off	Event Streams is switched off	Turn on event_streams in Settings → Feature Toggles
A stream shows Standby	This appliance is the standby of a redundant pair; the active one runs the streams	Nothing to do. Look at the active appliance
Nothing arrives from a tag	The tag does not change, or its readings are not good quality	Check the tag is live; turn on Include bad and uncertain quality readings if needed
HTTP dead letters with Loopback addresses are refused…	The URL points at a local service on the appliance	Only the appliance's own API port is allowed
HTTP dead letters with This script may not reach the plant network…	Plant network (or Internet) is off on the handler	Switch on the network the destination is on
Script network access is turned off for this appliance	The appliance-wide script network switch is off	Turn it on in Settings → Scripts
Workflow "<name>" is already running and retries	Batches arrive faster than the workflow finishes	Raise the batch size or wait, so one run handles more events
Message "<name>" not sent: it would be message 6…	A stream sends a message that comes back to itself	Use a different message name for the handler

See also

• Integrations: Choosing the Right Interface
• Expression Syntax Reference
• Store & Forward and Data Infrastructure
• Communication Nodes
• Scripting API Reference