Flow ist ein Automatisierungsmenü, mit dem Sie die Datenanbindung an externe Systeme (MES/ERP/SCADA usw.) sowie die automatische Erstellung/Aktualisierung von Domänenobjekten (Tag, Anlage, Arbeitsauftrag, Mitarbeiter usw.) als visuellen Graphen definieren. Ohne Programmierkenntnisse können Operatoren Automatisierungsszenarien selbst entwerfen, bereitstellen und betreiben.

Über 100 Node-Typen werden per Drag & Drop auf der Canvas platziert und über Wires verbunden, um Datenverarbeitungspipelines aufzubauen.

Pfad: Linkes Menü > Automation > Flow

Lernpfad — Empfohlene Einstiegsreihenfolge nach Rolle

Dieses Handbuch ist ein umfassender Leitfaden mit über 4.000 Zeilen. Es ist effizienter, mit dem Abschnitt zu beginnen, der zu Ihrer Rolle und Ihrem Ziel passt.

👶 Einsteiger (ersten Flow in 1 Stunde erstellen)

Kernkonzepte — 5 Minuten
Erste Schritte — 5 Minuten
End-to-End-Tutorial — 30 Minuten (Schritte 1–10 durcharbeiten)
Bildschirmaufbau + Bearbeitungsansicht — 10 Minuten
Beispiel-Flows 1·2·3 nachbauen — 10 Minuten

→ Erster Flow bereitgestellt. Danach benötigte Nodes im Node-Katalog suchen.

🧑‍🏭 Vor-Ort-Operator (Automatisierungsszenarien erstellen)

Payload-Beispiele nach Trigger — reale Datenstrukturen verstehen
Detaillierte Node-Optionsreferenz — häufig genutzte Node-Optionen kennenlernen
Cookbook für Datentransformation — gängige Transformationsmuster kopieren
Katalog der Graph-Muster — Verdrahtungsmuster auswählen
Beispiel-Flows (17 Arten) — fertige Beispiele je Szenario ansehen

🛠 Systemadministrator (Betrieb·Tuning·Störungsbehebung)

Flow-Metriken und Alarme — welche Kennzahlen zu beachten sind
Was man in der Ausführungshistorie sieht — Diagnoseprotokolle interpretieren
Cluster·HA-Verhalten — Multi-Node-Umgebung verstehen
End-to-End-Trace — problematische Nachrichten verfolgen
Performance-Grenzen und Tuning + Notfallmaßnahmen

🔌 Entwickler·Integrationsingenieur (Anbindung externer Systeme)

Cookbook für externe Systemintegration — Slack/Teams/Jira/SAP-Beispiele
Flow-REST-API — Flows programmgesteuert steuern
Webhook-Trigger — Flow von außen auslösen
OPC/PLC-Industrieintegrationsmuster — industrielle Vor-Ort-Szenarien
JS-Laufzeitumgebungsspezifikation + Sammlung wiederverwendbarer Skripte

🔐 Sicherheitsverantwortlicher (Audit·Authentifizierung)

Sicherheit·Umgang mit sensiblen Daten — Aufbewahrung von Zugangsdaten
Automatische Erneuerung externer Auth-Token — Betrieb von OAuth2-Token
Berechtigungen — rollenspezifisch mögliche Aktionen
Audit·Verlaufsverfolgung — Aufbewahrung von Änderungs-/Ausführungshistorie

📚 Schnellreferenz (bereits vertraute Nutzer)

Gesuchte Information	Abschnitt
Node-ID und Einzeilenbeschreibung	Node-Katalog
Standardwerte der Node-Optionen	Detaillierte Node-Optionsreferenz
JSON-Beispiel der Nachricht	Payload-Beispiele nach Trigger
Direkt einsetzbares Skript	Sammlung wiederverwendbarer Skripte · Cookbook für Datentransformation
API-Aufruf curl	Flow-REST-API
Interpretation der Ausführungshistorie	Was man in der Ausführungshistorie sieht
Leitfaden zur Fehlerbehebung	Schrittweiser Debugging-Leitfaden · Häufige Probleme
Schnellübersicht der Optionen	Node-Schnellkonfigurationsreferenz

Alle Abschnittslinks sind Anker innerhalb desselben Dokuments. Auch die Stichwortsuche mit Ctrl+F ist effektiv.

Kernkonzepte

Begriff	Beschreibung
Flow	Ein Automatisierungs-Workflow, bestehend aus einer Menge von Nodes und Wires (Beziehungen). Es handelt sich um einen gerichteten Graphen mit Einstiegspunkt, dessen Aktivierung über den Bereitstellen/Aufheben-Schalter gesteuert wird
Node	Eine Einheit, die eine Nachricht empfängt, verarbeitet und an den nächsten Node weiterleitet. Unterteilt in 7 Kategorien (Trigger·Filter·Transformation·Aktion·Externe Anbindung·Ablaufsteuerung·Edge)
Beziehung (Relation)	Das Label eines Wires von einem Node-Ausgang zum nächsten Node. Der Node bestimmt selbst Labels wie `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` zur Verzweigung. Alle Labels sind auf Großbuchstaben standardisiert
Nachricht (Message)	Die Payload, die innerhalb des Flows fließt. Enthält `type` (Klassifizierung), `originator` (Subjekt-Entität), `data` (Inhalt), `metadata` (Kontext)
Trigger	Der Node, der den Startpunkt eines Flows bildet. Es gibt drei Arten: Domänenereignis (Tag-Point, Alarm, Anlagenereignis usw.), externer Eintritt (Webhook, MQTT, externe DB) und zeitbasiert (Zeitplan)

Erste Schritte

Mit den folgenden 4 Schritten machen Sie sich am schnellsten mit Flow vertraut.

Erstellen Sie in der Listenansicht über die Schaltfläche 새 플로우 einen leeren Flow (nur Name und Beschreibung eingeben).
Ziehen Sie in der Bearbeitungsansicht aus der linken Palette nacheinander einen Trigger-Node (z. B. flow_on_tag_alarm) → Filter → Aktion (z. B. flow_send_email), platzieren Sie diese und verbinden Sie die Nodes mit Wires.
Klicken Sie auf jeden Node, geben Sie im rechten Inspector die Optionen ein und überprüfen Sie das Verhalten anschließend mit Speichern → Testlauf oben rechts.
Aktivieren Sie oben den Bereitstellen-Schalter — der Flow wird bei jedem eintreffenden Triggerereignis automatisch ausgeführt, und Sie können die Ergebnisse im Live-Debug-Panel und in der Ausführungshistorie einsehen.

Für Verdrahtungsmuster von Automatisierungsszenarien lesen Sie bitte Beispiel-Flows und Anwendungsbeispiele in diesem Dokument. Ein schrittweises Tutorial zum vollständigen Erstellen eines Flows von Anfang bis Ende finden Sie im End-to-End-Tutorial.

End-to-End-Tutorial

Ein schrittweises Tutorial, in dem ein tatsächlich in der Produktion einsetzbarer Automatisierungs-Flow von Anfang bis Ende erstellt wird. Szenario: Wenn die Temperatur eines Motors 80 °C überschreitet, wird automatisch ein dringender Instandhaltungsauftrag erstellt und der zuständige Mitarbeiter per E-Mail benachrichtigt.

Schritt 1 — Neuen Flow erstellen

Im linken Menü Automation > Flow klicken → Listenansicht öffnen
Oben rechts auf Neuer Flow klicken
Folgende Angaben eingeben und mit Bestätigen abschließen

Feld	Eingabewert
Name	`모터 과열 자동 정비 발행`
Beschreibung	`[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com`

Nach Erstellung des Flows öffnet sich automatisch die Bearbeitungsansicht mit einer leeren Canvas.

Schritt 2 — Trigger-Node platzieren

Motortemperaturdaten werden aus dem Telemetrie-Ereignis der Anlage empfangen.

Kategorie Trigger in der linken Palette aufklappen
Node flow_on_tag_point auf die Canvas ziehen
Node anklicken → im rechten Inspector folgende Optionen eingeben

Option	Wert
Anzeigename	`태그 포인트 인입`
`tag_id_pattern`	`MOTOR-*.TEMP`

Wenn Sie mit tag_id_pattern nur das Motortemperatur-Tag durchlassen, verringert sich das nachgelagerte Verarbeitungsvolumen erheblich. Andere Tags werden als SKIPPED behandelt und nicht mitgezählt.

Schritt 3 — Schwellenwertfilter

Fügen Sie einen Skriptfilter hinzu, der nur Nachrichten mit über 80 °C durchlässt.

flow_script_filter aus der Kategorie Filter ziehen
Wire vom Ausgangsport des Trigger-Nodes zum Eingangsport des neuen Filter-Nodes verbinden
Im Inspector folgendes eingeben

Option	Wert
Anzeigename	`임계값 필터 (80°C 초과)`
`language`	`JS`
`script`	`data.value > 80`

Schritt 4 — Domänenwechsel (Tag → Anlage)

Da Alarme·Arbeitsaufträge natürlicherweise auf Anlagenebene ausgelöst werden, wird der originator der Nachricht vom Tag auf die übergeordnete Anlage transformiert.

flow_change_originator aus der Kategorie Transformation ziehen
Vom TRUE-Ausgang des Filter-Nodes einen Wire verbinden
Inspector:

Option	Wert
Anzeigename	`Tag → Asset 변경`
`entity_type`	`Asset`
`id_field`	`metadata.asset_id`

metadata.asset_id wird beim Eintreffen eines Tag-Points automatisch befüllt. Falls in der Nachricht nicht vorhanden, können Sie den Präfixteil der Tag-ID (z. B. MOTOR-001.TEMP → MOTOR-001) auch per Skript extrahieren.

Schritt 5 — Arbeitsauftrag erstellen

Ein dringender Instandhaltungsauftrag wird automatisch erstellt.

flow_create_work_order aus der Kategorie Aktion (Action) — Domänen-CRUD ziehen
Vom SUCCESS-Ausgang des Transformations-Nodes einen Wire verbinden
Inspector:

Option	Wert
Anzeigename	`긴급 정비 작업지시 생성`
`asset_id_field`	`originator.id`
`title_field`	(statischer Wert) `긴급 점검 — 모터 과열`
`master_id_field`	(optional) `data.master_id` (wird sonst automatisch erzeugt)
`description_field`	(statischer Wert) `자동 발행: 임계 온도 초과로 긴급 점검 필요`
`default_priority`	`HIGH`

Schritt 6 — E-Mail-Benachrichtigung (Erfolgszweig)

Bei erfolgreicher Erstellung des Arbeitsauftrags wird der zuständige Mitarbeiter benachrichtigt.

flow_send_email aus der Kategorie Externe Anbindung (External) ziehen
Vom SUCCESS-Ausgang des Arbeitsauftrags-Nodes einen Wire verbinden
Inspector:

Option	Wert
Anzeigename	`정비 담당자 이메일`
`to`	`ops@example.com`
`subject_template`	`[과열 정비] ${originator.id} 작업지시 ${data.work_order_id}`
`body_template`	`자산 ${originator.id} 의 온도가 ${data.value}°C 로 상승하여 자동으로 긴급 정비가 발행되었습니다.\n작업지시 ID: ${data.work_order_id}`

Schritt 7 — Behandlung des Fehlerzweigs

Die Erstellung des Arbeitsauftrags selbst kann fehlschlagen (z. B. Anlage nicht mehr vorhanden oder Berechtigungsproblem). Der Operator wird sofort per Push-Benachrichtigung informiert.

flow_send_push aus Externe Anbindung ziehen
Vom FAILURE-Ausgang des Arbeitsauftrags-Nodes einen Wire verbinden
Inspector:

Option	Wert
`title`	`자동화 실패`
`body_template`	`${originator.id} 정비 자동 발행 실패: ${data.error}`

Schritt 8 — Speichern und Testlauf

Oben rechts auf Speichern klicken. Ein automatischer Snapshot wird abgelegt, sodass später ein Rollback möglich ist.
Oben rechts auf Testlauf klicken → folgendes im JSON-Editor eingeben → Veröffentlichen

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data":     { "value": 92.5 },
  "metadata": { "ts": 1746247200000, "tag_id": "MOTOR-001.TEMP", "asset_id": "MOTOR-001" }
}

Unten im Dialog wird ✓ JSON OK (type=POST_TELEMETRY) angezeigt → auf Veröffentlichen klicken.

Im Live-Debug-Panel Folgendes prüfen:
- Trigger-Node leuchtet grün → Nachricht durchgelaufen
- Filter-Node: da 92.5 > 80, Verzweigung TRUE
- Transformations-Node: originator geändert zu Asset/MOTOR-001
- Arbeitsauftrags-Node: SUCCESS, data.work_order_id automatisch vergeben
- E-Mail-Node: Versand versucht

Schritt 9 — Validierung und Bereitstellung

Im Arbeitsauftragsbildschirm prüfen, ob der neue Arbeitsauftrag registriert wurde
Prüfen, ob die E-Mail ordnungsgemäß angekommen ist (Postfach der Testumgebung)
Noch einmal mit einer Nachricht unter 80 °C testen (sollte am Filter blockiert werden):

{ "type": "POST_TELEMETRY", "originator": {"entity_type":"Tag","id":"MOTOR-001.TEMP"},
  "data": {"value": 70}, "metadata": {"asset_id":"MOTOR-001"} }

Die Verzweigung des Filter-Nodes ist FALSE, nachfolgende Nodes grau — normal.

Nach Validierung aller Zweige die Statistiken mit Alle Zähler zurücksetzen neu starten
Oben den Schalter Bereitstellen aktivieren (ON)

Ab jetzt wird in der tatsächlichen Produktionsumgebung sofort bei Überschreiten von 80 °C an einem Motor automatisch ein Instandhaltungsauftrag erstellt und der zuständige Mitarbeiter per E-Mail informiert.

Schritt 10 — Betriebsüberwachung

Es empfiehlt sich, unmittelbar nach der Bereitstellung 5–10 Minuten lang Folgendes zu prüfen.

Ort	Prüfpunkt
Listenansicht	Ob die Anzahl der Ausführungen der betreffenden Flow-Zeile im normalen Bereich steigt (kein Ausufern)
Live-Debug-Panel	Ob keine Node-Fehler auftreten
Ausführungshistorie	Bei fehlgeschlagenen Nachrichten die Ursache über `NODE_ERROR` prüfen
Empfangene E-Mails	Ob keine unbeabsichtigt häufigen Benachrichtigungen versendet werden

Nächste Schritte

Um diesen Flow weiterzuentwickeln, können Sie:

Retry-Verdrahtung hinzufügen — Backoff-Wiederholung bei fehlgeschlagenem E-Mail-/Push-Versand (siehe Beispiel 8)
Automatische Bandanpassung — Schwellenwerte automatisch anhand des 6-Stunden-Durchschnitts ± 3σ korrigieren (Beispiel 6)
Ausfallzeit-Kumulierung — Überhitzungshistorie in Anlagen-Attributen kumulieren (Beispiel 14)
Qualitätslinien-Isolation — Linie bei wiederholter Überhitzung automatisch stoppen (Beispiel 15)

Bildschirmaufbau

Flow besteht aus den folgenden drei Bildschirmen.

Bildschirm	Zweck
Liste	Übersicht·Suche·Massenbereitstellung·Import der registrierten Flows
Bearbeitung	Nodes über visuelle Canvas platzieren·verbinden·konfigurieren
Ausführungshistorie	Node-bezogene Ausführungsprotokolle·Zeitachsenabfrage

Listenansicht

Besteht aus dem oberen Such-/Erstellungsbereich und der Flow-Übersichtstabelle.

Werkzeuge oben

Element	Beschreibung
Statusfilter	`전체` / `배포` / `해제`
Name·Beschreibung durchsuchen	Flows nach Schlüsselwort filtern
Neuer Flow	Erstellt einen leeren Flow (Name·Beschreibung eingeben)
Importieren	Stellt einen Flow durch Hochladen einer per Export erhaltenen JSON-Datei wieder her
Alle neu bereitstellen	Lädt alle aktivierten Flows auf einmal neu
Aktualisieren	Lädt die Liste erneut

Übersichtstabelle

Die Listentabelle ist auf 25 Zeilen pro Seite paginiert, und jede Zeile zeigt zusätzlich eine Verlaufs-Sparkline sowie ein Fehlerraten-Donut-Diagramm. Der Diagrammstatus bleibt auch beim Seitenwechsel erhalten.

Spalte	Beschreibung
Auswahl	Checkbox für Massenbereitstellung·-aufhebung
Status	Badge `배포` / `해제`
Flow-ID	Sequenz-ID im Format `FLOW_NNNNN`
Name \| Beschreibung	Vom Operator festgelegte Metainformationen
Nodes	Anzahl enthaltener Nodes
Trigger	Anzahl der Trigger-Nodes
Ausführungen	Kumulierte Anzahl verarbeiteter Nachrichten
Verarbeitungszeit	Durchschnittliche/letzte Verarbeitungszeit der Nodes
Fehler	Kumulierte Fehleranzahl
Letzte Änderung	Zeitpunkt der letzten Speicherung des Graphen
Aktion	Schaltflächen `편집` / `삭제`

Massenbereitstellung·-aufhebung

Die ausgewählten Flows werden gemeinsam bereitgestellt (aktiviert) oder aufgehoben (deaktiviert). Nur bereitgestellte Flows empfangen Triggerereignisse.

Alle neu bereitstellen

Die Schaltfläche Alle neu bereitstellen oben lädt alle aktivierten Flows neu. Verwenden Sie sie in folgenden Situationen:

Unmittelbar nach einem Massenimport von Graphen aus externen Quellen
Wenn Sie Trigger mit eigenem Scheduler (Zeitplan·externes MQTT-Abonnement·externes DB-Polling) neu registrieren möchten
Bei Verdacht auf Probleme mit der Cache-Konsistenz im laufenden Betrieb

Bearbeitungsansicht (visuelle Canvas)

Ziehen Sie Nodes aus der linken Palette der Canvas, um sie zu platzieren, und verbinden Sie sie durch Klicken & Ziehen des Ausgangsports mit dem nächsten Node.

Werkzeugleiste oben

Schaltfläche	Aktion
Name·Beschreibung	Bearbeitet die Metainformationen des Flows
Bereitstellen/Aufheben	Aktiviert/deaktiviert den aktuellen Flow sofort
Exportieren	Lädt den gesamten Graphen als JSON-Datei herunter
Testlauf	Führt eine einmalige Ausführung mit einer beliebigen injizierten JSON-Nachricht durch und zeigt das Ergebnis (siehe Testlauf-Nutzung unten)
Speichern	Speichert den aktuellen Graphen auf dem Server. Beim Speichern wird ein automatischer Snapshot abgelegt, sodass später ein Rollback möglich ist

Beim Versuch, einen Graphen ohne Trigger zu speichern, erscheint die Warnung „Wird nur durch manuellen Testlauf gestartet“. Sie können den Trigger absichtlich weglassen und den Flow ausschließlich für manuelle Ausführung nutzen.

Testlauf-Nutzung

Beim Klicken auf die Schaltfläche 테스트 실행 öffnet sich ein JSON-Bearbeitungsdialog. Der Operator kann direkt eine Nachricht verfassen und einmalig veröffentlichen, um das Graphenverhalten zu überprüfen, ohne auf ein Triggerereignis zu warten.

Element	Beschreibung
Editor	JSON-Editor mit Zeilennummern und Syntaxhervorhebung. Der Nachrichteninhalt kann frei verfasst werden
Validierungsanzeige	`type` bei vorhandenen Pflichtfeldern, `✓ JSON OK (type=X)`; bei fehlenden Feldern `⚠ type 필수`
Veröffentlichen	Mit der Schaltfläche `발행` wird die Nachricht in den Dispatcher injiziert. Ergebnisse im Live-Debug-Panel und in der Ausführungshistorie einsehbar

Beispiel-Standardvorlage

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data":     { "speed": 1500, "temp": 75.3 },
  "metadata": { "ts": 1746247200000, "site_id": "SITE-01" }
}

Wenn Sie bei nicht gespeicherten Änderungen auf Testlauf klicken, erscheint der Hinweis „Der Server führt die gespeicherte Version aus“. Speichern Sie zuerst, um Änderungen zu validieren.

Links — Node-Palette

Kategorien können auf- und zugeklappt werden, und über das Suchfeld ist eine sofortige Filterung möglich.

Kategorie	Farbe	Anzahl Nodes
Trigger	Grau	22 Arten
Filter	Blau	6 Arten
Transformation	Grün	8 Arten
Aktion (Action) — Integration·Speicherung·Anlagenpublikation·Befehlsaufruf	Orange	7 Arten
Aktion (Action) — Domänen-CRUD	Orange	34 Arten
Externe Anbindung (External)	Lila	9 Arten
Ablaufsteuerung (Control)	Grau	7 Arten
Edge	Türkis	22 Arten

Mitte — Canvas

Werkzeug	Tastenkombination / Bedienung	Aktion
Vergrößern/Verkleinern	`Ctrl/⌘ +` / `Ctrl/⌘ -` · Mausrad	Canvas-Zoom
100%	`Ctrl/⌘ 0`	Zoom zurücksetzen
An Bildschirm anpassen	`Ctrl/⌘ 1`	Automatische Anpassung, sodass alle Nodes sichtbar sind
Ausgewählte Nodes löschen	`Del` / `Backspace`	Löscht die ausgewählten Nodes/Wires
Canvas verschieben (Panning)	Linksklick auf leeren Bereich und ziehen	Wenn Sie den leeren Hintergrund der Canvas fassen und ziehen, ohne einen Node anzuklicken, bewegt sich die gesamte Canvas mit

Unten rechts auf der Canvas wird eine Minimap angezeigt. Ein Klick auf die Minimap ermöglicht eine sofortige Navigation zur entsprechenden Position.

💡 Tipp zum Panning: Ziehen auf einem Node bewegt den Node — um die Canvas zu verschieben, müssen Sie unbedingt einen leeren Hintergrundbereich ohne Nodes/Wires fassen. Bei großen Flows ist Panning oft schneller als die Minimap.

Rechts — Node-Konfiguration

Wenn Sie auf der Canvas einen Node anklicken, wird im rechten Inspector das Konfigurationsformular für diesen Node angezeigt. Die Eingabefelder werden je nach Node-Typ automatisch generiert.

Eingabemethode	Beschreibung
Statischer Wert	Verwendet den direkt im Formular eingegebenen Wert unverändert
*`_field` dynamischer Wert**	Extrahiert den Wert aus einem Pfad der Nachrichten-Payload (z. B. `data.tag_id`, `metadata.site_id`). Fällt bei fehlendem Wert auf den statischen Wert zurück

Skript-Nodes (Filter·Transformation·Switch) können direkt im Inspector über den Code-Editor bearbeitet werden und lassen sich auch in einem separaten Dialog vergrößern, um auf einem großen Bildschirm zu arbeiten.

Die Schriftart des Code-Editors verwendet einen lesbarkeitsoptimierten Monospace-Font-Stack (Priorität: Cascadia Code · JetBrains Mono · Consolas · Menlo). Auch koreanische Kommentare werden stabil ausgerichtet dargestellt.

Zähler zurücksetzen

Oben rechts im Node-Konfigurationsbereich werden zwei Zurücksetzen-Schaltflächen als Icons angezeigt. Beim Überfahren mit der Maus erscheint zu jeder Schaltfläche ein Tooltip.

Icon-Schaltfläche	Aktion
🩹 (Pflaster) — Fehlerzähler zurücksetzen	Setzt nur den kumulierten Fehlerzähler aller Nodes in diesem Flow auf 0 zurück
🔄 (Kreisförmiger Pfeil) — Alle Zähler zurücksetzen	Setzt Verarbeitung·Fehler·Verarbeitungszeit sowie alle flow-weiten Statistiken auf 0 zurück. Nutzen Sie diese Funktion, wenn Sie die Betriebsvalidierung abgeschlossen haben und die Statistik neu beginnen möchten

Beide Schaltflächen wirken sofort ohne Bestätigungsdialog — es werden nur die Statistiken zurückgesetzt, das Node-Verhalten und die Nachrichtenverarbeitung sind davon nicht betroffen.

Rechts — Live-Debug

Das Live-Debug-Panel wird unterhalb des Inspectors angezeigt.

Element	Beschreibung
Aktualisierungsintervall	2 Sekunden
Farbcodierung der Ebene	INFO (blau)/WARN (gelb)/ERROR (rot) am linken Rand
Angezeigte Informationen	Node-Anzeigename · Verarbeitungszeit (ms) · Nachrichtenvorschau
Pausieren	Pausiert die Aktualisierung über den Schalter oben rechts im Panel
Leeren	Leert die kumulierten Debug-Einträge nur auf dem Bildschirm

Oben rechts am Node auf der Canvas wird eine kleine Statusanzeige (LED) angezeigt.

Farbe	Bedeutung
Grau	Wartend — keine Nachricht empfangen
Grün	Nachricht wird durchgeleitet
Rot	Fehler bei der Verarbeitung aufgetreten

Unten rechts am Node wird die Verarbeitungszeit in der Form 평균 X · 최근 Y angezeigt.

Nachrichtenstruktur

Die innerhalb des Flows fließende Nachricht besteht aus den folgenden 4 Bereichen.

{
  "type": "POST_TELEMETRY",
  "originator": {
    "entity_type": "Asset",
    "id": "MOTOR-001"
  },
  "data": { "speed": 1500, "temp": 75.3 },
  "metadata": {
    "ts": 1746247200000,
    "site_id": "SITE-01",
    "shift": "DAY",
    "tag_id": "MOTOR-001.SPEED"
  }
}

Bereich	Bedeutung
`type`	Nachrichtenklassifizierung. Verzweigungskriterium des Filter-Nodes
`originator`	Subjekt-Entität der Nachricht (auf welche Anlage/welches Tag/welche Bestellung bezogen)
`data`	Payload-Inhalt
`metadata`	Kontext (Zeit·Standort·Schicht·Tag-ID usw.)

Nachrichtentypen

Typ	Auslösender Einstiegspunkt
`POST_TELEMETRY` / `TAG_POINT`	Eingang eines Tag-Points
`POST_ATTRIBUTES`	Aktualisierung von Tag-/Anlagen-Metadaten
`TAG_ALARM`	Tag-bezogener Alarm
`ENTITY_CREATED` / `UPDATED` / `DELETED`	Entitäts-Lebenszyklusereignis
`ASSET_DATA` / `ASSET_EVENT` / `ASSET_ALARM` / `ASSET_COMMAND` / `ASSET_AGGREGATION` / `ASSET_CONTEXT`	Anlagen-Domänenereignis
`ASSET_HEALTH_STATUS` / `ASSET_CONNECTION_STATUS`	Periodische Anlagenbewertung (Zustand/Verbindungsstatus)
`OEE_EVENT` / `RAM_EVENT` / `EMS_EVENT`	ISO-Analyseergebnisereignis
`OPC_STATUS` / `EDGE_STATUS`	OPC-/Edge-Gerätestatus
`DIAGNOSTIC` / `DOMAIN_CHANGED`	Diagnose·Domänenänderung
`ALARM`	Alarmauslösung
`WEBHOOK`	HTTP-Webhook-Empfang
`KAFKA_INBOUND` / `MQTT_INBOUND`	Empfang aus externem Topic
`TIMER`	Zeitplan-Auslösung

Payload-Beispiele nach Trigger

Beim Schreiben von Skript-Nodes müssen Sie genau wissen, auf welche Felder Sie zugreifen können. Nachfolgend finden Sie tatsächliche Nachrichten-JSON-Beispiele, die von jedem Trigger erzeugt werden. Alle Trigger-Nachrichten enthalten gemeinsam type, originator, data, metadata.

`flow_on_tag_point` — Eingang eines Tag-Points

Wird bei jedem eingehenden Wert eines einzelnen Tags ausgelöst. Der häufigste Trigger.

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data": {
    "value": 1500.7,
    "quality": "GOOD",
    "ts": 1746247200123
  },
  "metadata": {
    "tag_id":      "MOTOR-001.SPEED",
    "site_id":    "SITE-01",
    "area_id":    "AREA-A",
    "line_id":    "LINE-1",
    "asset_id":   "MOTOR-001",
    "opc_id":     "OPC-LINE-1",
    "java_type":  "Float",
    "unit":       "rpm",
    "shift":      "DAY"
  }
}

Feld	Bedeutung	Skriptzugriff
`data.value`	Empfangener Wert (numerisch/Text/boolesch)	`msg.data.value`
`data.quality`	OPC-Qualität (GOOD/BAD/UNCERTAIN)	`msg.data.quality`
`data.ts`	Empfangszeitpunkt (Epoch ms)	`msg.data.ts`
`metadata.tag_id`	Tag-ID	`msg.metadata.tag_id`

`flow_on_tag_alarm` — Auftreten eines Tag-Alarms

Wird ausgelöst, sobald das Alarmband (hi/lo/...) eines Tags überschritten wird.

{
  "type": "TAG_ALARM",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data": {
    "alarm_band": "HI_HI",
    "value":      95.3,
    "threshold":  90.0,
    "priority":   "ERROR",
    "band_message": "온도 위험"
  },
  "metadata": {
    "tag_id": "MOTOR-001.TEMP",
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

Wert von `data.alarm_band`	Bedeutung
`NORMAL` / `HI` / `LO` / `HI_HI` / `LO_LO` / `TRIP_HI` / `TRIP_LO`	Numerische Alarmstufe
`BOOL_TRUE` / `BOOL_FALSE`	Boolescher Alarm

`flow_on_asset_data` / `flow_on_asset_event` — Anlagenereignis

Wird ausgelöst, wenn ein auf Anlagenebene aggregiertes Ereignis (CEP-Verarbeitungsergebnis·Plugin-Bewertungsergebnis usw.) auftritt.

{
  "type": "ASSET_EVENT",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "event_type":  "STARTUP",
    "details":     { "rpm_target": 1500 }
  },
  "metadata": {
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

flow_on_asset_data überträgt zeitreihenbasierte Daten auf Anlagenebene (data.values als Key-Value-Map), flow_on_asset_aggregation überträgt Minuten-/Stundenaggregate.

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — Periodische Bewertung

Die Plattform bewertet im 1-Minuten-Intervall den Zustand/Verbindungsstatus auf Anlagenebene.

{
  "type": "ASSET_HEALTH_STATUS",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "status":       "WARN",
    "info_count":   12,
    "warn_count":   3,
    "error_count":  0,
    "prev_status":  "OK"
  },
  "metadata": { "asset_id": "MOTOR-001", "ts": 1746247200123 }
}

Wert von `data.status`	Bedeutung
`OK` / `WARN` / `ERROR` / `UNKNOWN`	Zustandsstufe
`CONNECTED` / `LATENT` / `ERROR` / `DISCONNECTED` / `UNKNOWN`	Verbindungsstufe (connection_status)

Es wird häufig das Muster verwendet, im Vergleich mit prev_status nachfolgende Aktionen nur zum Zeitpunkt des Statusübergangs auszulösen.

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — Plugin-Ereignis

Wird ausgelöst, wenn ein OEE/RAM/EMS-Bewertungsergebnis auf Arbeitsauftragsebene aktualisiert wird.

{
  "type": "OEE_EVENT",
  "originator": { "entity_type": "WorkOrder", "id": "WO-20260512-001" },
  "data": {
    "oee":          0.78,
    "availability": 0.95,
    "performance":  0.85,
    "quality":      0.97,
    "good_count":   1560,
    "bad_count":    42,
    "target_count": 2000
  },
  "metadata": {
    "order_id":   "WO-20260512-001",
    "asset_id":   "LINE-1.PRESS",
    "shift_id":   "DAY-A",
    "ts": 1746247200123
  }
}

`flow_on_opc_status` / `flow_on_edge_status` — OPC-/Edge-Status

{
  "type": "OPC_STATUS",
  "originator": { "entity_type": "OPC", "id": "OPC-LINE-1" },
  "data": {
    "connection_status": "CONNECTED",
    "scan_status":       "START",
    "prev_status":       "DISCONNECTED"
  },
  "metadata": { "opc_id": "OPC-LINE-1", "edge_id": "EDGE-A", "ts": 1746247200123 }
}

`flow_on_diagnostic` — System-Diagnosemeldung

Wird ausgelöst, wenn eine Diagnosemeldung aus einem Servermodul eingeht.

{
  "type": "DIAGNOSTIC",
  "originator": { "entity_type": "Module", "id": "cep-engine" },
  "data": {
    "level":   "WARN",
    "code":    "PATTERN_LAG",
    "summary": "EQL 패턴 평가 지연 1.2s",
    "module":  "cep-engine"
  },
  "metadata": { "ts": 1746247200123 }
}

`data.level`	Bedeutung
`INFO` / `WARN` / `ERROR`	Diagnoseschweregrad

`flow_on_domain_changed` — Domänenänderungsereignis

Wird ausgelöst, wenn Domänenentitäten wie Anlage·Tag·Standort·Arbeitsauftrag usw. per CRUD verändert werden.

{
  "type": "DOMAIN_CHANGED",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "action":      "UPDATED",
    "before":      { "asset_name": "Motor1" },
    "after":       { "asset_name": "Motor 01 - Renamed" },
    "changed_by":  "admin"
  },
  "metadata": { "ts": 1746247200123 }
}

`flow_on_entity_event` — Entitäts-Lebenszyklus

Integriert für die drei Typen ENTITY_CREATED / ENTITY_UPDATED / ENTITY_DELETED ausgelöst. Ein einzelner Node empfängt alle drei Lebenszyklen.

{
  "type": "ENTITY_CREATED",
  "originator": { "entity_type": "Customer", "id": "CUST-9001" },
  "data": { "customer_name": "신규 고객", "external_id": "ERP-CUST-9001" },
  "metadata": { "ts": 1746247200123 }
}

`flow_on_webhook` — Externer HTTP-Push

Die von einem externen System an POST /api/v4/flow/webhook/{flow_id} gesendete Payload wird unverändert in eine Nachricht umgewandelt. Authentifizierung über {flow_id} im URL-Pfad und Header X-API-Key.

{
  "type": "WEBHOOK",
  "originator": { "entity_type": "External", "id": "ERP" },
  "data": {
    "order_no":  "PO-20260512-001",
    "customer":  "ACME",
    "quantity":  1000
  },
  "metadata": {
    "http_method": "POST",
    "remote_addr": "10.20.0.55",
    "request_id":  "req-7c0a...",
    "ts": 1746247200123
  }
}

Der gesamte vom externen System gesendete JSON-Body wird unverändert in data eingefügt. HTTP-Header werden teilweise (remote_addr/method/request_id) in metadata angezeigt.

`flow_on_mqtt_subscribe` — MQTT-Topic-Abonnement

{
  "type": "MQTT_INBOUND",
  "originator": { "entity_type": "Topic", "id": "factory/line1/events" },
  "data": { "event": "STARTUP", "rpm": 1500 },
  "metadata": {
    "topic":   "factory/line1/events",
    "qos":     1,
    "broker":  "tcp://mqtt.example.com:1883",
    "ts": 1746247200123
  }
}

`flow_jdbc_poll` — Periodisches externes DB-Polling

Führt eine konfigurierte SELECT-Abfrage periodisch aus und löst für jede Zeile eine Nachricht aus.

{
  "type": "KAFKA_INBOUND",
  "originator": { "entity_type": "DB", "id": "mes_db" },
  "data": {
    "PO_NO":    "PO-20260512-001",
    "CUSTOMER": "ACME",
    "QTY":      1000,
    "DUE_DATE": "2026-05-20"
  },
  "metadata": {
    "datasource": "mes_db",
    "query":      "SELECT * FROM po WHERE status='NEW'",
    "row_index":  0,
    "ts": 1746247200123
  }
}

Bei 100 Zeilen werden 100 Nachrichten nacheinander ausgelöst. Um zu vermeiden, dass dieselbe Zeile wiederholt verarbeitet wird, aktualisieren Sie innerhalb der SELECT-Abfrage entsprechend ein Verarbeitungsflag oder fügen Sie eine Vergleichsbedingung für die Spalte processed_at hinzu.

`flow_schedule` — Zeitbasierte Auslösung

Wird über einen Cron-Ausdruck oder ein festes Intervall ausgelöst. Die Payload ist leer, nur metadata.ts wird befüllt.

{
  "type": "TIMER",
  "originator": { "entity_type": "Schedule", "id": "daily-report" },
  "data": {},
  "metadata": {
    "cron":      "0 0 8 * * ?",
    "fired_at":  1746247200000,
    "ts":        1746247200000
  }
}

Hinweise bei der Payload-Transformation

originator.id ist die Domänen-ID — wenn Sie sie mit dem flow_change_originator-Node ändern, arbeiten nachfolgende Aktions-Nodes wie flow_save_attributes·flow_publish_asset_* mit dem neuen originator.
Beim Austauschen von data/metadata per Skript wird eine direkte Zuweisung anstelle einer flachen Kopie (Object.assign) empfohlen — eine Änderung am Original kann andere Flows beeinflussen, die denselben Trigger abonnieren.
Alle Epoch-Zeitfelder sind in Millisekunden (ms). Falls Sekunden benötigt werden, Math.floor(msg.metadata.ts / 1000).

Node-Katalog

Detaillierte Node-IDs und Optionen können Sie im Inspector der Bearbeitungsansicht einsehen.

Trigger (22 Arten)

Der Startpunkt jedes Flows ist ein Trigger-Node. Es gibt keine separaten Einstiegs-/Endpunkt-Nodes.

Automatischer Domänenempfang — Interne Systemereignisse werden automatisch dispatcht.

Kategorie	Trigger-Node
Tag	`flow_on_tag_point` (Telemetrie), `flow_on_tag_alarm`
Anlage	`flow_on_asset_data`, `flow_on_asset_event`, `flow_on_asset_alarm`, `flow_on_asset_command`, `flow_on_asset_aggregation`, `flow_on_asset_context`, `flow_on_asset_health_status`, `flow_on_asset_connection_status`
Plugin	`flow_on_oee_event`, `flow_on_ram_event`, `flow_on_ems_event`
OPC/Edge	`flow_on_opc_status`, `flow_on_edge_status`
Diagnose·Domäne	`flow_on_diagnostic`, `flow_on_domain_changed`
Entität	`flow_on_entity_event`

Einen Node namens flow_on_alarm gibt es nicht

Alarm-Trigger sind je nach Ziel in zwei getrennt — Tag-Alarm ist flow_on_tag_alarm, Anlagen-Alarm ist flow_on_asset_alarm. Ein Beispiel aus einem älteren Dokument verwendete flow_on_alarm, ein Name, der nicht in der Palette vorhanden ist — wenn Sie es genauso nachahmen, ist der Node nicht auffindbar.

Alle Trigger-Nodes ermöglichen mit der Option *_pattern (Glob: *, ?) eine nachrichtenweise Vorabfilterung. Nicht dem Muster entsprechende Nachrichten werden nicht an nachfolgende Nodes weitergegeben, und auch der Ausführungszähler erhöht sich nicht (SKIPPED-Behandlung). Es wird empfohlen, bereits in der Triggerstufe zu filtern, um die Betriebslast zu minimieren.

Externer Eintritt

Node	Aktion
`flow_on_webhook`	Empfängt eine von einem externen System per HTTP gepushte Payload
`flow_on_mqtt_subscribe`	Abonniert ein Topic eines externen MQTT-Brokers
`flow_jdbc_poll`	Liest periodisch das SELECT-Ergebnis einer externen Datenbank und löst pro Zeile aus

Zeitbasiert

Node	Aktion
`flow_schedule`	Cron-/Zeitplan — löst die Nachricht `TIMER` aus

Filter (6 Arten)

Node	Beschreibung
`flow_msg_type_filter`	Wenn `type` in der angegebenen Liste enthalten ist, `TRUE`
`flow_originator_type_filter`	Wenn `originator.entity_type` in der angegebenen Liste enthalten ist, `TRUE`
`flow_script_filter`	Boolesche Auswertung per Skript
`flow_check_existence_field`	Prüft, ob ein bestimmtes Feld `data`/`metadata` existiert
`flow_switch`	Verzweigung mit mehreren Cases (jeder Case eine eigene Relation)
`flow_check_relation`	Verzweigung basierend auf der Relation der vorherigen Stufe

Transformation (8 Arten)

Node	Anzeigename	Beschreibung
`flow_script_transform`	Skript-Transformation	Transformiert `data`/`metadata` per Skript
`flow_change_originator`	Absender ändern	Ändert `originator` zu einer anderen Entität
`flow_rename_keys`	Schlüsselnamen ändern	Ändert Feldnamen von `data` in großer Menge
`flow_template`	Vorlage	Erzeugt Ersatztext auf Basis von `${path}`
`flow_split`	Teilen	Wenn `data` ein Array ist, wird pro Element eine Nachricht erzeugt
`flow_merge`	Zusammenführen	Führt mehrere Nachrichten innerhalb eines Zeitfensters zusammen — Gegenteil von `flow_split`
`flow_flatten`	Abflachen	Hebt untergeordnete Schlüssel eines verschachtelten Objekts auf die oberste Ebene (`data.data_map.x` → `data.x`)
`flow_to_email`	E-Mail-Transformation	Wandelt die Nachricht in ein E-Mail-Format um

flow_flatten wird verwendet, wenn Sie bei Nachrichten wie Tag-Points, bei denen der Wert eine Ebene tiefer in data_map liegt, im nachfolgenden Node direkt über ${data.x} darauf verweisen möchten.

Aktion — Integration·Speicherung (2 Arten)

Node	Beschreibung
`flow_save_tag_point`	Lädt Tag-Points (identischer Pfad wie beim normalen Ingest)
`flow_save_attributes`	Teilweise Aktualisierung von Tag-/Anlagen-Metadaten

Wenn Sie flow_dds_publish suchen (freie Publikation auf den internen Nachrichtenkanal), finden Sie es nicht hier, sondern in der Palettengruppe Externe Anbindung.

Aktion — Anlagenereignis-Publikation (4 Arten)

Publiziert Anlagenereignisse über denselben Verarbeitungspfad wie CEP (EQL) (konsistente Verarbeitung von persistenter Speicherung + Cache + Timeline + Plugin + Nachrichtenkanal).

Node	Kanal
`flow_publish_asset_event`	Anlagenereignis
`flow_publish_asset_context`	Anlagenkontext
`flow_publish_asset_aggregation`	Anlagenaggregation
`flow_publish_asset_command`	Anlagenbefehl

`flow_asset_command_invoke` — Ausführung statt Publikation

Node	Anzeigename	Aufgabe
`flow_asset_command_invoke`	Anlagenbefehlsaufruf	Führt einen an der Anlage definierten Befehl synchron aus und wartet auf das Ergebnis

Leicht mit flow_publish_asset_command zu verwechseln

flow_publish_asset_command — Publiziert nur ein Anlagen-Befehlsereignis auf den Kanal. Damit ist es beendet.
flow_asset_command_invoke — Führt einen an der Anlage definierten Befehl tatsächlich aus und wartet, bis das Ergebnis vorliegt.

Wenn Sie eigentlich wollten, dass der Befehl ausgeführt wird, aber den Publikations-Node verwenden, sieht es so aus, als würde nichts passieren.

Aktion — Domänen-CRUD (34 Arten)

Alle Domänenoperationen werden an denselben Domänenservice delegiert, wodurch Audit·Konsistenz gewahrt bleiben. Mit der dynamischen Option *_field können Werte aus der Nachrichten-Payload extrahiert werden.

Domäne	Create	Update	Delete
Anlage (Asset)	`flow_create_asset`	`flow_update_asset`	`flow_delete_asset`
Tag	`flow_create_tag`	`flow_update_tag`	`flow_delete_tag`
Standort/Bereich/Linie	`flow_create_site`	`flow_update_site`	`flow_delete_site`
Arbeitsauftrag (WorkOrder)	`flow_create_work_order`	`flow_update_work_order`	`flow_delete_work_order`
Alarmkonfiguration (EQL)	`flow_create_alarm_config`	`flow_update_alarm_config`	`flow_delete_alarm_config`
Kunde (Customer)	`flow_create_customer`	`flow_update_customer`	`flow_delete_customer`
Produkt (Product)	`flow_create_product`	`flow_update_product`	`flow_delete_product`
Mitarbeiter (Employee)	`flow_create_employee`	`flow_update_employee`	`flow_delete_employee`
Kalender (Schicht)	`flow_create_calendar`	`flow_update_calendar`	`flow_delete_calendar`

Teilweise Aktualisierung des Tag-Alarmbands (2 Arten)

Node	Beschreibung
`flow_update_tag_alarm_band_numeric`	Aktualisiert nur die eingegebenen Felder eines numerischen Alarmbands (`hi`/`lo`/`hi_hi`/`lo_lo`/`trip_hi`/`trip_lo`/`band_message`/`use_alarm`)
`flow_update_tag_alarm_band_boolean`	Aktualisiert nur die eingegebenen Felder eines booleschen Alarmbands (`bool_true`/`bool_false`/Priorität/Nachricht/`use_alarm`)

Statusübergänge des Arbeitsauftrags (WorkOrder) (5 Arten)

Da nicht direkt numerische Spalten aktualisiert, sondern die Statusübergangsmethoden des Domänenservices aufgerufen werden, bleibt die OEE/RAM/EMS-Sichtbarkeit erhalten.

Node	Übergang
`flow_start_work_order`	`WAIT` → `START`
`flow_pause_work_order`	`START` → `PAUSED`
`flow_resume_work_order`	`PAUSED` → `START`
`flow_end_work_order`	`START` oder `PAUSED` → `END`
`flow_abort_work_order`	`START` oder `PAUSED` → `ABORTED` (Optionen `abort_code`, `notes`)

Automatische NOT-NULL-Ergänzung — Create-Nodes befüllen NOT-NULL-Spalten automatisch mit Default-Werten. Beispiel: Beim Arbeitsauftrag sind status="WAIT" / master_id die MES-Master-ID (bei Fehlen automatisch nachgefüllt), Kundenmanager-Informationen "admin"/"admin@example.com", Mitarbeiter org_id fällt auf site_id zurück, insert_user_id="flow" für alle Zeilen. Bei FK-Spalten (z. B. customer_id/product_id) wird ein leerer String in NULL umgewandelt.

Update-Node — Teilaktualisierung — Die Update-Nodes für Customer/Product/Employee/Calendar und Alarmband fragen zunächst den bestehenden Datensatz ab und speichern dann nur die eingegebenen Felder zusammengeführt. Leere Strings·null-Werte werden ignoriert, sodass der bestehende Wert erhalten bleibt. Für eine vollständige Überschreibung verwenden Sie die Kombination Delete + Create.

Direkter Alarm-Trigger-Node ist bewusst ausgeschlossen. Alarme dürfen nur über den flow_create_alarm_config-Pfad entstehen, damit die Konsistenz der Alarmhistorie gewahrt bleibt.

Edge (22 Arten)

Automatisiert die Registrierung von OPC-Servern, Tag-CRUD, Lesen/Schreiben von Tag-Werten, Abfragen und die Steuerung von Docker-Apps durch Aufruf der REST-API des Edge-Geräts (OPC Agent). Alle Nodes teilen sich semantisch dieselbe Aktion, unterscheiden sich nur in der zugrunde liegenden HTTP-Methode (GET/POST/PUT/DELETE).

Gruppe	Node
OPC-Server-Verwaltung	`flow_edge_opc_create`, `flow_edge_opc_update`, `flow_edge_opc_delete`, `flow_edge_opc_start`, `flow_edge_opc_stop`, `flow_edge_opc_list`
Tag-Verwaltung	`flow_edge_tag_create`, `flow_edge_tag_update`, `flow_edge_tag_delete`, `flow_edge_tag_read`, `flow_edge_tag_write`, `flow_edge_tag_list`
Abfrage	`flow_edge_monitoring`, `flow_edge_info`, `flow_edge_transfer`
Docker-App-Steuerung	`flow_edge_app_list`, `flow_edge_app_inspect`, `flow_edge_app_start`, `flow_edge_app_stop`, `flow_edge_app_restart`, `flow_edge_app_logs`, `flow_edge_app_stats`

3 Abfrage-Arten

Node	Anzeigename	Aufgabe
`flow_edge_monitoring`	Monitoring	Fragt Edge-Monitoring-Kennzahlen ab
`flow_edge_info`	Edge-Info	Fragt Edge-ID · Version · Uptime ab
`flow_edge_transfer`	Übertragungsstatus	Status der MQTT/Sparkplug-Übertragung

Node	Anzeigename	Aufgabe
`flow_edge_opc_list`	OPC-Liste	Fragt die Liste der OPC-Server des Edge ab
`flow_edge_tag_list`	Tag-Liste	Fragt die Tag-Liste des OPC ab

7 Arten der Docker-App-Steuerung

Automatisiert per Flow, was zuvor manuell im Docker-Panel der Edge-Detailansicht erledigt wurde.

Node	Anzeigename	Aufgabe
`flow_edge_app_list`	App-Liste	Liste der Edge-Docker-Container
`flow_edge_app_inspect`	App-Details	Container-Details (inspect)
`flow_edge_app_start`	App starten	Container starten
`flow_edge_app_stop`	App stoppen	Container stoppen
`flow_edge_app_restart`	App neu starten	Container neu starten
`flow_edge_app_logs`	App-Log	Container-Log (`line=N`)
`flow_edge_app_stats`	App-Statistik	Container CPU / Speicher / I/O

App-Steuerungs-Nodes greifen real auf Vor-Ort-Geräte zu

app_stop · app_restart stoppen und starten tatsächlich Container, die auf dem Edge vor Ort laufen. Wenn Sie dies durch Anhängen an einen Trigger automatisch ausführen lassen, grenzen Sie die Bedingung eng ein — wenn Sie app_restart an einen flatternden Alarm hängen, startet der Container immer wieder neu.

Gemeinsame Konfiguration

Option	Beschreibung
`url`	Edge-REST-Endpunkt. Unterstützt `${data.x}`/`${metadata.y}`-Vorlagenersetzung
`method`	HTTP-Methode (falls nicht gesetzt, Node-spezifischer Standardwert — z. B. create=POST, update=PUT, delete=DELETE, read/monitoring=GET)
`headers`	JSON-Header (z. B. `{"Authorization":"Bearer ${TOKEN}"}`)
`body_template`	Request-Body (falls nicht gesetzt, wird `data` unverändert gesendet, bei GET/DELETE kein Body gesendet)
`timeout_ms`	Timeout (Standard 5000)

Antwort·Verzweigung

data.response_status — HTTP-Statuscode
data.response — Antwortkörper (String)
data.error — Fehlermeldung (bei Fehlschlag)
SUCCESS (200–399) / FAILURE (sonst oder bei Exception)

Externe Anbindung (9 Arten)

Alle External-Nodes unterstützen die dynamische Option *_field.

Node	Dynamische Option
`flow_dds_publish`	Freie Publikation auf den internen Nachrichtenkanal (kein externer Aufruf, aber in dieser Gruppe)
`flow_http_request`	`url_field` / `method_field` / `body_field`
`flow_kafka_publish`	`topic_field` / `key_field`
`flow_mqtt_publish`	`topic_field`
`flow_webhook_callback`	`url_field`
`flow_send_email`	`to_field` / `cc_field` / `subject_field` / `body_field`
`flow_send_sms`	`to_field` / `text_field`
`flow_send_push`	`title_field` / `body_field`
`flow_jdbc_query`	SQL statisch (SELECT/INSERT/UPDATE/DELETE)

Ablaufsteuerung (7 Arten)

Node	Beschreibung
`flow_log`	Debug-Log (level / prefix)
`flow_noop`	Durchgang
`flow_delay`	Nach `delay_ms` Weiterleitung an den nächsten Node
`flow_throttle`	Begrenzung `max_msgs` / `window_ms` (bei Überschreitung Relation `THROTTLED`)
`flow_debounce`	Löst nach Stabilisierung von `window_ms` nur die letzte Nachricht aus
`flow_merge`	Kumuliert Eingaben während `window_ms` und emittiert einmalig als `data.merged`-Array
`flow_subflow`	`target_flow_id` — Aufruf eines anderen Flows
`flow_retry`	`max_attempts` (Standard 3) / `backoff_ms` (Standard 1000) / `backoff_multiplier` (Standard 2.0). Nach Backoff-Wartezeit wird die Nachricht in den `SUCCESS`-Zweig weitergeleitet, bei Erreichen der maximalen Versuche in den `EXHAUSTED`-Zweig. `metadata.retry_count` / `metadata.retry_exhausted` automatisch aktualisiert

Empfohlenes Verdrahtungsmuster für Retry

[risky_node] ─[FAILURE]─▶ [flow_retry] ─[SUCCESS]─▶ (다시 risky_node 로 루프 연결)
                                       └[EXHAUSTED]─▶ [에러 핸들러 / 알림]

Detaillierte Node-Optionsreferenz

Beschreibt die Optionen komplexer Nodes ausführlich mit Standardwerten·Beispielen·Fehlerbehandlung je Option, nicht nur in einer einzeiligen Tabelle.

`flow_script_filter` — Skriptfilter

Option	Typ	Standard	Beschreibung
`script`	text	(erforderlich)	Auswertungsausdruck — gibt boolean zurück. `true` → Verzweigung `TRUE` / `false` → Verzweigung `FALSE`
`script_type`	enum	`javascript`	`javascript` / `eql`
`on_error`	enum	`FALSE`	Bei Skript-Exception — ob in Verzweigung `TRUE` / `FALSE` / `FAILURE` gesendet wird

Skriptkontext

Variable	Bedeutung
`msg.type`	Nachrichtentyp (z. B. POST_TELEMETRY)
`msg.data`	Payload (veränderbar, hat aber im Filter keine Bedeutung)
`msg.metadata`	Kontext
`msg.originator`	originator-Objekt

Beispiel

// 온도가 임계 초과 + 야간 시프트만
msg.data.value > 80 && msg.metadata.shift === 'NIGHT'

// 사이트별 임계 분기
var th = {'SITE-A': 80, 'SITE-B': 90, 'SITE-C': 75};
msg.data.value > (th[msg.metadata.site_id] || 100);

`flow_script_transform` — Skript-Transformation

Option	Typ	Standard	Beschreibung
`script`	text	(erforderlich)	Transformationsausdruck — ändert das `msg`-Objekt oder gibt ein neues zurück
`script_type`	enum	`javascript`	`javascript` / `eql`
`mode`	enum	`mutate`	`mutate` (in-place) / `return` (verwendet Rückgabewert)

Beispiel

// data 에 계산 필드 추가
msg.data.fahrenheit = msg.data.value * 9/5 + 32;
msg.metadata.processed_at = Date.now();

// 페이로드 통째로 교체 (mode=return)
return {
  type: 'WEBHOOK',
  originator: msg.originator,
  data: { temp: msg.data.value, level: msg.data.value > 80 ? 'HIGH' : 'OK' },
  metadata: msg.metadata
};

Im Modus mutate werden return-Anweisungen ignoriert, selbst wenn vorhanden. Um durch ein neues Objekt zu ersetzen, muss mode=return eingestellt werden.

`flow_switch` — Mehrfachverzweigung

Option	Typ	Standard	Beschreibung
`cases`	array	(erforderlich)	`[{expression, relation}]`-Array — Auswertung von oben nach unten, erster Treffer wird verwendet
`default_relation`	string	`DEFAULT`	Falls kein Case zutrifft
`script_type`	enum	`javascript`	—

Beispiel

// cases 설정
[
  { "expression": "msg.data.value > 90", "relation": "CRITICAL" },
  { "expression": "msg.data.value > 80", "relation": "WARN" },
  { "expression": "msg.data.value > 70", "relation": "INFO" }
]
// default_relation: "NORMAL"

Im nachfolgenden Node ist mit 4 Verzweigungslabels (CRITICAL/WARN/INFO/NORMAL) jeweils eine unterschiedliche Verarbeitung möglich.

`flow_retry` — Automatischer Wiederholungsversuch

Erholt sich automatisch von vorübergehenden Fehlern externer IO-Nodes.

Option	Typ	Standard	Beschreibung
`max_attempts`	int	3	Maximale Versuchsanzahl (bei Überschreiten dieses Werts `EXHAUSTED`)
`backoff_ms`	long	1000	Erste Wartezeit (ms)
`backoff_multiplier`	double	2.0	Exponentieller Backoff-Multiplikator — 1. Versuch 1 s → 2. Versuch 2 s → 3. Versuch 4 s
`max_backoff_ms`	long	30000	Obergrenze der einzelnen Wartezeit
`jitter_pct`	int	0	Zufälliges Jitter von ±N% beim Backoff (zur Vermeidung von Thundering Herd)

Automatische Ergänzung von `metadata`

Feld	Bedeutung
`metadata.retry_count`	Bisherige Anzahl der Versuche
`metadata.retry_exhausted`	Bei true erfolgt der Eintritt in den EXHAUSTED-Zweig
`metadata.retry_last_error`	Grund des letzten Fehlschlags

Wenn die Summe der Backoff-Zeiten 60 Sekunden übersteigt, kann die Trigger-Verarbeitungswarteschlange blockiert werden. Wenn die Antwort des externen Systems durchgängig langsam ist, begrenzen Sie zunächst mit flow_throttle die Eintrittsgeschwindigkeit.

`flow_on_webhook` — HTTP-Trigger

Option	Typ	Standard	Beschreibung
`auth_required`	boolean	`true`	Ob der Header `X-API-Key` erforderlich ist — bei Deaktivierung kann jeder aufrufen
`allowed_origins`	csv	`*`	Whitelist der CORS-Origins
`max_body_kb`	int	256	Obergrenze der Body-Größe (bei Überschreitung Antwort 413)
`payload_pattern`	glob	`*`	Glob zur Vorabfilterung von Nachrichten

Aufrufmethode

curl -X POST \
  https://platform.example.com/api/v4/flow/webhook/{flow_id} \
  -H "X-API-Key: {edge_or_token_key}" \
  -H "Content-Type: application/json" \
  -d '{"order_no":"PO-001","customer":"ACME","quantity":1000}'

{flow_id} im Pfad kann in der Listenansicht kopiert werden
X-API-Key ist ein Token, das im Bildschirm API Key oder API-Authentifizierungstoken des Edge ausgestellt wurde
Antworten: 200 OK (in die Nachrichtenwarteschlange eingereiht) / 401 (Authentifizierungsfehler) / 404 (Flow nicht vorhanden oder nicht bereitgestellt) / 413 (Body überschritten)

`flow_http_request` — Externer HTTP-Aufruf

Option	Typ	Standard	Beschreibung
`url`	string	(erforderlich)	Aufzurufende URL. `${data.x}`-Vorlagenersetzung
`url_field`	string	—	Wenn die URL dynamisch aus der Payload bezogen werden soll — z. B. `data.endpoint`
`method`	enum	`GET`	`GET` / `POST` / `PUT` / `DELETE` / `PATCH`
`method_field`	string	—	Methode dynamisch aus der Payload
`headers`	json	`{}`	Format `{"Authorization": "Bearer ${TOKEN}"}`
`body`	text	—	Statischer Body (unterstützt Vorlagenersetzung)
`body_field`	string	—	Wenn der Body aus der Payload bezogen werden soll — üblicherweise `data`
`timeout_ms`	int	5000	Obergrenze der Antwortwartezeit
`follow_redirect`	boolean	`true`	Automatisches Folgen von 3xx-Redirects
`verify_ssl`	boolean	`true`	TLS-Zertifikatsprüfung (nur zu Testzwecken deaktivieren)

Ergänzung der Antwort-Payload

Feld	Bedeutung
`data.response_status`	HTTP-Statuscode (200 / 404 / 500 ...)
`data.response_body`	Antwortkörper (bei JSON automatisch geparst)
`data.response_headers`	Antwort-Header-Objekt

Verzweigung

SUCCESS — 2xx/3xx
FAILURE — 4xx/5xx oder Exception/Timeout

`flow_send_email` — E-Mail-Versand

Option	Typ	Standard	Beschreibung
`to`	string	—	Statischer Empfänger (kommagetrennt)
`to_field`	string	—	Empfänger aus der Payload extrahieren — z. B. `data.recipient`
`cc` / `cc_field`	string	—	CC
`bcc` / `bcc_field`	string	—	BCC
`subject` / `subject_field`	string	(mind. 1 erforderlich)	Betreff — Vorlagenersetzung
`body` / `body_field`	text	(mind. 1 erforderlich)	Text (HTML erlaubt)
`is_html`	boolean	`true`	Bei reiner Textmail deaktivieren
`attachments`	json	`[]`	`[{"url":"...","filename":"..."}]`

Die SMTP-Konfiguration wird im Voraus vom Operator unter System → Einstellungen in den E-Mail-Einstellungen registriert. Vor der Registrierung fallen alle E-Mail-Nodes in den Zweig FAILURE.

`flow_jdbc_poll` — Externes DB-Polling

Option	Typ	Standard	Beschreibung
`datasource_id`	string	(erforderlich)	Externe DB-Kennung, registriert unter System → Einstellungen
`query`	sql	(erforderlich)	SELECT-Abfrage — liefert maximal 1.000 Zeilen pro Aufruf
`poll_interval_ms`	int	60000	Polling-Intervall (Standard 1 Minute)
`marker_column`	string	—	Spalte „Letzter Verarbeitungszeitpunkt" — SELECTiert nur Zeilen nach dem Marker
`marker_initial`	string	`1970-01-01 00:00:00`	Ausgangswert des Markers beim ersten Polling
`row_limit`	int	1000	Maximale Zeilenanzahl pro Polling (sicher auch bei Überschreitung)
`on_error_continue`	boolean	`true`	Bei DB-Fehler nur Diagnose protokollieren und mit dem nächsten Polling fortfahren

Anwendungsbeispiel für `marker_column`

SELECT po_no, customer, qty, created_at
FROM po
WHERE created_at > :marker
ORDER BY created_at

→ An der Stelle von :marker wird automatisch der maximale created_at-Wert aus dem letzten Polling eingesetzt.

`flow_kafka_publish` / `flow_mqtt_publish` — Externe Publikation

Option	Typ	Standard	Beschreibung
`broker`	string	(erforderlich)	`kafka:9092` oder `tcp://mqtt:1883`
`topic`	string	—	Statisches Topic — `${data.x}`-Ersetzung möglich
`topic_field`	string	—	Topic aus der Payload extrahieren (z. B. `data.target_topic`)
`key` / `key_field`	string	—	(nur Kafka) Nachrichtenschlüssel
`body` / `body_field`	json/text	(mind. 1 erforderlich)	Zu publizierender Inhalt — falls nicht gesetzt, `data` unverändert
`qos`	int	1	(nur MQTT) 0/1/2
`retain`	boolean	false	(nur MQTT) Retained-Flag

`flow_publish_asset_*` — Anlagenereignis publizieren

Anlagenereignisse (Ereignis/Kontext/Aggregation/Befehl) werden über einen einheitlichen Kanal publiziert, der gleichzeitig von CEP/Plugin/Timeline erkannt wird. Gemeinsame Optionen der 4 Nodes:

Option	Typ	Standard	Beschreibung
`asset_id`	string	—	Statische Anlagen-ID
`asset_id_field`	string	—	Anlagen-ID aus der Payload extrahieren (üblicherweise `metadata.asset_id`)
`event_type`	string	—	Klassifizierung des Anlagenereignisses (z. B. `STARTUP`, `SHUTDOWN`, `MAINTENANCE`)
`event_type_field`	string	—	Klassifizierung aus der Payload extrahieren
`payload`	json	`${data}`	Zu publizierender Inhalt — falls nicht gesetzt, `data` unverändert

Bei flow_publish_asset_command wird die Nachricht sofort über das Befehlsempfangs-Topic der Anlage (asset_id/cmd/{event_type}) an das Edge zugestellt.

`flow_create_` / `flow_update_` — Domänen-CRUD

Alle Create/Update-Nodes teilen sich folgendes Optionsmuster.

Option	Typ	Beschreibung
`{컬럼명}`	string	Statischer Wert (falls nicht eingegeben, NULL/Default)
`{컬럼명}_field`	string	Wert aus der Payload extrahieren (`data.foo` / `metadata.bar`)
`id_strategy`	enum	`auto` (systemvergeben) / `field` (aus `{도메인}_id_field` extrahiert)
`on_duplicate`	enum	`error` (Standard) / `skip` / `update` — nur bei Create-Nodes

Automatische NOT-NULL-Ergänzung

Create-Nodes befüllen NOT-NULL-Spalten automatisch mit Default-Werten.

Domäne	Automatisch befüllte Spalte
Arbeitsauftrag	`status="WAIT"` / `master_id` (bei Fehlen automatisch nachgefüllt) / `insert_user_id="flow"`
Kunde	Managerinfo `"admin"` / `"admin@example.com"` (falls nicht registriert)
Mitarbeiter	`org_id=site_id` (Fallback)
Gemeinsam	`insert_date=now()` / `insert_user_id="flow"`

Behandlung leerer Strings bei FK

Wenn bei einer FK-Spalte (z. B. customer_id/product_id) ein leerer String "" eingeht, wird er automatisch in NULL umgewandelt. In JS ist msg.data.customer_id = '' sicherer als delete msg.data.customer_id.

`flow_edge_*` — Edge-REST-Aufruf

Die 22 Nodes, die den REST-Endpunkt eines Edge-Geräts aufrufen, teilen sich gemeinsame Optionen.

Option	Typ	Standard	Beschreibung
`edge_id`	string	—	Statische Edge-ID
`edge_id_field`	string	—	Edge-ID aus der Payload extrahieren
`path`	string	(Node-spezifischer Standard)	Edge-REST-Pfad (z. B. `/api/v1/opc`, `/api/v1/app/grafana/start`)
`body_template`	text	—	Request-Body — falls nicht gesetzt, `data` unverändert
`timeout_ms`	int	5000	—

Automatische Authentifizierung

Der Edge-Node führt beim Aufruf automatisch ein Lookup von api_key für dieses Edge in der Mastertabelle mm_edge durch und hängt es als X-API-Key-Header an. Der Operator muss keine separate Konfiguration vornehmen.

Antwort-Payload

Feld	Bedeutung
`data.edge_response_status`	Edge-Antwortcode
`data.edge_response`	Antwortkörper
`data.edge_id`	Ziel-Edge-ID des Aufrufs (zur Bestätigung)

Die Verzweigung ist identisch mit flow_http_request (SUCCESS / FAILURE).

Skript-Nodes schreiben

Die Nodes flow_script_filter / flow_script_transform / flow_switch unterstützen zwei Ausdruckstypen.

JavaScript (Standard, empfohlen)

Standard-ECMAScript-Syntax. Mehrzeiligkeit·var/let/const·Funktionen·Objektliterale werden alle unterstützt.

Bindings

Variable	Beschreibung
`msg`	Gesamte Nachricht. Direkter Zugriff auf `msg.data.x`, `msg.metadata.topic`, `msg.type`, `msg.originator.id`
`data`	Kurzalias für `msg.data`
`metadata`	Kurzalias für `msg.metadata`

Filterbeispiel

data.temp > 80

Transformationsbeispiel

data.temp_f = data.temp * 1.8 + 32;
data.alert = data.temp > 80 ? 'HIGH' : 'OK';
msg

Switch-Beispiel (boolean je Case)

data.t > 100   // case "Critical"

Häufig verwendete Transformationsmuster

// 1) 단위 변환 (섭씨 → 화씨) + 라벨링
data.temp_f = data.temp * 1.8 + 32;
data.alert  = data.temp > 80 ? 'HIGH' : 'OK';
msg

// 2) 메타데이터 보강 — 시간대·시프트 자동 부여
const h = new Date(metadata.ts).getHours();
metadata.shift = (h >= 6 && h < 18) ? 'DAY' : 'NIGHT';
msg

// 3) 외부 페이로드를 도메인 모델로 매핑 (MES PO → WorkOrder)
const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty
};
msg

// 4) 실패 분기로 명시적 라우팅 (필수 필드 누락 시)
if (!data.tag_id || data.value == null) throw new Error('필수 필드 누락');
msg

// 5) 배열 분할 후 데이터 정제 — split 노드 후에 사용
data.value     = parseFloat(data.raw);
data.threshold = data.value > 100;
msg

Häufig verwendete Filtermuster

// 우선순위 화이트리스트
['ERROR', 'CRITICAL'].includes(data.priority)

// 시간대 기반 필터 (주간만 허용)
new Date(metadata.ts).getHours() >= 8 && new Date(metadata.ts).getHours() < 20

// 자산 ID 패턴 매칭
/^MOTOR-.*$/.test(originator.id)

// 임계값 + 안정성 (값이 5번 이상 누적된 경우)
data.value > data.threshold && data.consecutive_count >= 5

Benutzerskripte laufen in einer sicheren Sandbox, in der Datei-, Netzwerk-, Thread- und beliebiger Klassenzugriff blockiert ist. Wenn ein externer Systemaufruf benötigt wird, verdrahten Sie separat einen externen Anbindungs-Node wie flow_http_request.

EQL-Ausdruck

Für Kompatibilität mit bestehenden EQL-Nutzern. Unterstützt nur einen einzelnen Ausdruck (keine Mehrzeiligkeit·keine semikolongetrennte Trennung), nur Muster mit Seiteneffekten sind nutzbar.

#msg.getData().getInt('temp') > 80

Wenn mehrere Aktionen benötigt werden, verwenden Sie bitte JavaScript.

Spezifikation der JavaScript-Laufzeitumgebung

Skripte werden in einer isolierten Sandbox ausgeführt. Sie sollten genau wissen, welche Funktionen möglich/nicht möglich sind, um stabile Skripte schreiben zu können.

Verfügbar (✅)

Funktion	Anmerkung
Standard-ECMAScript	`var`/`let`/`const`·Funktionen·Klassen·Destrukturierung·Spread·`?.`·`??` usw.
Objektliterale	`{ key: value, ... }`
Array-Methoden	`map` / `filter` / `reduce` / `forEach` / `find` / `some` / `every` / `flat` / `slice`
String-Methoden	`split` / `replace` / `includes` / `match` / `padStart` / `repeat`
Mathematische Funktionen	Gesamter `Math.*`
JSON	`JSON.parse` / `JSON.stringify` (falls `data` bereits ein Objekt ist, ist erneutes stringify nicht nötig)
Date	`new Date()` / `Date.now()` / `getHours()` / `toISOString()` usw.
Reguläre Ausdrücke	`/pattern/`-Literal + `RegExp`-Konstruktor
Error throw	`throw new Error('...')` — automatische Verzweigung in `FAILURE`
try/catch/finally	Fehlerbehandlung

Nicht verfügbar (❌)

Funktion	Grund / Alternative
Netzwerkaufrufe (`fetch` / `XMLHttpRequest`)	Sandbox-Blockade — separat den Node `flow_http_request` verdrahten
Dateisystem (`require('fs')`)	Sandbox-Blockade
Threads (`setTimeout` / `setInterval` / `Worker`)	Nur synchrone Ausführung erlaubt — bei benötigter Verzögerung den Node `flow_delay` verwenden
`require` / `import`	Kein Laden externer Module möglich — benötigte Funktionen müssen im selben Skript definiert werden
`eval` / `new Function(string)`	Aus Sicherheitsgründen blockiert
Beliebige Java-Klassen	Auch im EQL-Kompatibilitätsmodus dem Benutzercode nicht zugänglich
`process` / `global` / `window`	Nicht definiert
WebSocket / EventSource	Blockiert — Nachrichtenempfang erfolgt über Trigger-Nodes

Ausführungsgrenzen — es gibt keine

Für Skripte sind keine Zeit-/Speicherlimits gesetzt

Ein älteres Dokument enthielt eine Tabelle mit 500 ms Ausführungszeit · 16 MB Speicher · 1024 Stack · 2 MB Ausgabe, aber keiner dieser vier Werte wird angewendet.

Es sieht so aus, als könnte man in der Node-Konfiguration timeout_ms verwenden, aber Skript-Nodes lesen diesen Wert nicht. (ScriptTransformNode · ScriptFilterNode lesen nur script und language.)
Auch im JS-Ausführungskontext sind keine Grenzen für Zeit·Anweisungsanzahl·Speicher gesetzt.

Deshalb blockiert ein Skript wie while (true) {} ununterbrochen einen Worker-Thread. Das flow-weite Limit von 30 Sekunden (flow_timeout_ms) wird geprüft, wenn die Ausführungsschleife die nächste Aufgabe entnimmt — eine Endlosschleife innerhalb des Skripts erreicht diese Prüfung nie.

Verwenden Sie nur Skripte, die von selbst enden. Vermeiden Sie Endlosschleifen·massive Wiederholungen·das Anhäufen großer Strings, und setzen Sie in Schleifen unbedingt eine Obergrenze.

Was tatsächlich blockiert ist — Sicherheits-Sandbox

Statt Zeit·Speicher sind Aktionen, die nach außen greifen, sicher blockiert.

Element	Status
Zugriff auf Java-Klassen (z. B. `Java.type`)	Blockiert
Datei-·Netzwerk-IO	Blockiert
Thread-Erstellung	Blockiert
Nativer Zugriff	Blockiert
Experimentelle Optionen	Blockiert
Zugriff auf Host-Objekte	Nur auf der Whitelist stehende

Wenn ein externer Aufruf nötig ist, tun Sie dies nicht im Skript, sondern trennen Sie es in einen speziellen Node wie flow_http_request — im Skript funktioniert es ohnehin nicht, und bei speziellen Nodes greift timeout_ms tatsächlich.

Bündeln Sie schwere Verarbeitung nicht in einem einzigen Skript, sondern verteilen Sie sie auf mehrere Nodes.

Rückgaberegel für `msg`

// flow_script_transform 의 두 가지 모드
// 1) mutate 모드 (기본) — msg 객체를 직접 수정, 마지막에 msg 또는 아무 값 반환
data.foo = 'bar';
msg

// 2) return 모드 — 완전히 새 객체로 교체
return {
  type: msg.type,
  originator: msg.originator,
  data: { ...data, foo: 'bar' },
  metadata: msg.metadata
};

Standardzeit·Locale

Element	Verhalten
Zeitzone des Server-Systems	UTC (direkte Verwendung von Epoch ms empfohlen)
Anzeige koreanischer Zeit	`toLocaleString('ko-KR', { timeZone: 'Asia/Seoul' })`
Zeitberechnung koreanischer Zeit	`new Date(ts + 9*3600000).getUTCHours()` oder obige Locale
Sekundengenauer Timestamp	`Math.floor(Date.now() / 1000) * 1000`

Speichersichere Schreibmuster

Skript-Nodes haben ein Speicherlimit von 16 MB, aber bei häufig aufgerufenen Nodes summiert sich ein kleines Speicherleck zu einer großen GC-Belastung des Engines. Vermeiden Sie folgende Muster.

Antipattern 1 — Closure hält große Daten fest

// ❌ 나쁜 예 — 큰 배열을 변환 함수에 캡쳐
const heavy = data.records || [];          // 1만 건
const summarize = (item) => heavy.find(r => r.id === item.id);
data.matched = data.targets.map(summarize);
msg

// ✅ 좋은 예 — 인덱스 미리 만들고 함수 안에서만 사용
const index = {};
(data.records || []).forEach(r => { index[r.id] = r; });
data.matched = data.targets.map(t => index[t.id]);
data.records = undefined;     // 변환 후 큰 원본 제거
msg

Antipattern 2 — Übertragung großer Arrays unverändert

// ❌ data.records 가 10,000 건이면 모든 후속 노드에서 메모리 차지
msg

// ✅ 필요한 통계만 남기고 원본 제거
data.summary = {
  count: (data.records || []).length,
  total: (data.records || []).reduce((s, r) => s + r.value, 0)
};
delete data.records;
msg

Antipattern 3 — Tiefe Objektkopie

// ❌ JSON.parse(JSON.stringify(obj)) 는 큰 객체에서 매우 느림
data.copy = JSON.parse(JSON.stringify(data.original));

// ✅ 얕은 복사 또는 필요한 필드만 직접 선택
data.summary = { id: data.original.id, name: data.original.name };

Antipattern 4 — Explosion regulärer Ausdrücke (Catastrophic Backtracking)

// ❌ (a+)+b 형태의 중첩 그룹은 입력에 따라 지수 시간
const re = /^(a+)+b$/;
if (re.test(data.text)) ...

// ✅ 비포획 그룹 + atomic 그룹 패턴 또는 단순 매칭
const re = /^a+b$/;
if (re.test(data.text)) ...

Antipattern 5 — Kumulative Variable `let` außerhalb der Funktion

// ❌ 함수 밖 let 은 매 노드 호출마다 0으로 리셋되지만, 의도와 다르게 동작 가능
let counter = 0;
data.items.forEach(() => counter++);
data.counter = counter;

// ✅ 명시적으로 함수 안에서 선언
data.counter = data.items.length;     // 같은 결과, 더 명확

Antipattern 6 — Verschachteltes try/catch mit endlosen Versuchen

// ❌ 실패해도 다시 던지지 않으면 그래프가 잘못된 분기로 이동
try {
  riskyCall();
} catch (e) { /* 무시 */ }
msg

// ✅ 실패 의도면 throw, 정상 의도면 명시적으로 기록
try {
  riskyCall();
  data.status = 'OK';
} catch (e) {
  data.status = 'ERROR';
  data.error_msg = e.message;
}
msg

Eine im Skript per throw ausgelöste Exception wird in den FAILURE-Zweig geleitet, wobei die Nachricht erhalten bleibt. Externe IO-Nodes haben eine eigene Verzweigung — umschließen Sie sie nicht mit try/catch.

Diagnose bei Speicherdruck

Es gibt keinen Code wie NODE_SCRIPT_OOM

Ein älteres Dokument beschrieb NODE_SCRIPT_OOM · FLOW_MSG_TRIMMED · ENGINE_GC_PRESSURE als Signal für Speicherdruck, aber das Produkt erzeugt keinen solchen Code. Es gibt weder eine Speicherobergrenze für Skripte noch eine automatische Payload-Kürzung.

Beurteilen Sie Speicherdruck nicht anhand von Code, sondern anhand von Kennzahlen und Symptomen.

Worauf achten	Druckhinweis
`FLOW_EXECUTOR_DROPPED_COUNT`	Beginnt bei 0 anzusteigen — der Verarbeitungspool ist gesättigt und verwirft Nachrichten
`FLOW_EXECUTOR_QUEUE_SIZE`	Steigt kontinuierlich ohne Rückgang
`FLOW_EXECUTION_TIME`	Maximalwert steigt auf ein Vielfaches des Normalwerts
Serverlog	Warnung `work_queue full ... task dropped`, Warnung `FlowExecutor timeout`
JVM	Zunehmende GC-Zeit · steigende Heap-Auslastung

Wenn Sie diese Signale bemerken, prüfen Sie zuerst Flows, die große Payloads verarbeiten, und stellen Sie fest, ob im Skript große Strings· Arrays angehäuft werden.

Mobile-Push-Benachrichtigungskanal

Mit dem Node flow_send_push senden Sie Push-Benachrichtigungen an die mobile App des Operators.

Node-Optionen

Option	Typ	Standard	Beschreibung
`to` / `to_field`	string	(mind. 1 erforderlich)	Empfänger-Benutzer-ID (kommagetrennt) oder Payload-Pfad
`title` / `title_field`	string	(mind. 1 erforderlich)	Titel der Benachrichtigung (max. 50 Zeichen empfohlen)
`body` / `body_field`	text	(mind. 1 erforderlich)	Text (max. 120 Zeichen empfohlen)
`priority`	enum	`NORMAL`	`NORMAL` / `HIGH` — HIGH wird auch auf dem Sperrbildschirm angezeigt
`sound`	enum	`default`	`default` / `silent` / benutzerdefinierter Ton
`data`	json	`{}`	Zusatzdaten, die von der App verarbeitet werden (Payload-Limit 4 KB)
`deep_link`	string	—	Bildschirm, der beim Tippen auf die Benachrichtigung geöffnet wird (z. B. `pp://asset/MOTOR-001`)
`ttl_sec`	int	86400	Aufbewahrungszeit bei Nichtempfang (Sekunden) — automatisches Löschen nach Ablauf

Automatisches Token-Routing pro Benutzer

Wenn sich ein Operator zum ersten Mal in der mobilen App anmeldet, wird das Gerätetoken automatisch unter Sicherheit → API-Authentifizierungstoken registriert. Im Flow müssen Sie das Token nicht direkt handhaben, sondern nur die Benutzer-ID angeben.

Wenn sowohl iOS- als auch Android-Token registriert sind, erfolgt der Versand an beide Geräte
Ist ein Token ungültig (App deinstalliert usw.), wird die Registrierung automatisch aufgehoben

Einfaches Versandbeispiel

[flow_on_asset_alarm]
  ↓ where priority='ERROR'
[flow_send_push]
   to_field:    "metadata.responsible_user"
   title:       "🚨 ${metadata.asset_id} 알람"
   body_field:  "data.band_message"
   priority:    HIGH
   deep_link:   "pp://alarm/view/${data.alarm_id}"

Mehrsprachiger Push

// flow_script_transform — 사용자 로케일에 따라 본문 분기
const locale = metadata.user_locale || 'ko-KR';
const templates = {
  'ko-KR': { title: '🚨 ${asset} 위험 알람',           body: '값 ${value}, 즉시 점검 바랍니다.' },
  'en-US': { title: '🚨 ${asset} Critical Alarm',     body: 'Value ${value}, please inspect immediately.' },
  'ja-JP': { title: '🚨 ${asset} 危険警報',            body: '値 ${value}, 即時点検が必要です。' }
};
const tpl = templates[locale] || templates['ko-KR'];
data.push_title = tpl.title.replace('${asset}', metadata.asset_id).replace('${value}', data.value);
data.push_body  = tpl.body.replace('${value}', data.value);
msg

Danach Verwendung als title_field=data.push_title / body_field=data.push_body von flow_send_push.

Gruppierte Benachrichtigungsbündelung (Inbox-Stil)

Mehrere Alarme auf einmal zu einer Benachrichtigung bündeln (im 5-Minuten-Takt):

[flow_on_asset_alarm]
   ↓
[flow_merge window=300s]
   ↓ (data.merged 배열)
[flow_script_transform — 요약 만들기]
   ↓ data.push_body="알람 N건: A자산, B자산, ..."
[flow_send_push]

Automatische Eskalation bei Abwesenheit des Benutzers

Automatische Eskalation per SMS oder E-Mail 30 Minuten nach unbestätigtem Push.

[flow_on_asset_alarm priority=ERROR]
   ↓
[flow_send_push]
   ↓ SUCCESS
   ↓ data.alert_id = response_id
[flow_delay 1800s (30분)]
   ↓
[flow_http_request GET /api/alert/${alert_id}/status]
   ↓ (response_body.read=false 면)
[flow_script_filter (data.response_body.read === false)]
   ↓ TRUE
[flow_send_sms]
   to_field: "metadata.responsible_phone"
   text:     "푸시 미확인 30분 경과: ${data.title}"

Empfehlung zur Begrenzung der Versandhäufigkeit

Priorität	Empfohlene Häufigkeit
HIGH (Anzeige auf Sperrbildschirm)	Max. 5 pro Stunde und Benutzer — zur Vermeidung von Alarmmüdigkeit
NORMAL	Max. 20 pro Stunde und Benutzer

Verdrahten Sie den Node flow_throttle vor dem Versand, oder senden Sie Alarme derselben Anlage erst nach Stabilisierung mit flow_debounce.

Automatische Erneuerung externer Auth-Token

Muster zur automatischen Erneuerung ablaufender Tokens wie OAuth2 beim Aufruf externer Systeme.

Einfache Erneuerung — zeitbasiert (Cron)

[flow_schedule cron='0 */50 * * * ?']    ← 50분마다 (만료 1시간 전)
   ↓
[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=client_credentials&client_id=${creds.client_id}&client_secret=${creds.client_secret}"
   ↓
[flow_script_transform]
   ↓ data.access_token = data.response_body.access_token
[flow_save_attributes]    ← 자격 증명 저장소에 저장
   target_id:   "creds:erp_api"
   attributes:  {"access_token": "${data.access_token}", "expires_at": ${data.response_body.expires_in * 1000 + ts}}

Danach in flow_http_request eines anderen Flows:

Headers: Authorization: Bearer ${creds:erp_api.access_token}

Proaktive Erneuerung — bei Erhalt von 401

[main flow]
   ↓
[flow_http_request]
   ↓ SUCCESS → 정상 처리
   ↓ FAILURE
[flow_script_filter (response_status === 401)]
   ↓ TRUE
[flow_subflow target_flow_id="refresh-token"]    ← 토큰 갱신
   ↓
[원래 노드로 루프]    ← 갱신된 토큰으로 재시도

Verwendung des Refresh Tokens

[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=refresh_token&refresh_token=${creds.refresh_token}"
   ↓
[flow_script_transform]
   // 새 access_token + 새 refresh_token (rotation)
   data.access_token  = data.response_body.access_token;
   data.refresh_token = data.response_body.refresh_token;
   data.expires_at    = Date.now() + data.response_body.expires_in * 1000;
   msg
   ↓
[flow_save_attributes]

Automatischer Alarm bei Ablauf-Schwelle des Tokens

[flow_schedule cron='0 0 * * * ?']    ← 매시
   ↓
[flow_jdbc_query]
   sql: "SELECT id, expires_at FROM credentials WHERE expires_at < NOW() + INTERVAL '1 day'"
   ↓
[flow_split]
   ↓
[flow_send_email]
   subject: "API 토큰 만료 임박: ${data.id}"
   body:    "${data.id} 토큰이 ${data.expires_at} 만료 예정입니다."

Empfohlener Speicherort für Zugangsdaten

Art	Empfohlener Speicherort
Statisch (kaum Änderungen)	Zugangsdatenspeicher unter System → Einstellungen
Dynamisch (automatische Erneuerung)	Mit obigem Muster über `flow_save_attributes` in Anlagen-Metadaten speichern
Benutzerspezifisches OAuth	Bildschirm Sicherheit → API-Authentifizierungstoken

Belassen Sie Zugangsdaten nicht im Klartext innerhalb des Graphen, sondern schreiben Sie sie unbedingt in Form eines Verweises auf einen externen Speicher (${creds.x}). Auch bei Export/Import werden Klartext-Token nicht in die JSON aufgenommen.

Mit der Option language wird ausgewählt, welche Sprache verwendet wird (JS oder EQL).

Ausführungshistorie

Sie können nodebezogene Ausführungsereignisse chronologisch abfragen.

Suchbedingungen

Element	Beschreibung
Zeit	Gibt den abzufragenden Zeitraum an
Ebene	`전체` / `INFO` / `WARN` / `ERROR`
Flow-ID	Filtert nur einen bestimmten Flow
Nachrichten-ID	Zur Verfolgung einer einzelnen Nachricht
Anzahl	Letzte 200/500/1.000 Einträge

Schneller Zeitbereich

Über die Schaltflächen oben im Zeitachsenpanel sofort navigieren: 10분전 / 30분전 / 1시간전 / 6시간전 / 12시간전 / 전체기간.

Ergebnisspalten

Spalte	Beschreibung
Ebene	INFO / WARN / ERROR
Zeit	Zeitpunkt des Ereignisses
Ereignis	`FLOW_START` / `NODE_IN` / `NODE_OUT` / `NODE_ERROR` / `FLOW_END`
Flow-ID	Welcher Flow
Node	Anzeigename des Nodes
Node-Typ	z. B. `flow_script_transform`
Relation	`SUCCESS` / `FAILURE` / `TRUE` / `FALSE` / `MATCH` / `NO_MATCH` / `DEFAULT` / `THROTTLED` / `EXHAUSTED` (alle in Großbuchstaben)
Nachricht	Zusammenfassung von Nachrichtentyp · Subjekt · Relation · Verarbeitungszeit · Datenvorschau

Mit der Schaltfläche CSV herunterladen können Sie das aktuelle Abfrageergebnis exportieren.

Die Aufbewahrungsdauer der Ausführungshistorie beträgt 7 Tage. Für eine langfristige Aufbewahrung laden Sie die Daten bitte in ein externes Log-System.

Allgemeiner Workflow

Neuen Flow erstellen — Schaltfläche 새 플로우 in der Listenansicht → Name·Beschreibung eingeben
Bearbeitungsansicht öffnen — automatisch öffnet sich eine leere Canvas
Trigger-Node platzieren — Trigger-Node aus der linken Palette ziehen
Verarbeitungs-Nodes hinzufügen — in der Reihenfolge Filter → Transformation → Aktion platzieren und mit Wires verbinden
Node konfigurieren — jeden Node anklicken und Optionen im rechten Inspector eingeben
Speichern — Schaltfläche 저장 oben rechts (automatischer Snapshot wird abgelegt)
Testlauf — eine beliebige Nachricht injizieren und das Ergebnis prüfen
Bereitstellen — mit dem 배포-Schalter aktivieren → automatische Ausführung bei eintreffendem Triggerereignis
Überwachung — Prüfung über Live-Debug-Panel und Ausführungshistorie

Beispiel-Flows

Jedes Beispiel besteht aus einem Node-Verdrahtungsdiagramm + zentraler Node-Konfiguration + Verhaltensbeschreibung. Die Node-Konfiguration in JSON-Form entspricht 1:1 den Werten, die Sie im rechten Inspector eingeben.

Beispiel 1: Automatische Erstellung eines MES-Arbeitsauftrags

Fragt neue POs vom MES jede Minute ab und wandelt sie in Arbeitsaufträge um.

[flow_schedule: 매분]
   │
   ▼ TIMER
[flow_http_request: MES /api/po/list?status=NEW]
   │
   ▼ SUCCESS
[flow_split: data → 각 PO별 메시지]
   │
   ▼
[flow_script_transform: PO → WorkOrder 매핑]
   │
   ▼
[flow_create_work_order]
   │
   ├── SUCCESS → [flow_log: 워크오더 생성됨]
   └── FAILURE → [flow_send_email: 실패 알림]

Node-Konfiguration

Node	Zentrale Konfiguration
`flow_schedule`	`cron`: `0 * * * * ?` (jede Minute bei 0 Sekunden)
`flow_http_request`	`method`: `GET`, `url`: `https://mes.example.com/api/po/list?status=NEW`, `headers`: `{"Authorization":"Bearer ${MES_TOKEN}"}`
`flow_split`	`path`: `data` (in Array aufteilen)
`flow_script_transform`	`language`: `JS`, siehe Skript unten
`flow_create_work_order`	`master_id_field`: `data.master_id`, `asset_id_field`: `data.asset_id`, `title_field`: `data.title`
`flow_send_email`	`to_field`: `metadata.alert_to`, `subject`: `[MES 동기화 실패] ${data.po}`

Beispiel Script Transform

data.master_id = 'WO-MES-' + data.po;
data.title     = data.product_name + ' (' + data.qty + ')';
data.asset_id  = data.line_id || 'UNASSIGNED';
data.due_date  = data.delivery_date;
metadata.alert_to = 'ops@example.com';
msg

Beispiel 2: Anlagenbefehl über externen Webhook auslösen

Validiert die von einem externen System gesendete HTTP-Payload und wandelt sie in einen internen Anlagenbefehl um.

[flow_on_webhook]
   │
   ▼ WEBHOOK
[flow_script_filter: payload 검증]
   │
   ├── TRUE
   ▼
[flow_publish_asset_command]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_webhook_callback: 외부에 실패 통보]

Aufrufmethode: Bei einem POST /flow/webhook/{flow_id} mit übertragenem JSON-Body wird der Trigger mit dem Body unverändert im Bereich data ausgelöst.

Beispiel Script Filter

// 인증 토큰 일치 + 필수 필드 존재 검사
if (data.token !== 'EXPECTED_TOKEN') return false;
if (!data.asset_id || !data.cmd_key) return false;
true

Beispiel 3: Alarm → automatische Erstellung eines dringenden Arbeitsauftrags

Bei Empfang eines Alarms mit Schweregrad ERROR oder höher wird für die übergeordnete Anlage ein dringender Instandhaltungsauftrag erstellt.

[flow_on_tag_alarm]
   │
   ▼ ALARM
[flow_script_filter: priority = ERROR/CRITICAL]
   │
   ├── TRUE
   ▼
[flow_change_originator: Tag → 상위 Asset]
   │
   ▼
[flow_create_work_order: 긴급 정비]
   │
   └── SUCCESS → [flow_send_email: 정비 담당자]

Beispiel Script Filter

['ERROR', 'CRITICAL'].includes(data.priority)

Beispiel 4: Externe DB-Synchronisation — Anlagen-Massenregistrierung

Fragt neue Anlagenzeilen aus einer Legacy-DB ab und registriert sie automatisch als Anlagen.

[flow_jdbc_poll: SELECT * FROM legacy_assets WHERE sync_status='NEW']
   │
   ▼
[flow_script_transform: 컬럼 매핑]
   │
   ▼
[flow_create_asset]
   │
   ├── SUCCESS → [flow_jdbc_query: UPDATE legacy_assets SET sync_status='OK' WHERE id=?]
   └── FAILURE → [flow_log: ERROR + flow_send_push]

Node-Konfiguration

Node	Zentrale Konfiguration
`flow_jdbc_poll`	`dsn`: externe DB-Verbindung, `sql`: `SELECT * FROM legacy_assets WHERE sync_status='NEW' LIMIT 100`, `interval_ms`: `60000`
`flow_create_asset`	`asset_id_field`: `data.legacy_id`, `asset_name_field`: `data.name`, `site_id_field`: `data.plant_code`
`flow_jdbc_query`	`sql`: `UPDATE legacy_assets SET sync_status='OK' WHERE id=?`, `params_field`: `data.legacy_id`

Beispiel 5: Automatischer Statusübergang des Arbeitsauftrags

Führt anhand von Start-/Stopp-Ereignissen der Anlage einen automatischen Statusübergang des Arbeitsauftrags durch.

[flow_on_asset_event]
   │
   ▼ ASSET_EVENT
[flow_switch: data.event_type 기준]
   │
   ├── case "RUN"   ─▶ [flow_start_work_order] ─▶ [flow_log]
   ├── case "STOP"  ─▶ [flow_pause_work_order] ─▶ [flow_log]
   ├── case "DONE"  ─▶ [flow_end_work_order]   ─▶ [flow_log]
   └── DEFAULT      ─▶ [flow_noop]

Beispiel für Switch-Node-Cases

RUN: data.event_type === 'RUN'
STOP: data.event_type === 'STOP'
DONE: data.event_type === 'DONE' && data.qty_done >= data.qty_planned

Wenn der Statusübergang fehlschlägt (ungültiger aktueller Status), erfolgt automatisch eine Weiterleitung nach FAILURE, sodass Sie auch ohne separaten Filter sicher verdrahten können.

Beispiel 6: Automatische Anpassung des Alarmbands

Passt anhand einer Anlagenaggregation (z. B. 6-Stunden-Durchschnitt) das obere/untere Alarmband des Tags dynamisch an.

[flow_on_asset_aggregation]
   │
   ▼ ASSET_AGGREGATION
[flow_script_transform: 통계 → 임계값 산출]
   │
   ▼
[flow_update_tag_alarm_band_numeric]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_send_email: 운영자 알림]

Beispiel Script Transform

// 6h 평균 ± 3σ 를 임계값으로 사용
const mean   = data.mean;
const stddev = data.stddev || 1;
data.tag_id  = originator.id + '.TEMP';
data.hi      = mean + 3 * stddev;
data.lo      = mean - 3 * stddev;
data.hi_hi   = mean + 4 * stddev;
data.lo_lo   = mean - 4 * stddev;
data.use_alarm = true;
msg

Beispiel 7: Automatische Registrierung eines Edge-Geräts

Registriert bei Empfang neuer OPC-Serverinformationen diese in großer Menge im Edge-Gerät.

[flow_on_webhook] (POST 본문에 OPC + 태그 목록)
   │
   ▼
[flow_edge_opc_create]
   │
   ▼ SUCCESS
[flow_split: data.tags 배열]
   │
   ▼
[flow_edge_tag_create]
   │
   ▼ SUCCESS (모든 태그 등록 완료 후)
[flow_edge_opc_start]
   │
   └── SUCCESS → [flow_log: 엣지 디바이스 가동]

Beispiel für Aufrufkörper

{
  "type": "WEBHOOK",
  "data": {
    "opc_id":   "OPC-LINE-A",
    "endpoint": "opc.tcp://line-a.local:4840",
    "tags": [
      { "tag_id": "MOTOR-001.SPEED", "address": "ns=2;s=Motor1.Speed" },
      { "tag_id": "MOTOR-001.TEMP",  "address": "ns=2;s=Motor1.Temp"  }
    ]
  }
}

Beispiel 8: Zuverlässigkeitssteigerung bei externer API (Retry)

Wendet Backoff-Wiederholung auf intermittierend fehlschlagende externe API-Aufrufe an und benachrichtigt den Operator, wenn die maximale Versuchszahl überschritten wird.

[flow_on_asset_event]
   │
   ▼
[flow_http_request: 외부 ERP API]
   │
   ├── SUCCESS  → [flow_log]
   └── FAILURE  ─▶ [flow_retry]
                       │
                       ├── SUCCESS    ─▶ (다시 flow_http_request 로 루프 결선)
                       └── EXHAUSTED  ─▶ [flow_send_email: 'ERP 동기화 N회 실패']

Retry-Node-Konfiguration

max_attempts: 5
backoff_ms: 2000
backoff_multiplier: 2.0 → Intervalle von 2, 4, 8, 16, 32 Sekunden

Beispiel 9: Weiterleitung von Plugin-Ereignissen (Benachrichtigung bei niedrigem OEE)

Sendet eine Push-Benachrichtigung an den Linienmanager, wenn der OEE-Wert unter den Schwellenwert fällt.

[flow_on_oee_event]
   │
   ▼
[flow_script_filter: data.availability * data.performance * data.quality < 0.6]
   │
   ├── TRUE
   ▼
[flow_template: '라인 ${originator.id} OEE ${data.oee_pct}%']
   │
   ▼
[flow_send_push: 라인 매니저]

Beispiel Template

body: 라인 ${originator.id} OEE ${data.oee_pct}% (목표 60% 미달, 주요 손실: ${data.top_loss})

Beispiel 10: Pipeline zur Nachrichtenbereinigung (Throttle + Debounce)

Begrenzt häufige Tag-Änderungen auf 1 pro Minute und löst zusätzlich nur dann den nachfolgenden Node aus, wenn 5 Sekunden lang keine Änderung erfolgt.

[flow_on_tag_point]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ├── SUCCESS    → [flow_debounce: window_ms=5000]
   │                                │
   │                                ▼
   │                          [flow_save_attributes]
   └── THROTTLED  → [flow_log: 차단됨]

Beispiel 11: Aggregation mehrerer Trigger (Merge)

Bündelt 3 Ereignistypen (OEE/RAM/EMS) in einem 30-Sekunden-Intervall zu einer einzigen Berichtsnachricht.

[flow_on_oee_event]  ─┐
[flow_on_ram_event]  ─┼─▶ [flow_merge: window_ms=30000]
[flow_on_ems_event]  ─┘                │
                                       ▼
                              [flow_script_transform: 요약 메시지 생성]
                                       │
                                       ▼
                              [flow_send_email: 일일 요약]

Wenn Sie mehrere Trigger-Nodes im selben Flow platzieren, werden sie alle zu Einstiegspunkten. Im data.merged-Array von flow_merge werden alle während des Fensters eingegangenen Nachrichten kumuliert.

Beispiel 12: Gemeinsame Verarbeitung als Subflow bündeln

Trennt gemeinsame Nachrichtenbereinigungslogik (Duplikatprüfung + Einheitenumrechnung + Speicherung) in einen separaten Flow und ruft ihn von mehreren Triggern auf.

Hauptflow (jeweils unabhängig)

[flow_on_tag_point]   ─▶ [flow_subflow: target_flow_id=FLOW_00099]
[flow_on_asset_data]  ─▶ [flow_subflow: target_flow_id=FLOW_00099]

Subflow FLOW_00099

(트리거 없음 — 호출 전용)
[flow_log: 'subflow in']
   │
   ▼
[flow_check_existence_field: data.value 존재]
   │
   ├── TRUE
   ▼
[flow_script_transform: 단위 변환]
   │
   ▼
[flow_save_tag_point]

Ein Subflow kann auch ohne Trigger gespeichert werden und ausschließlich für externe Aufrufe verwendet werden (beim Speichern erscheint eine Warnung).

Beispiel 13: Automatischer Versand eines Tagesberichts bei Schichtwechsel

Sendet täglich zum Ende der Nachtschicht (z. B. 06:00 Uhr) eine E-Mail mit einer Zusammenfassung von Produktion·Qualität·Ausfallzeit über den Zeitraum von gestern bis heute.

[flow_schedule: 0 0 6 * * ?]   (매일 06:00)
   │
   ▼ TIMER
[flow_http_request: 내부 통계 API /api/report/daily]
   │
   ▼ SUCCESS
[flow_script_transform: 본문 마크다운 생성]
   │
   ▼
[flow_send_email]

Beispiel Script Transform

const r = data.report;
data.subject = `[${r.site_id}] ${r.date} 일일 운영 요약`;
data.body    =
  `■ 생산: ${r.qty_done}/${r.qty_planned} (${(100*r.qty_done/r.qty_planned).toFixed(1)}%)\n` +
  `■ 가동률: ${r.availability}%\n` +
  `■ 품질률: ${r.quality}%\n` +
  `■ 다운타임 Top3:\n` +
  r.downtimes.slice(0,3).map(d => ` - ${d.code} ${d.minutes}분`).join('\n');
msg

Beispiel 14: Automatische Ausfallzeitaggregation

Aktualisiert bei Eingang eines STOP-Ereignisses der Anlage die kumulierte Ausfallzeit je Schicht und sendet bei Schwellenüberschreitung eine Benachrichtigung.

[flow_on_asset_event]
   │
   ▼
[flow_msg_type_filter: type in [ASSET_EVENT]]
   │
   ▼ TRUE
[flow_script_filter: data.event_type === 'STOP']
   │
   ▼ TRUE
[flow_script_transform: 다운타임 분 단위 계산]
   │
   ▼
[flow_save_attributes: 자산 누적 다운타임 갱신]
   │
   ▼ SUCCESS
[flow_script_filter: data.shift_downtime_min > 30]
   │
   ▼ TRUE
[flow_send_push: '시프트 다운타임 30분 초과']

Beispiel 15: Automatische Isolation einer Qualitätsfehlerlinie

Bei 5 oder mehr aufeinanderfolgend gemeldeten Fehlern in der Qualitätsprüfung wird der Anlagenlinie ein Stoppbefehl erteilt und der Arbeitsauftrag abgebrochen.

[flow_on_asset_event]   (event_type=QUALITY_FAIL)
   │
   ▼
[flow_script_transform: data.consecutive_fail = (... + 1)]
   │
   ▼
[flow_save_attributes]
   │
   ▼ SUCCESS
[flow_script_filter: data.consecutive_fail >= 5]
   │
   ▼ TRUE
[flow_publish_asset_command: cmd_key='STOP']
   │
   ▼
[flow_abort_work_order: abort_code='QUALITY']
   │
   └── SUCCESS → [flow_send_email: 품질 매니저 + 라인 매니저]

Beispiel 16: Überschreitung des Energieschwellenwerts — Empfehlung zur vorübergehenden Linienstilllegung

Sendet dem Operator eine SMS mit Handlungsempfehlung, wenn der Stundenenergieverbrauch der Linie das Budget übersteigt.

[flow_on_ems_event]
   │
   ▼
[flow_script_filter: data.power_kwh > data.budget_kwh * 1.2]
   │
   ▼ TRUE
[flow_template: '${originator.id} 시간당 ${data.power_kwh}kWh (예산 ${data.budget_kwh}kWh 초과)']
   │
   ▼
[flow_send_sms]
   │
   └── SUCCESS → [flow_save_attributes: 자산에 마지막 경고 시각 기록]

Beispiel 17: Bidirektionale Synchronisation mit externem System — Spiegelung des Arbeitsauftragsstatus

Synchronisiert den vom externen ERP eingehenden Status bidirektional mit dem internen Arbeitsauftragsstatus.

Abwärts (ERP → intern)

[flow_on_mqtt_subscribe: erp/work-order/status]
   │
   ▼
[flow_script_transform: 메시지 → 도메인 매핑]
   │
   ▼
[flow_switch: data.status]
   │
   ├── case "STARTED" ─▶ [flow_start_work_order]
   ├── case "PAUSED"  ─▶ [flow_pause_work_order]
   ├── case "DONE"    ─▶ [flow_end_work_order]
   └── DEFAULT        ─▶ [flow_log: WARN]

Aufwärts (intern → ERP)

[flow_on_entity_event]   (originator.entity_type=Order)
   │
   ▼
[flow_msg_type_filter: type in [ENTITY_UPDATED]]
   │
   ▼ TRUE
[flow_template: ERP 형식으로 변환]
   │
   ▼
[flow_mqtt_publish: erp/work-order/status]

Um bei bidirektionaler Synchronisation Endlosschleifen zu vermeiden, kennzeichnen Sie die Nachrichtenherkunft mit einem Schlüssel wie metadata.source und filtern Sie in der Triggerstufe eigene, selbst publizierte Nachrichten heraus.

Import/Export

Flow-Definitionen können als JSON serialisiert werden, um sie in andere Umgebungen zu portieren oder zu sichern.

Aktion	Ort	Beschreibung
Exportieren	Schaltfläche `내보내기` oben in der Bearbeitungsansicht	Lädt Graph + Nodes + Wires + Node-Konfiguration vollständig als JSON herunter
Importieren	Schaltfläche `가져오기` oben in der Listenansicht	JSON-Text einfügen oder hochladen
Automatischer Snapshot	Automatisch beim Speichern	Wird beim Speichern des Graphen versionsweise abgelegt (für Rollback)

Verhalten beim Importieren

Es wird immer eine neue Flow-ID vergeben (zum Schutz vor Überschreiben der bestehenden ID)

Auch Node-IDs werden neu vergeben, die Wire-Verknüpfungen werden automatisch neu abgebildet

Ein soeben importierter Flow wird im aufgehobenen Zustand abgelegt; der Operator muss ihn nach Prüfung selbst bereitstellen

Fehlerbehandlung

Auf Node-Ebene

Tritt während der Node-Verarbeitung eine Exception auf, erfolgt automatisch eine Weiterleitung über die FAILURE-Relation.
Ist am FAILURE-Ausgang kein Node angeschlossen, wird die Nachricht verworfen (drop) und nur ein Fehlerlog verbleibt.
Nachrichten, die dem *_pattern des Trigger-Nodes nicht entsprechen, werden als SKIPPED behandelt und nicht an nachfolgende Nodes weitergegeben; sie werden auch nicht im Verarbeitungs-/Fehlerzähler mitgezählt.
Alle Fehler werden im Live-Debug-Panel und in der Ausführungshistorie angezeigt.

Auf Flow-Ebene

Element	Standard	Beschreibung
max_depth	100	Begrenzt die kumulierte Anzahl der Node-Besuche während der Verarbeitung einer Nachricht
max_revisit	3	Begrenzt die Anzahl der Wiederbesuche desselben Nodes (verhindert Endlosschleifen bei Zyklen)
flow_timeout_ms	30.000	Erzwungener Abbruch bei Überschreitung der Verarbeitungszeit

Wiederholung bei externen IO-Nodes

HTTP·Kafka·externe DB usw. externe IO-Nodes können mit den Optionen retry_count / retry_delay_ms eine sofortige interne Wiederholung durchführen; bei Fehlschlag aller Wiederholungen erfolgt eine Weiterleitung über die FAILURE-Relation. Falls ein feiner abgestimmter Backoff oder eine EXHAUSTED-Zweigbehandlung benötigt wird, verwenden Sie separat den flow_retry-Node.

Betriebsdiagnose

Administratoren können im Systemmenü den Status der Dispatch-Warteschlange der Flow-Engine einsehen (Worker-Betrieb, Warteschlangengröße, kumulierte Verarbeitungs-/Fehler-/Drop-Anzahl, letzter Fehler).

Der Server prüft im Hintergrund periodisch die Last des Flow-Worker-Pools; bleibt die Last über einen bestimmten Zeitraum bestehen, wird im Betriebsprotokoll eine Statusänderung (HEALTHY → DEGRADED → CRITICAL) in einer Zeile protokolliert. Bei Wiederherstellung des Normalzustands wird zusätzlich ein Wiederherstellungslog protokolliert.

Berechtigungen

Flow führt kein eigenes Berechtigungsmodell ein, sondern nutzt unverändert die bestehende Systemauthentifizierung/-berechtigung.

Funktion	Erforderliche Berechtigung
Listenabfrage / Ausführungshistorie abfragen	Alle authentifizierten Benutzer
Flow erstellen·bearbeiten·bereitstellen·löschen	ADMIN
Import/Export / Alle neu bereitstellen	ADMIN

Betriebsmuster (Recipes)

Eine Bibliothek häufig verwendeter Verdrahtungsformen. Jedes Muster können Sie unverändert kopieren und als Ausgangspunkt eines neuen Flows verwenden.

Muster 1 — Verarbeitung + Benachrichtigungsverzweigung

Doppelte Verzweigung: bei Erfolg Speicherung, bei Fehlschlag Benachrichtigung.

[Action Node]
   ├── SUCCESS → [flow_log] / [flow_save_attributes] / ...
   └── FAILURE → [flow_send_email] / [flow_send_push]

Muster 2 — Sicherer Wiederholungsversuch

Konfiguriert Backoff + EXHAUSTED-Handler, damit externe IO robust gegen vorübergehende Fehler ist.

[risky_node] ─[FAILURE]→ [flow_retry] ─[SUCCESS]→  (risky_node 로 루프)
                                       └[EXHAUSTED]→ [에러 핸들러]

Muster 3 — Lastblockierung durch Vorabfilter

Reduziert nachfolgendes Verarbeitungsvolumen durch Vorfilterung der Nachrichten in der Triggerstufe mit *_pattern. Nicht dem Muster entsprechende Nachrichten werden als SKIPPED behandelt und nicht im Zähler mitgezählt.

[flow_on_tag_point]    (옵션 tag_id_pattern: "MOTOR-*.SPEED")
   │
   ▼
[필터/변환/액션 ...]

Muster 4 — Mehrfachverzweigung nach Case

Verzweigt je nach Status/Typ in mehrere Richtungen.

[flow_switch]   (case별 boolean 표현식)
   ├── case A → [...]
   ├── case B → [...]
   └── DEFAULT → [...]

Muster 5 — Fensteraggregation + einmaliges Emit

Sammelt hochfrequente Eingaben über einen bestimmten Zeitraum und wandelt sie in eine Nachricht um.

[High-rate Trigger]
   │
   ▼
[flow_merge: window_ms=10000]   → data.merged 배열에 누적
   │
   ▼
[flow_script_transform: 요약]
   │
   ▼
[Action / 외부 발송]

Muster 6 — Retry + Umgehung nach Erreichen der Wiederholungsgrenze

Weicht nach Ende der Wiederholungsversuche auf einen alternativen Pfad (andere API, Benachrichtigung, DB-Speicherung) aus.

[Primary HTTP] ─[FAILURE]→ [flow_retry]
                            ├─[SUCCESS]    → (Primary HTTP)
                            └─[EXHAUSTED]  → [Backup HTTP] ─[FAILURE]→ [flow_log/Email]

Muster 7 — Nachfolgende Aktion nach Domänenwechsel

Wandelt einen Tag-bezogenen Alarm in einen Anlagen-bezogenen um und delegiert die Verarbeitung an die jeweilige Anlage.

[flow_on_tag_alarm]
   │
   ▼
[flow_change_originator: Tag → Asset]
   │
   ▼
[자산 단위 액션 (Create Work Order / Publish Asset Event 등)]

Muster 8 — Kombination aus Throttle + Debounce

Throttle auf höchstens 1× pro Minute + Verarbeitung nur bei ausbleibender Änderung über 5 Sekunden.

[High-rate Trigger]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ▼ SUCCESS
[flow_debounce: window_ms=5000]
   │
   ▼
[Action]

Muster 9 — Modularisierung gemeinsamer Logik als Subflow

Trennung in einen Subflow, wenn mehrere Einstiegspunkte dieselbe nachfolgende Verarbeitung (Validierung·Bereinigung·Speicherung) teilen müssen.

메인1: [Trigger A] → [flow_subflow: target=FLOW_99]
메인2: [Trigger B] → [flow_subflow: target=FLOW_99]

서브 (FLOW_99): (트리거 없음)
[검증] → [정제] → [적재]

Muster 10 — Umgehung des direkten Triggers

Um das Ergebnis eines Flows als Eingabe für den nächsten Flow fließen zu lassen, mit flow_dds_publish auf den Domänenkanal publizieren; der andere Flow empfängt denselben Nachrichtentyp als Trigger.

플로우 A: [...] → [flow_dds_publish: type=ASSET_EVENT, originator=...]
플로우 B: [flow_on_asset_event] → [...]

Anwendungsbeispiele (Übersicht)

Szenario	Aufbau
MES-Anbindung	`flow_schedule` → `flow_http_request` → `flow_script_transform` → `flow_create_work_order`
Ereignisweiterleitung nach außen	`flow_on_asset_event` → `flow_msg_type_filter` → `flow_mqtt_publish`
Datenbereinigung·-speicherung	`flow_on_tag_point` → `flow_script_transform` → `flow_save_tag_point`
Alarmautomatisierung	`flow_on_tag_alarm` → `flow_script_filter` → `flow_send_email` + `flow_create_work_order`
Externe DB-Synchronisation	`flow_jdbc_poll` → `flow_script_transform` → `flow_create_asset`
Webhook-Empfang	`flow_on_webhook` → `flow_script_filter` → `flow_publish_asset_command`
Automatische Anpassung des Alarmbands	`flow_on_asset_aggregation` → `flow_script_transform` → `flow_update_tag_alarm_band_numeric`
Automatisierung des Arbeitsauftragsstatus	`flow_on_asset_event` → `flow_switch` → `flow_start/end/pause/resume_work_order`
Zuverlässigkeitssteigerung der API	`flow_http_request` ─FAILURE→ `flow_retry` ─SUCCESS→ Schleife / EXHAUSTED→ Benachrichtigung
Benachrichtigung bei niedrigem OEE	`flow_on_oee_event` → `flow_script_filter` → `flow_template` → `flow_send_push`
Massenregistrierung von Edges	`flow_on_webhook` → `flow_edge_opc_create` → `flow_split` → `flow_edge_tag_create` → `flow_edge_opc_start`
Hochfrequenzbereinigung	`flow_throttle` → `flow_debounce` → Folgeschritt
Aggregation mehrerer Ereignisse	mehrere Trigger → `flow_merge` → Zusammenfassungstransformation → Versand

Schrittweiser Debugging-Leitfaden

Das Standardverfahren, dem Sie folgen sollten, wenn ein Flow nicht wie beabsichtigt funktioniert.

Schritt 1 — Auslöseprüfung

Prüfen Sie in der Listenansicht, ob die Ausführungsanzahl des betreffenden Flows steigt.

Beobachtung	Bedeutung·nächste Aktion
Zähler ist 0	Trigger löst nicht aus. Prüfen Sie den Bereitstellen-Status des Flows + Trigger-Node-Verdrahtung + Option `*_pattern`
Zähler steigt, aber auch die Fehler steigen mit	Fehlschlag im Aktions-Node. Weiter zu Schritt 2
Zähler steigt, aber Folgeverarbeitung erfolgt nicht	Fehlende Verzweigungsverdrahtung. Alle Ausgänge wie `FAILURE`/`THROTTLED`/`EXHAUSTED` überprüfen

Schritt 2 — Nodeweisen Ablauf mit Live-Debug prüfen

Öffnen Sie die Bearbeitungsansicht und aktivieren Sie das Live-Debug-Panel (Aktualisierung alle 2 Sekunden).

Beobachtung	Bedeutung
LED eines bestimmten Nodes bleibt grau	Nachricht erreicht ihn nicht — im vorherigen Node wurde in `FAILURE` verzweigt oder im Filter blockiert
LED rot + ERROR-Zeile	Exception während der Node-Verarbeitung. Feld `data.error` der Nachricht und `NODE_ERROR`-Ereignis in `/flow/log` prüfen
LED grün, aber nächster Node grau	Diskrepanz beim Ausgangs-Relation-Label. Prüfen, ob das Wire-Label genau der Ausgangs-Relation des Nodes (SUCCESS/TRUE usw.) entspricht

Schritt 3 — Nachrichtenverfolgung über die Ausführungshistorie

Verfolgen Sie im Bildschirm /flow/log den gesamten Ablauf einer einzelnen Nachricht anhand der Nachrichten-ID.

Im Nachrichten-ID-Filter die aus Live-Debug gesehene msg_id eingeben
Chronologisch sortieren in der Reihenfolge FLOW_START → NODE_IN → NODE_OUT → FLOW_END
Wenn NODE_ERROR erscheint, ist der error_message dieses Nodes die Ursache

Schritt 4 — Node-Konfiguration prüfen

Häufige Fehler:

Fehler	Prüfung
Tippfehler im `*_field`-Pfad	Stimmt `data.tag_id`? Ist es `metadata.tag_id`? Tatsächlichen Schlüssel über die Nachrichtenvorschau im Live-Debug prüfen
`${...}`-Vorlagenvariable nicht ersetzt	Prüfen, ob der Variablenpfad in der Nachricht existiert, ob kein Tippfehler vorliegt
Timeout bei externem IO	`timeout_ms` erhöhen. Antwortzeit des externen Systems selbst
Berechtigungs-·Auth-Header	Prüfen, ob das JSON-Format von `headers` korrekt ist, ob das Token gültig ist

Schritt 5 — Isolierter Test

Verschieben Sie den problematischen Node allein in einen neuen temporären Flow und lassen Sie mit Testlauf nur eine Nachricht auslösen, um das Ergebnis isoliert zu validieren. Funktioniert er dort normal, ist die Verdrahtung·Nachrichtenform der vorherigen Stufe im ursprünglichen Flow verdächtig.

Schritt 6 — Nach Zurücksetzen der Zähler reproduzieren

Nach Zurücksetzen aller Zähler nur eine Nachricht auslösen — mit sauberer Statistik reproduziert wird das Problem klarer.

Häufige Probleme

Symptom	Ursache·Maßnahme
Trigger löst nicht aus	Prüfen, ob der Flow bereitgestellt ist, ob `_pattern` des Trigger-Nodes nicht zu eng ist. Bei Zeitplan-/externen Abo-Nodes nach Alle neu bereitstellen* erneut versuchen
Live-Debug ist leer	Prüfen, ob der Nachrichtenzähler nicht steigt — nicht dem Trigger-Muster entsprechende Nachrichten (`SKIPPED`) werden nicht im Zähler mitgezählt. Auch prüfen, ob der Pause-Schalter oben im Panel aktiviert ist
Zählerwert kumuliert sich abnormal	Mit der Schaltfläche Alle Zähler zurücksetzen oben rechts in der Node-Konfiguration das Statistikfenster zurücksetzen
Dieselbe Nachricht wird wiederholt verarbeitet	Verdacht auf einen Zyklus. Ein Wiederbesuch desselben Nodes ist nur innerhalb der Standardgrenze von 3× (`max_revisit`) erlaubt. Mit `flow_subflow` trennen und die Eingabemenge über `flow_throttle`/`flow_debounce` anpassen
Aufruf des externen Systems schlägt intermittierend fehl	Backoff-Wiederholung über die Optionen `retry_count`/`retry_delay_ms` des Nodes oder über den Node `flow_retry` verdrahten
Zeitplan funktioniert nach Import nicht	Alle neu bereitstellen oben in der Liste ausführen
Nach Ausführung des Update-Nodes bleiben einige Felder unverändert	Beabsichtigtes Verhalten. Der Update-Node aktualisiert nur die eingegebenen Felder und lässt die übrigen unverändert. Für eine vollständige Aktualisierung Delete + Create kombinieren
Create-Node schlägt mit NOT-NULL-Fehler fehl	Prüfen Sie in der Node-Konfiguration, ob der dynamische Optionspfad `*_field` in der Nachricht tatsächlich existiert. Einige NOT-NULL-Spalten werden automatisch ergänzt, aber Kern-IDs (z. B. `asset_id`, `tag_id`) müssen Sie direkt befüllen
`flow_kafka_publish` / `flow_mqtt_publish` wird nicht publiziert	Verbindungsdaten des externen Brokers (`bootstrap_servers`/`broker_url`) und Topic-Berechtigungen prüfen. `flow_log` daneben verdrahten, um zu prüfen, ob die Nachricht unmittelbar vor der Publikation ankommt
`flow_http_request` antwortet nicht	`timeout_ms` (Standard 5000) erhöhen und den Body direkt als `${msg.data.x}`-Form in `body_template` angeben. Die Antwort wird in `data.response_status` / `data.response` abgelegt
`flow_retry` wird nicht erneut versucht	Fall einer fehlerhaften Verdrahtung. Der `SUCCESS`-Ausgang von `flow_retry` muss zurück zum ursprünglich fehlgeschlagenen Node verdrahtet sein, damit ein erneuter Versuch stattfindet (siehe Muster 2 unten)
Eine Zeile wie `flow_jdbc_poll` wird jedes Mal erneut eingelesen	Die Bedingung `WHERE` der Polling-SQL muss eine Statusaktualisierung nach der Verarbeitung enthalten (z. B. `WHERE sync_status='NEW'` + am Ende desselben Flows `UPDATE ... sync_status='OK'` über `flow_jdbc_query`)
Kann keinen direkten Alarm-Trigger-Node finden	Beabsichtigter Ausschluss. Alarme müssen ausschließlich über den `flow_create_alarm_config`-Pfad entstehen, damit die Konsistenz der Alarmhistorie gewahrt bleibt
Skript-Node schlägt mit einem Zugriffsfehler auf `Java class` fehl	Wird in der Skript-Sandbox blockiert. Für externe Aufrufe verdrahten Sie einen separaten externen Anbindungs-Node
Statusübergangs-Node des Arbeitsauftrags liefert nur `FAILURE`	Der aktuelle Status ist kein Startstatus, der einen Übergang zulässt. Beispiel: `flow_pause_work_order` funktioniert nur im Status `START`. Status vorab mit `flow_check_existence_field` / `flow_script_filter` prüfen
Testlauf funktionierte, aber der Graph blieb im geänderten Zustand	Der Testlauf löst immer die gespeicherte Version aus. Um Änderungen zu validieren, zuerst speichern → dann Testlauf
Importierter Flow ist im inaktiven Zustand	Beabsichtigtes Verhalten. Nach Prüfung den Bereitstellen-Schalter selbst aktivieren

Häufig gestellte Fragen (FAQ)

Häufige Fragen, wenn Operatoren Flow zum ersten Mal nutzen.

F: Kann ich mehrere Trigger in einem Flow platzieren? A. Ja. Wenn Sie mehrere Trigger-Nodes im selben Flow platzieren, werden alle zu Einstiegspunkten und lösen jeweils unabhängig aus. In Kombination mit flow_merge können Sie mehrere Ereignistypen zu einer gemeinsamen Folgeverarbeitung zusammenführen.

F: Kann ich auch einen Flow ohne Trigger erstellen? A. Ja. Beim Speichern erscheint eine Warnung, aber der Flow wird gespeichert. Sie können ihn als „Bibliotheks"-Form nutzen, die nur über Testlauf oder den Aufruf flow_subflow eines anderen Flows ausgelöst wird.

F: Kann eine Nachricht mehrere Nodes gleichzeitig durchlaufen? A. Ja. Wenn Sie am Ausgangsport eines Nodes mehrere Wires verbinden, wird dieselbe Nachricht gleichzeitig verzweigt und an alle nachfolgenden Nodes weitergeleitet.

F: Kann der Transformations-Node die Nachricht selbst erzeugen und in eine andere Nachricht umwandeln? A. Ja. In flow_script_transform können msg.type, msg.originator, msg.data, msg.metadata alle frei geändert werden. Selbst wenn Sie die Nachricht wie einen anderen Triggertyp aussehen lassen, wird dadurch nicht automatisch ein anderer Flow aufgerufen (die ursprüngliche Trigger-Zuordnung bleibt erhalten). Um einen anderen Flow aufzurufen, verwenden Sie flow_subflow oder flow_dds_publish.

F: Wie viele automatische Snapshots werden aufbewahrt? Kann ich sie selbst wiederherstellen? A. Bei jedem Speichern des Graphen wird automatisch einer abgelegt, die Aufbewahrungsrichtlinie richtet sich nach der Betriebsumgebungskonfiguration. Eine UI zur direkten Wiederherstellung im Bildschirm wird nicht angeboten; bei Bedarf können Sie den Administrator bitten, auf eine bestimmte Version zurückzusetzen. Am sichersten ist es, vor Änderungen mit 내보내기 die JSON-Datei extern zu sichern.

F: Wie lange wird die Ausführungshistorie aufbewahrt? A. Standardmäßig 7 Tage. Für eine langfristige Aufbewahrung nutzen Sie flow_kafka_publish oder flow_jdbc_query, um die Daten in ein externes Log-System zu laden.

F: Ich möchte nur die Statistik eines Flows separat zurücksetzen. A. Verwenden Sie Fehlerzähler zurücksetzen (nur Fehler) oder Alle Zähler zurücksetzen (gesamt) oben rechts in der Node-Konfiguration der Bearbeitungsansicht. Andere Flows werden davon nicht beeinflusst.

F: Ich möchte einen Flow in eine andere Umgebung (Entwicklung/Staging/Produktion) verschieben. A. Laden Sie die JSON über 내보내기 in der Bearbeitungsansicht herunter und laden Sie sie über 가져오기 in der Listenansicht der Zielumgebung hoch. Ein importierter Flow wird immer mit einer neuen ID + im aufgehobenen Zustand abgelegt, sodass der Operator ihn nach Prüfung selbst bereitstellen kann.

F: Dieselbe Payload wird zweimal verarbeitet. Wie kann ich das verhindern? A. Grenzen Sie in der Triggerstufe mit *_pattern ein, oder begrenzen Sie die Häufigkeit mit flow_throttle/flow_debounce. Bei externen Eingaben (Webhook·MQTT) verwenden Sie auf der Sendeseite einen Idempotenzschlüssel und filtern Duplikate mit flow_check_existence_field oder einem Filter auf Basis der Nachrichten-ID.

F: Ist eine zyklische (Loop-)Verdrahtung sicher? A. Ein beabsichtigter Zyklus wie flow_retry ist sicher. Nicht beabsichtigte Zyklen werden durch max_revisit (Standard 3×) automatisch blockiert. Es empfiehlt sich jedoch, im laufenden Betrieb den Ablauf mit flow_log zu visualisieren, um unbeabsichtigte Zyklen frühzeitig zu erkennen.

F: Werden Flow-Änderungen in Echtzeit übernommen? A. Sie werden sofort beim Speichern des Graphen wirksam. Trigger mit eigenem Scheduler wie Zeitplan·externes Abonnement·externes DB-Polling müssen jedoch über Alle neu bereitstellen oder den Wechsel Aufheben→Bereitstellen erneut registriert werden.

F: Kann ich die ID eines bereits registrierten Nodes ändern? A. Die Node-ID wird vom System automatisch vergeben. Der Anzeigename kann im Konfigurationsformular des Nodes geändert werden, und dieser wird in Live-Debug·Ausführungshistorie angezeigt.

F: Können nicht berechtigte Benutzer einen Flow versehentlich ändern? A. Erstellen·Bearbeiten·Bereitstellen·Löschen eines Flows erfordert die ADMIN-Berechtigung. Normale Benutzer können nur die Listenabfrage und die Abfrage der Ausführungshistorie durchführen.

Bewährte Betriebspraktiken

Grundsätze, die Sie beim sicheren Betrieb von Flow in der Produktion beachten sollten.

Lastmanagement

Filtern Sie in der Triggerstufe mit *_pattern vorab, um das nachfolgende Verarbeitungsvolumen zu reduzieren. Tag-Points, die jede Minute zehntausendfach eingehen, sind der teuerste Einstiegspunkt.
Kombinieren Sie externe Systemaufrufe (flow_http_request/flow_jdbc_query usw.) mit flow_throttle / flow_debounce, um die externe Last zu steuern.
Wenn mehrere Folge-Nodes an derselben Eingabe hängen müssen, trennen Sie diese über flow_subflow, um die Gesamtzahl der Graph-Nodes zu reduzieren. Je mehr Nodes vorhanden sind, desto höher auch die Kosten des Live-Debug-Pollings.
Aktivieren Sie den Debug-Modus nur in der Betriebsvalidierungsphase. Je mehr Daten geladen werden, desto schneller füllt sich das 7-Tage-Fenster der Ausführungshistorie.

Sicheres Schreiben

Verdrahten Sie an jedem Aktions-Node unbedingt einen FAILURE-Zweig (mindestens ein flow_log). Ist FAILURE leer, wird die Nachricht stillschweigend verworfen, was die Fehlerverfolgung erschwert.
Kombinieren Sie externe IO-Nodes nach Möglichkeit mit flow_retry, um vorübergehende Störungen abzufangen.
Der Update-Node aktualisiert nur die eingegebenen Felder (Teilaktualisierung). Für eine vollständige Überschreibung nutzen Sie die Kombination Delete + Create.
Ein Graph ohne Trigger-Node dient ausschließlich für manuellen Testlauf. Prüfen Sie beim Speichern die Warnung, um nicht versehentlich einen Trigger zu vergessen.
Erstellen Sie Zyklen (Verdrahtungen, die zum selben Node zurückführen) nur in beabsichtigter Form wie flow_retry; ansonsten schützt max_revisit, aber visualisieren Sie verdächtige Graphen mit flow_log.

Änderungsverfahren (Change Management)

Empfohlene Reihenfolge bei Änderungen an einem produktiv laufenden Flow.

Sicherung — Laden Sie den aktuellen Graphen mit 내보내기 in der Bearbeitungsansicht als JSON-Datei herunter und bewahren Sie ihn auf (automatische Snapshots werden ebenfalls abgelegt, aber eine externe Aufbewahrung ist sicherer).
Duplizieren oder in neuem Flow arbeiten — Nehmen Sie Änderungen eher in einem neuen Flow (oder einer per Import erstellten Kopie) vor und validieren Sie diese, statt den laufenden Flow sofort zu bearbeiten.
Testlauf — Injizieren Sie direkt JSON-Nachrichten, um alle Zweige (SUCCESS/FAILURE/EXHAUSTED usw.) einmal auszulösen, und prüfen Sie das Ergebnis in der Ausführungshistorie.
Übernahme in den Betrieb — Übertragen Sie die validierte Graph-JSON via 내보내기 → in der Produktionsumgebung 가져오기 → nach Prüfung durch den Operator den Bereitstellen-Schalter.
Rollback bei Problemen — Sofort mit dem Aufheben-Schalter deaktivieren. Da automatische Snapshots abgelegt wurden, kann ein Entwickler beauftragt werden, auf die vorherige Version zurückzusetzen.

Nutzung von Zählern·Statistiken

Wenn die Verlaufs-Sparkline in der Listenansicht plötzlich abflacht, prüfen Sie eine mögliche fehlende Triggerauslösung oder SKIPPED-Behandlung.
Bei einem abnormalen Fehlerraten-Donut-Diagramm sollten Sie den *_field-Pfad in der Node-Konfiguration überprüfen. Fehlt ein Schlüssel in der Nachrichten-Payload, kommt es häufig zu Fehlschlägen.
Setzen Sie nach der Betriebsvalidierung das Statistikfenster mit Alle Zähler zurücksetzen zurück, um die normale Baseline neu zu messen.

Notfallmaßnahmen

Schrittweises Verfahren, das Sie bei Problemen im laufenden Betrieb schnell anwenden sollten.

Szenario 1 — Bestimmter Flow läuft Amok (Nachrichtenflut)

Symptom: Die Ausführungsanzahl eines Flows steigt gegenüber dem Normalwert um das Zehn- bis Hundertfache, gleichzeitig steigen auch die Fehler

Maßnahme

Deaktivieren Sie den betreffenden Flow sofort mit dem Aufheben-Schalter (in der Listenansicht 1 Sekunde)
Prüfen Sie im Live-Debug-Panel·Ausführungshistorie, welcher Trigger die Flut verursacht hat
Grenzen Sie die Option *_pattern des Trigger-Nodes ein oder verdrahten Sie unmittelbar danach flow_throttle / flow_debounce
Bei Bedarf mit flow_check_existence_field oder flow_msg_type_filter auf bestimmte Nachrichtentypen begrenzen
Nach der Korrektur mit Testlauf reproduzieren → nach Bestätigung des Normalzustands Bereitstellen wieder aufnehmen

Szenario 2 — Massenausfall durch Störung eines externen Systems

Symptom: Fehler an externen IO-Nodes wie HTTP/Kafka/externer DB häufen sich gleichzeitig

Maßnahme

Bei größerem Auswirkungsbereich die betroffenen Flows gebündelt aufheben (in der Listenansicht mehrfach auswählen und gebündelt aufheben)
Wiederherstellung des externen Systems bestätigen
Bei externen IO-Nodes ohne flow_retry-Verdrahtung diese ergänzen
Bei verlangsamter Antwort des externen Systems timeout_ms anpassen
Je nach Betriebsumgebung Alle neu bereitstellen ausführen und danach Bereitstellen wieder aufnehmen

Szenario 3 — Endlosschleife durch Zyklus

Symptom: Eine einzelne Nachricht durchläuft kontinuierlich denselben Node, Verarbeitungszeit kumuliert sich

Maßnahme

Durch den max_revisit-Schutz wird zwar automatisch nach maximal 3 Wiederbesuchen blockiert, aber prüfen Sie zur Betriebssicherheit trotzdem nach dem Aufheben
Verfolgen Sie den Zyklus visuell im Graphen (Wires in der Bearbeitungsansicht nachverfolgen)
Bei beabsichtigter Schleife (flow_retry) prüfen, ob max_attempts angemessen ist
Bei unbeabsichtigtem Zyklus die Verdrahtung entfernen oder mit flow_check_relation eine Verzweigung hinzufügen
Nach der Korrektur Alle Zähler zurücksetzen → erneut bereitstellen

Szenario 4 — Verdacht auf Datenbeschädigung (fehlerhafte automatische Aktualisierung)

Symptom: Domänendaten wurden durch Automatisierung unbeabsichtigt aktualisiert

Maßnahme

Betreffenden Flow sofort aufheben
Extern gesicherte vorherige Graph-JSON heranziehen oder die vorherige Version aus dem automatischen Snapshot prüfen (mit Unterstützung des Administrators)
Graph analysieren: unbeabsichtigte Update-Node-Verdrahtung·fehlerhafter *_field-Pfad·Fehler in der Skript-Transformation prüfen
Betroffene Domänendaten über ein separates Verfahren im Back-Office-Bildschirm korrigieren
Den korrigierten Graphen mit Testlauf validieren und erneut bereitstellen

Szenario 5 — Stopp während Systeminspektion·Wartung

Symptom: Externe IO-Aufrufe sollen während der Wartungszeit eines externen Systems vorübergehend gestoppt werden

Maßnahme

Betroffene Flows gebündelt aufheben (Mehrfachauswahl in der Liste)
Nach Ende der Wartung Bereitstellen wieder aufnehmen
Bei Triggern mit eigenem Scheduler (Zeitplan·externes MQTT-Abonnement·externes DB-Polling) einmal Alle neu bereitstellen ausführen und die ordnungsgemäße Registrierung bestätigen

Szenario 6 — Live-Debug-Warteschlange voll

Symptom: Warteschlangengröße in der Betriebsdiagnoseseite nahe der Schwelle, Drop-Anzahl steigt

Maßnahme

Reduzieren Sie die Anzahl der Flows mit aktiviertem Debug-Modus und die geladene Menge (schalten Sie den Debug-Modus bei validierten Flows aus)
Reduzieren Sie mit *_pattern in der Triggerstufe das nachfolgende Verarbeitungsvolumen selbst
Wenn das System automatisch in den Erholungsmodus wechselt, wird im Lastwächter ein Log mit DEGRADED/CRITICAL protokolliert — teilen Sie dies dem Administrator mit

Bei allen Szenarien lautet die Priorität stets sofort aufheben → Ursache klären → korrigieren → validieren → erneut bereitstellen. Beachten Sie ergänzend auch das Änderungsverfahren.

Sicherheit·Umgang mit sensiblen Daten

Da Flow sensible Ein-/Ausgaben wie externe API-Aufrufe·E-Mail/SMS-Versand·Webhook-Empfang verarbeitet, beachten Sie bitte folgende Grundsätze.

Token·Zugangsdaten

Geben Sie API-Token·Passwörter nicht im Klartext direkt in der Node-Konfiguration ein. Verwenden Sie eine Umgebungsvariablen-Ersetzung in der Form ${ENV_VAR}, um sie aus dem Geheimnisspeicher der Betriebsumgebung einzuspeisen.
Trennen Sie die Token-Position innerhalb der headers-JSON möglichst als Umgebungsvariable ab, damit das Token in der Produktionsumgebung nicht offengelegt wird.

// 권장
{ "Authorization": "Bearer ${MES_TOKEN}" }

// 비권장
{ "Authorization": "Bearer eyJhbGciOi..." }

Schutz des Webhook-Eintrittspunkts

flow_on_webhook ist ein interner Einstiegspunkt, der von authentifizierten Benutzern aufgerufen wird; wenn Sie ihn jedoch extern zugänglich machen, verdrahten Sie unbedingt einen Filter zur Payload-Validierung.

// flow_script_filter — 토큰 + 필수 필드 검사
if (data.token !== '${WEBHOOK_TOKEN}') return false;
if (!data.asset_id || !data.cmd_key)   return false;
true

Maskierung sensibler Daten

Im Live-Debug-Panel und in der Ausführungshistorie wird eine Nachrichtenvorschau angezeigt. Maskieren Sie personenbezogene Daten (Name·Kontakt·Konto) und Token unmittelbar vor der Speicherung.

// 디버그/로그 적재 직전 변환
data.email_masked = data.email
  ? data.email.replace(/(.{2}).+(@.+)/, '$1***$2') : null;
delete data.email;
delete data.token;
msg

Verarbeitung des Antwortkörpers bei externem IO

Bei data.response von flow_http_request wird der gesamte Antwortkörper geladen. Enthält die Antwort sensible Daten, extrahieren Sie sofort mit einem nachfolgenden Transformations-Node nur die benötigten Schlüssel und entfernen Sie das Original.

// 응답에서 필요한 필드만 보존
data = { id: data.response_obj.id, status: data.response_obj.status };
msg

Isolation der Skript-Nodes

Skript-Nodes werden in einer sicheren Sandbox ausgeführt, in der Datei-, Netzwerk- und beliebiger Klassenzugriff blockiert sind. Verdrahten Sie bei Bedarf für externe Aufrufe unbedingt einen separaten externen Anbindungs-Node.

Flow-Metriken und Alarme

Automatisch erfasste Kennzahlen im laufenden Flow-Betrieb — wo sie einzusehen sind und wie sich Alarme darauf einrichten lassen.

Flow-bezogene KPIs

In jeder Zeile der Listenansicht (/flow/index) werden folgende KPIs angezeigt.

Kennzahl	Bedeutung	Abnormales Signal
Gesamtausführungsanzahl	Anzahl der durch Triggerauslösung einmal durchlaufenen Graph-Durchläufe (Summe aus Erfolg/Fehler)	Starker Rückgang gegenüber üblich → Trigger tot / starker Anstieg → Amoklauf
Fehleranzahl	Anzahl der Fälle, in denen an irgendeinem Node im Graphen in den `FAILURE`-Zweig gefallen wurde	Bei ≥5% des Gesamtwerts überprüfen
Fehlerrate	`에러 / 전체 × 100`	Bei Überschreiten eines Schwellenwerts einen Alarm einrichten
Letzte Ausführung	Zeitpunkt der letzten Auslösung	„Keine Auslösung seit mehr als 5 Minuten" ist nur normal, wenn dies erwartet ist
Durchschnittliche Verarbeitungszeit	Durchschnittliche ms, die eine Nachricht benötigt, um den gesamten Graphen zu durchlaufen	Große Änderung bei Hinzufügen/Entfernen externer IO-Nodes
Trend der letzten Verarbeitungszeit	5-Minuten-Sparkline — Live-Debug-Panel	Bei Spikes die Antwort des externen Systems prüfen

Node-bezogene KPIs

Wenn Sie in der Bearbeitungsansicht auf einen Node klicken, wird Folgendes unten rechts am Node angezeigt.

[Node Name]
처리 999 · 에러 3 · 평균 12ms · 최근 18ms

Ort	Anzeige
LED oben	Grau (wartend) / Grün (in Verarbeitung) / Rot (Fehler)
Label unten	`처리 N · 에러 M · 평균 Xms · 최근 Yms`

4 Node-Statistiken

Zähler	Bedeutung
Verarbeitung	Anzahl der Nachrichten, die in den Node eingingen und über einen normalen Zweig verlassen wurden
Fehler	Anzahl der Fälle mit `FAILURE`-Zweig oder Exception
Durchschnittliche Verarbeitungszeit	Verarbeitungszeit des Nodes selbst in ms (inkl. externer IO)
Letzte Verarbeitungszeit	Verarbeitungszeit des zuletzt verarbeiteten Falls in ms

Systemweite Metriken

Die Metriken der gesamten Flow-Engine können Sie im Bildschirm System → Monitoring einsehen.

Die Metriken werden unter der JMX-Domäne plantpulse.core.engine als folgende fünf Metriken offengelegt.

Metrik	Art	Bedeutung	Ablesen
`FLOW_EXECUTOR_QUEUE_SIZE`	Gauge	Warteschlangengröße des Flow-Ausführungspools	Kontinuierliches Anwachsen bedeutet, dass die Verarbeitung nicht mit dem Eingang mithält
`FLOW_EXECUTOR_ACTIVE_COUNT`	Gauge	Anzahl aktiver Threads im Ausführungspool	Sättigung, wenn nahe der Poolgröße
`FLOW_EXECUTOR_DROPPED_COUNT`	Gauge (kumuliert)	Kumulierte Anzahl verworfener Tasks durch Pool-Sättigung	Ungleich 0 bedeutet Nachrichtenverlust — der Wert, den Sie zuerst prüfen sollten
`FLOW_DEBUG_QUEUE_SIZE`	Gauge	Wartewarteschlangengröße für die Debug-Ereignisspeicherung	Wird groß, wenn viele Flows den Debug-Modus aktiviert haben
`FLOW_EXECUTION_TIME`	Timer	Verteilung der Verarbeitungszeit pro Flow	Verfolgung von Durchschnitt·Maximum

Es gibt keine Metrik namens flow.engine.*

Ein älteres Dokument enthielt eine Tabelle mit flow.engine.queue_depth · in_flight · dispatch_lag_ms · exec_p95_ms · failed_per_min · script_timeout_per_min, aber Metriken mit diesen Namen existieren nicht. Die genannten fünf sind alle, die es gibt.

(flow.engine.enabled ist keine Metrik, sondern ein Konfigurationsschlüssel von engine.properties — das Master-Gate zum Abschalten der gesamten Flow-Engine.)

Muster für metrikbasierte Alarmeinrichtung

4 Muster zur Überwachung der eigenen Flow-Metriken über EQL-Alarm oder CEP → Trigger.

Muster A — Überschreitung des Fehlerratenschwellenwerts

context EVERY_1_MINUTES
SELECT flow_id, count(CASE WHEN status='FAILURE' THEN 1 END) * 100.0 / count(*) AS err_pct
FROM   AssetEvent.win:time(5 min)
WHERE  event_type = 'FLOW_EXEC'
GROUP  BY flow_id
HAVING err_pct > 5

Muster B — Backpressure in der Verarbeitungswarteschlange

Alarm, wenn flow.engine.queue_depth > 500 länger als 1 Minute anhält — eine Situation, in der die Auslösegeschwindigkeit des Triggers die Verarbeitungsgeschwindigkeit übersteigt.

Muster C — Ausbleiben der Auslösung eines bestimmten Flows

SELECT * FROM pattern [
  every a = AssetEvent(event_type='FLOW_EXEC', flow_id='my-flow')
        -> ( timer:interval(15 min)
             and not AssetEvent(event_type='FLOW_EXEC', flow_id=a.flow_id) )
]

Alarm, wenn ein Flow, der normalerweise N-mal pro Minute auslöst, 15 Minuten oder länger schweigt.

Muster D — Explosion von Timeouts bei externem IO

context EVERY_5_MINUTES
SELECT flow_id, node_id, count(*) AS timeout_count
FROM   Log.win:time(5 min)
WHERE  module = 'flow-engine' AND code = 'NODE_TIMEOUT'
GROUP  BY flow_id, node_id
HAVING count(*) > 10

Empfehlung für Alarmausgabekanäle

Alarmtyp	Empfohlener Kanal
Fehlerrate·Auslöseausfall (direkter Betrieb)	E-Mail + Push-Benachrichtigung
Warteschlangen-Backpressure·Timeout-Explosion (System)	Slack/Teams-Webhook
Automatische Diagnose (Referenz)	Nur Diagnoselog

⚠️ Verlassen Sie sich bei einem Alarm-Flow selbst niemals auf Metriken, für die Alarme eingerichtet sind — stirbt der Alarm-Flow, kommt der Alarm selbst nicht mehr. Überwachen Sie den Alarm-Flow über einen externen Health-Check unter System → Monitoring.

Nachrichtenverarbeitungssemantik und Backpressure

Zusicherungsniveau und Backpressure-Verhalten bei der Nachrichtenverarbeitung durch die Flow-Engine.

Zustellungsgarantie — At-Least-Once

Die Flow-Engine garantiert At-Least-Once-Zustellung.

Fall	Verhalten
Normale Verarbeitung	Einmal ausgelöst → einmal Graph durchlaufen → einmal abgeschlossen
Neustart der Engine während der Verarbeitung	In der Trigger-Warteschlange verbliebene unverarbeitete Nachrichten werden nach dem nächsten Hochfahren erneut verarbeitet
Node-bezogene Exception	Übergang nur in den `FAILURE`-Zweig. Die Nachricht geht nicht verloren
Timeout bei externem IO	`FAILURE`-Zweig + automatische Wiederholung mit `flow_retry` möglich

Möglichkeit von Duplikaten — Es handelt sich nicht um Exactly-Once. Wird flow_create_* zwischendurch unterbrochen und erneut versucht, kann dieselbe ID zweimal eingehen — verwenden Sie deshalb on_duplicate=skip oder den Idempotenzschlüssel des externen Systems.

Muster für Idempotenzschlüssel

Muster zur Sicherstellung der Idempotenz beim Aufruf externer Systeme aus einem Flow.

// flow_script_transform — 멱등성 키 생성
msg.data.idempotency_key =
  msg.metadata.asset_id + '|' +
  msg.metadata.ts + '|' +
  msg.data.event_type;

Beim anschließenden externen Aufruf im Header anhängen:

"headers": { "Idempotency-Key": "${data.idempotency_key}" }

Verarbeitungsreihenfolge — Trigger-bezogenes FIFO

Nachrichten desselben Originators	Andere Originators
Trigger-bezogenes FIFO (in Ankunftsreihenfolge)	Parallele Verarbeitung (unabhängig von der Reihenfolge)

Selbst wenn 2 Ereignisse der Anlage A fast gleichzeitig auftreten, werden die Ereignisse von A in der Reihenfolge ihres Auftretens verarbeitet. Ereignisse der Anlage A und der Anlage B können in unterschiedlichen Workern parallel verarbeitet werden.

Backpressure

Verhalten, wenn die Verarbeitungsgeschwindigkeit nicht mit der Auslösegeschwindigkeit mithält.

Situation	Verhalten
Warteschlangentiefe < 80%	Normal — neuer Trigger wird sofort enqueued
Warteschlangentiefe 80%–100%	Diagnose-WARN-Log tritt auf — Warteschlange nimmt weiterhin auf
Warteschlangentiefe = 100% (voll)	Neuer Trigger wird verworfen + Diagnose-ERROR-Log tritt auf

Was der Operator tun sollte, wenn die Warteschlange voll ist

Prüfen Sie flow.engine.queue_depth unter System → Monitoring
Identifizieren Sie den Flow mit explodierter Ausführungsanzahl in der Listenansicht
Grenzen Sie den Vorabfilter des Triggers (*_pattern) dieses Flows ein, um die Last zu reduzieren
Oder heben Sie die Bereitstellung des betreffenden Flows vorübergehend auf, um im Notfall zu handeln

Circuit Breaker (externes IO)

Externe IO-Nodes (HTTP/Kafka/MQTT/E-Mail) werden bei Erfüllung folgender Bedingungen 30 Sekunden lang gebündelt blockiert.

Bedingung	Schwelle
Aufeinanderfolgende Fehlschläge in der letzten Minute	≥ 10
Durchschnittliche Antwortzeit	≥ 10 Sekunden

Während der Blockade werden alle an diesen Node eingehenden Nachrichten sofort in den FAILURE-Zweig verzweigt. Erholt sich das externe System, wird die Blockade automatisch aufgehoben. Dieses Verhalten dient dazu, eine externe Störung zu isolieren, damit sie die Flow-Engine selbst nicht lahmlegt.

Der Zeitpunkt, an dem der Circuit geöffnet wurde, wird im Diagnoselog als code = CIRCUIT_OPENED protokolliert. Hat sich das externe System schnell erholt, können Sie über metadata.retry_count in der Ausführungshistorie verlorene Nachrichten manuell erneut ausführen.

Vermeidung von Zyklen

Wenn im Flow ein Zyklus wie folgt entsteht, kann eine Endlosschleife auftreten.

A → B → C → A  (잘못된 결선)

Die Flow-Engine blockiert Zyklen mit zwei Verteidigungslinien.

Verteidigungslinie	Verhalten
`max_depth=100`	Erzwungenes Ende nach Durchlaufen von 100 Nodes
`max_revisit=3`	Nachricht wird beim 4. Besuch desselben Nodes verworfen + Diagnose-ERROR

Die Schleife SUCCESS-Zweig von flow_retry → ursprünglicher Node ist ein beabsichtigter Zyklus; da metadata.retry_count dabei mitsteigt, endet dieser normal, bevor max_revisit greift.

Was man in der Ausführungshistorie sieht

Der „Fehlercode-Katalog" eines älteren Dokuments existierte nicht wirklich

Codes wie NODE_SCRIPT_TIMEOUT · FLOW_TIMEOUT · FLOW_MAX_DEPTH · TRIGGER_QUEUE_FULL · WEBHOOK_AUTH_FAIL · DOMAIN_DUP_KEY waren in einer Tabelle aufgeführt, aber das Produkt erzeugt diese Zeichenketten nirgendwo. Suchen Sie im Log nach diesem Code, erhalten Sie für immer 0 Treffer.

Was die Flow-Engine tatsächlich protokolliert, sind die folgenden 5 Ereignistypen.

In der Flow-Ausführungshistorie (mm_flow_log, linkes Menü Automation > Flow-Ausführungshistorie) werden fünf Arten von Ereignissen gespeichert.

Ereignistyp	Wann	Speicherbedingung
`FLOW_START`	Beim Eintritt in den Flow	Immer
`FLOW_END`	Beendigung der Flow-Verarbeitung (Erfolg·Timeout beide)	Immer
`NODE_IN`	Wenn ein Node eine Nachricht empfängt	Nur im Debug-Modus
`NODE_OUT`	Wenn ein Node eine Nachricht ausgibt	Nur im Debug-Modus
`NODE_ERROR`	Exception während der Node-Verarbeitung	Immer

Felder, die jede Zeile gemeinsam hat.

Feld	Inhalt
`level`	`INFO` / `WARN` / `ERROR`
`flow_id` · `flow_node_id` · `node_name` · `node_type`	Welcher Flow, welcher Node
`msg_id`	Zur nachrichtenbezogenen Verfolgung — um den Ablauf einer Nachricht nachzuverfolgen, gruppieren Sie nach diesem Wert
`message` · `error_message`	Für Menschen lesbare Beschreibung, Fehlermeldung
`duration_ns`	Node-Ausführungszeit (Nanosekunden) — nur bei `NODE_OUT` · `NODE_ERROR` von Bedeutung

Normalerweise werden keine nodebezogenen Logs gespeichert

NODE_IN · NODE_OUT werden nur im Debug-Modus gespeichert. Dass in der normalen Ausführungshistorie nur FLOW_START · FLOW_END zu sehen sind, ist normal. Um Node für Node nachzuverfolgen, aktivieren Sie den Debug-Modus des betreffenden Flows — dies erhöht jedoch die Logmenge erheblich.

Suche nach Symptomen

Statt vom Code auszugehen, gehen wir vom Symptom aus.

Symptom	Zuerst prüfen
Trigger löst aus, aber es scheint kein Node zu laufen	Ob der Flow bereitgestellt (Schalter ON) ist · ob `*_pattern` des Triggers die Nachricht nicht herausfiltert
Verarbeitung bricht mittendrin ab	Überschreitung von 30 Sekunden (`flow_timeout_ms`) pro Nachricht. Prüfen Sie die Warnung `FlowExecutor timeout` im Serverlog
Dieselbe Nachricht wird wiederholt verarbeitet	Zyklus. Wird durch `max_revisit` (3×) automatisch blockiert, prüfen Sie dennoch die Verdrahtung
Graph ist zu lang und endet nicht	Überschreitung von `max_depth` (100). In Subflows aufteilen
Node wirft eine Exception	Prüfen Sie `error_message` der `NODE_ERROR`-Zeile in der Ausführungshistorie
Externer Aufruf schlägt fehl	Prüfen Sie `data.response_status` · `data.error` des betreffenden Nodes im nachfolgenden Node

Die Grenzwerte (100 · 3 · 30.000 ms) sind die Standardwerte von ExecutionState.

Cluster·Hochverfügbarkeits(HA)-Verhalten

Ist die Plattform als Cluster-Umgebung installiert, verhält sich die Flow-Engine gemäß folgenden Regeln verteilt.

Trennung der Node-Rollen

Rolle	Verhalten
Leader	Der einzelne Node, der Änderungen am Flow-Graphen (Speichern/Bereitstellen/Aufheben) seriell verarbeitet
Worker	Allgemeine Nodes, die die Triggerauslösung·Nachrichtenverarbeitung parallel ausführen (alle Nodes fungieren gleichzeitig als Worker)
Scheduler	Zuständig für die `flow_schedule` Cron-Bewertung — derselbe Node wie der Leader

Der Leader-Node wird beim Hochfahren der Plattform automatisch gewählt; fällt der Leader aus, wird automatisch einer der übrigen Nodes zum Leader befördert. Der Operator muss dies nicht manuell festlegen.

Nachrichtenverteilung — konsistentes Routing auf Anlagenebene

Verteilungsschlüssel	Verhalten
`originator.id` (Anlagen-/Tag-/Bestellungs-ID)	Nachrichten derselben Anlage werden immer an denselben Worker geroutet — Sequenzgarantie
Externer Eintritt (`flow_on_webhook` usw.)	Round-Robin-Verteilung

Diese Regel verhindert, dass Ereignisse der Anlage A gleichzeitig auf mehreren Workern verarbeitet werden und die Reihenfolge durcheinandergerät. Im Ergebnis gilt auf Anlagenebene die Garantie einer Verarbeitung durch einen einzigen Worker, während zwischen Anlagen gleichzeitig eine parallele Verarbeitung gilt.

Verhalten bei Leader-Ausfall — Failover

[Leader 다운 t=0]
        ↓
[다른 노드가 리더 승격 시도 t=0~3s]
        ↓
[새 리더 확정 t=3~5s]  ← cron 스케줄·플로우 배포 변경 재개
        ↓
[기존 워커들은 정상 동작 유지 — 트리거 처리 영향 없음]

Phase	Auswirkung
0–3 Sekunden	Auslösung von `flow_schedule` vorübergehend gestoppt / Triggerverarbeitung nicht betroffen
3–5 Sekunden	Neuer Leader bestimmt, Zeitplan wird fortgesetzt
Ab 5 Sekunden	Normal

Die Cron-Auslösung wird gemäß der misfire-Richtlinie sofort für eine einmal ausgefallene Auslösung nachgeholt. Im laufenden Betrieb kann bei einer Bewertungseinheit von flow_schedule unter 5 Sekunden ein kurzzeitiges Ausbleiben sichtbar werden.

Persistenz der Trigger-Warteschlange

Element	Verhalten
Position der Trigger-Warteschlange	In-Memory-Warteschlange + permanenter Speicher (Transaktionsprotokoll)
Bei Neustart des Nodes	In der Warteschlange verbliebene unverarbeitete Nachrichten werden nach dem nächsten Hochfahren erneut dispatcht
Node-Ausfall während der Verarbeitung	Diese Nachricht wird erneut verarbeitet, aber mit der At-Least-Once-Garantie muss beim externen System ein Idempotenzschlüssel vorhanden sein

Checkliste für Operatoren bei Cluster-Bereitstellung

Zeitsynchronisation (NTP) aller Nodes — die Auslösezeitpunkte der Trigger stimmen zwischen Nodes überein
Externe Systeme (MQTT/Kafka-Broker) an einem Netzwerkstandort platzieren, den alle Nodes erreichen können
Die SMTP-Konfiguration von flow_send_email nur einmal in den Systemeinstellungen registrieren — wird von allen Nodes gemeinsam genutzt
Worker-Poolgröße (flow.engine.workers) an die Anzahl der CPU-Kerne pro Node anpassen

Single-Node-Modus (Entwicklung·Kleinbetrieb)

Leader·Worker·Scheduler laufen alle in einem Prozess
Kein Failover — fällt der Node aus, stoppt der Flow selbst
Trigger-Warteschlange wird durch In-Memory + Disk gesichert, sodass eine Wiederherstellung nach Neustart möglich ist

End-to-End-Trace

Methode zur nachträglichen Verfolgung, wie eine Nachricht den gesamten Graphen durchlaufen hat.

Automatische Vergabe einer Korrelations-ID

Die Flow-Engine vergibt automatisch eine Korrelations-ID (correlation ID) an alle durch Trigger ausgelösten Nachrichten.

{
  "type": "POST_TELEMETRY",
  ...,
  "metadata": {
    "trace_id":   "tr-a1b2c3d4-e5f6-7890-...",
    "span_id":    "sp-01",
    "parent_id":  null,
    ...
  }
}

Feld	Bedeutung
`metadata.trace_id`	Eine einzigartige ID für eine gesamte Triggerauslösung — von allen Nodes im Graphen durchlaufende Nachrichten geteilt
`metadata.span_id`	Nodebezogene eindeutige ID — wird bei jedem Node-Durchlauf aktualisiert
`metadata.parent_id`	`span_id` des vorherigen Nodes

Suche nach trace_id in der Ausführungshistorie

Wenn Sie im Ausführungshistorie-Bildschirm die ID in das Eingabefeld trace_id einfügen, werden die chronologisch sortierten Logs aller Nodes angezeigt, die diese Nachricht durchlaufen hat.

🔍 trace_id = tr-a1b2c3d4-...

[12:34:56.123] [trg]      flow_on_tag_point      태그=MOTOR.TEMP, value=87
[12:34:56.125] [filter]   flow_script_filter     score>80 → TRUE
[12:34:56.126] [transform] flow_change_originator Tag → Asset
[12:34:56.130] [action]   flow_create_work_order WO-20260513-001 생성
[12:34:56.241] [external] flow_send_email        admin@... 발송 성공

Trace-Weitergabe bei Subflow-Aufruf

Wenn Sie mit dem flow_subflow-Node einen anderen Flow aufrufen, wird dieselbe trace_id unverändert übernommen. Das heißt, alle Node-Logs des Hauptflows + des aufgerufenen Subflows lassen sich über denselben Trace durchsuchen.

Weitergabe an externe Systeme

Bei externen HTTP-Aufrufen wird automatisch der Header X-Trace-Id hinzugefügt.

GET /api/orders HTTP/1.1
Host: erp.example.com
X-Trace-Id: tr-a1b2c3d4-e5f6-...
X-Span-Id:  sp-04

Wenn das externe System diesen Header empfängt und ebenfalls im Log erfasst, können die Logs beider Systeme mit derselben ID abgeglichen werden.

Anzeige von trace_id in Alarm·E-Mail

Wenn Sie ${metadata.trace_id} im Alarmtext oder E-Mail-Template einschließen, kann der Operator das Ereignis unmittelbar nach Erhalt des Alarms im Historienbildschirm sofort verfolgen.

제목: [긴급] 모터 과열 — ${metadata.asset_id}
본문:
시각: ${metadata.ts}
값: ${data.value}°C
trace: ${metadata.trace_id}
이력 보기: https://platform.example.com/flow/log?trace_id=${metadata.trace_id}

Aufbewahrungsdauer der Traces

Daten	Aufbewahrungsdauer
trace_id und nodebezogene Span-Logs	7 Tage (identisch mit der Ausführungshistorie)
Externe Speicherung (langfristige Aufbewahrung)	Externes Spiegeln von Systemen unter Audit·Verlaufsverfolgung nutzen

Falls die trace_id lang ist und die Nachrichtengröße belastet, können Sie mit flow_script_transform ein Kurzformat erstellen, das nur die letzten 4 Zeichen (wie tr-...d4) für Alarm·E-Mail verwendet. Für die Suche wird jedoch die vollständige ID benötigt.

Katalog der Graph-Muster

10 häufig verwendete Verdrahtungsmuster im Flow-Graphen.

Muster 1 — Pipeline (einfach seriell)

[trigger] → [filter] → [transform] → [action]

Muster 2 — Fan-out (1 zu N)

                     ┌─→ [action 1]
[trigger] → [t] ─────┼─→ [action 2]
                     └─→ [action 3]

| Merkmal | Eine Nachricht wird von mehreren Aktionen gleichzeitig verarbeitet | | Anwendungsbeispiel | Bei Alarmauslösung → E-Mail + SMS + Slack + Arbeitsauftragserstellung |

Kopien derselben Nachricht werden auf mehrere Nodes verteilt. Jeder Zweig wird unabhängig verarbeitet, ein Fehlschlag in einem Zweig hat keinen Einfluss auf andere.

Muster 3 — Fan-in (N zu 1) — Merge

[trigger A] ──┐
[trigger B] ──┼─→ [flow_merge] → [aggregator] → [action]
[trigger C] ──┘

| Merkmal | Nachrichten mehrerer Trigger werden innerhalb eines Zeitfensters gesammelt und auf einmal verarbeitet | | Anwendungsbeispiel | Alle innerhalb von 5 Minuten aufgetretenen Alarme in einem Tagesbericht zusammenfassen |

Muster 4 — Scatter-Gather (Verteilung → Sammlung)

[trigger] → [split] ─┬─→ [process] ──┐
                     ├─→ [process] ──┼─→ [merge] → [action]
                     └─→ [process] ──┘

| Merkmal | Array wird elementweise aufgeteilt verarbeitet, danach Ergebnisse wieder zusammengeführt | | Anwendungsbeispiel | 100 externe Bestellungen parallel validieren und danach das Ergebnis gebündelt melden |

Muster 5 — Switch (bedingte Verzweigung)

                            ┌─[CRITICAL]→ [긴급 알람]
[trigger] → [flow_switch] ──┼─[WARN]    → [경고 알람]
                            └─[NORMAL]  → [통과]

Muster 6 — Retry with Fallback

[risky] ─[FAILURE]─→ [flow_retry] ─[SUCCESS]─→ (다시 risky)
                                  └[EXHAUSTED]→ [fallback action]

| Merkmal | Vorübergehende Störungen automatisch wiederholen, dauerhafte Störungen alternativ behandeln | | Anwendungsbeispiel | Bei fehlgeschlagener externer API 3 automatische Wiederholungen, bei weiterem Fehlschlag Benachrichtigung an eine Person |

Muster 7 — Circuit Breaker (Nutzung des Circuits)

[trigger] → [throttle] → [external_io] ─[SUCCESS]─→ [save]
                                       └[FAILURE]─→ [log only]

Muster 8 — Dead Letter Queue (DLQ)

[main flow] ─[FAILURE]─→ [flow_save_attributes] → 별도 자산에 적재
                                                  ↑
                              운영자가 주기적 점검 후 수동 재처리

| Merkmal | Dauerhaft fehlgeschlagene Nachrichten in einem separaten Speicher ablegen | | Anwendungsbeispiel | Sammlung fehlgeschlagener Synchronisationsnachrichten mit externem ERP (Ziel für manuelle Nachverarbeitung) |

Muster 9 — Sliding Window Aggregation

[trigger] → [flow_throttle 60s] → [transform: 누적] → [action]
                                  (마지막 60건만 유지)

| Merkmal | Entscheidung anhand des Zustands innerhalb eines Fensters der letzten N Einträge | | Anwendungsbeispiel | Notfallmodus für den Standort bei mehr als 30 Alarmen in den letzten 5 Minuten |

Muster 10 — Saga (mehrstufige Transaktion)

[start] → [step1] ─OK→ [step2] ─OK→ [step3] ─OK→ [complete]
              │            │            │
              └─FAIL→[rollback1]        │
                           │            │
                           ┌─FAIL→[rollback1+2]
                                        │
                                        └─FAIL→[rollback1+2+3]

| Merkmal | Behandlung teilweiser Fehlschläge einer Aufgabe über mehrere externe Systeme hinweg durch Kompensation | | Anwendungsbeispiel | Arbeitsauftragserstellung → Anlagenreservierung → ERP-Synchronisation → Mitarbeiterbenachrichtigung (bei Fehlschlag einer Stufe werden alle vorherigen Stufen storniert) |

Leitfaden zur Musterauswahl

Anforderung	Empfohlenes Muster
Einfacher Schwellenwertalarm	Pipeline (1)
Ein Ereignis → mehrere Kanäle	Fan-out (2)
Mehrere Ereignisse → eine Zusammenfassung	Fan-in / Merge (3)
Massenverarbeitung eines Arrays	Scatter-Gather (4)
Zweigverarbeitung	Switch (5)
Zuverlässigkeit einer externen API	Retry (6)
Schutz eines externen Systems	Circuit (7)
Aufbewahrung fehlgeschlagener Nachrichten	DLQ (8)
Entscheidung anhand der letzten N kumulierten Einträge	Sliding (9)
Konsistenz über mehrere externe Systeme	Saga (10)

Bewährte Praktiken für Flow-Tests

Teststrategie für sicheres Ändern·Bereitstellen von Flows.

3-stufiger Test — Unit → Integration → Simulation

Stufe	Werkzeug	Prüfobjekt
① Unit	Bearbeitungsansicht ▶ Testlauf	Isoliertes Verhalten eines einzelnen Nodes (Skriptausdruck, Antwort auf externen Aufruf)
② Integration	Derselbe Bildschirm, temporäre Payload + Live-Debug ON	Gesamte Sequenz·Verzweigung des Graphen
③ Simulation	Bereitstellen + Warten auf realen Trigger (Staging-Umgebung)	Realer Nachrichtenfluss·Verbindung mit externem System

Payload-Bibliothek für Testläufe

Trigger-bezogene Payload-Beispiele, die in den Testlauf-Dialog eingefügt werden können, finden Sie im Abschnitt Payload-Beispiele nach Trigger. Erstellen Sie auch im Voraus Grenzfälle, die im Produktionsbetrieb häufig auftreten.

Beispiele für Grenzfälle

Fall	Payload
null-Feld	`{"value": null}` — ob das Skript null-sicher ist
Leerer String	`{"value": ""}` — validieren, ob ein leerer String als 0 interpretiert wird
Negative Zahl	`{"value": -1}` — ob die Schwellenwertprüfung dem ± Vorzeichen entsprechend funktioniert
Sehr große Zahl	`{"value": 1e20}` — Überlauf/Genauigkeit
Unicode	`{"name": "한글-Émoji-🚀"}` — Kodierung im externen System

Sicheres Änderungsverfahren

기존 플로우를 [내보내기] (JSON 파일 저장)
새 플로우를 사본으로 만듦 (이름 끝에 `_v2`)
사본의 트리거 패턴을 좁혀 일부 자산만 매칭 (예: TEST-* 사이트만)
신구 동시 배포 — 새 버전 데이터 확인
1주일 안정성 검증 후 신 버전을 전체 트리거 패턴으로 변경
구 버전 해제 + 보관

Aufbewahrung von Regressionstestszenarien

Es wird empfohlen, häufig verwendete Szenarien als Textdatei aufzubewahren und nach Änderungen unbedingt erneut auszuführen.

# regression-tests/alarm-to-workorder.json
{
  "case": "고온 알람 → 워크오더 자동 생성",
  "input": {
    "type": "TAG_ALARM",
    "data": { "alarm_band": "HI_HI", "value": 95.0 }, ...
  },
  "expected": {
    "domain_changes": ["WorkOrder.CREATED"],
    "notifications": ["email:admin@example.com"]
  }
}

Mock-Muster für externe Systeme

Um Antworten externer Systeme in der Staging-Umgebung zu simulieren:

Methode	Beschreibung
url von flow_http_request auf einen Mock-Server	Betriebs-URL als Staging-URL-Variable abtrennen
datasource von flow_jdbc_poll ändern	Von Betriebs-DB → nur auf Staging-DB ändern
to_field von flow_send_email auf fake@example.com	Verhindert versehentlichen Versand

Lasttest nach Zurücksetzen der Zähler

Zähler zurücksetzen (gesamt) der neuen Flow-Version
Normale Triggerauslösung 5 Minuten lang durchführen
In der Listenansicht Verarbeitungs-/Fehlerzähler, durchschnittliche Verarbeitungszeit prüfen
Steigt die p95-Verarbeitungszeit um mehr als 20% gegenüber vorher, Ursachenanalyse durchführen und Rollback erwägen

Performance-Grenzen und Tuning

Verarbeitungsgrenzen und Tuning-Tipps der Flow-Engine.

Standardgrenzen

Element	Standard	Bemerkung
Anzahl der Node-Besuche pro Nachricht (`max_depth`)	100	Erzwungener Abbruch, wenn nach Durchlaufen von 100 Nodes noch nicht beendet
Wiederbesuch desselben Nodes (`max_revisit`)	3	Verhinderung von Endlosschleifen bei Zyklen
Flow-Verarbeitungszeit (`flow_timeout_ms`)	30.000 ms	Erzwungener Abbruch bei Überschreitung von 30 Sekunden pro Nachricht
Timeout für externe IO-Nodes	5.000 ms (`timeout_ms`)	HTTP/Kafka/MQTT usw.
Timeout für Skriptausführung	500 ms (`timeout_ms`)	Nodeweise
Aufbewahrung der Ausführungshistorie	7 Tage	Danach automatischer Ablauf

Empfohlene, häufig verwendete Fenstergrößen

Node	Empfohlenes Fenster	Bemerkung
`flow_throttle`	1.000–60.000 ms	An das API-Limit des externen Systems anpassen
`flow_debounce`	500–5.000 ms	Wenn nur stabile Werte durchgelassen werden sollen
`flow_merge`	5.000–60.000 ms	Zu kurz führt zu Fragmentierung, zu lang zu erhöhter Latenz
Backoff bei `flow_retry`	Start 1.000 ms × 2×	Bei 5 Wiederholungen: 1·2·4·8·16 Sekunden

Möglichkeiten zur Steigerung des Durchsatzes

Trigger-Vorabfilter — mit *_pattern nur benötigte Nachrichten einlassen (am effektivsten)
Vereinfachung des Graphen durch Subflow — Hauptgraph nur für Verzweigung·Routing, schwere Verarbeitung in Subflow
Externe IO asynchron verdrahten — mit flow_delay / flow_throttle die Last externer APIs glätten
Debug-Modus nur in der Validierungsphase — nach stabilem Betrieb Debug-Modus ausschalten
Nur benötigte Kategorien verdrahten — schwere Nodes wie Edge·externe Anbindung nicht zwangsläufig an alle Zweige anschließen

Nachrichtengröße

data / metadata der Nachricht werden JSON-serialisiert in der Ausführungshistorie abgelegt. Bei sehr großen Payloads (z. B. Antwortkörper von mehreren MB) extrahieren Sie möglichst mit einem Transformations-Node nur die benötigten Schlüssel. Sowohl bei der Aufbewahrung der Historie, der Debug-Anzeige als auch beim Subflow-Aufruf ist die Verarbeitungskosten proportional zur Nachrichtengröße.

Audit·Verlaufsverfolgung

Orte, an denen Sie Änderungen·Ausführungshistorie von Flows nachverfolgen können.

Graph-Änderungshistorie

Automatischer Snapshot — wird bei jedem Speichern des Graphen versionsweise abgelegt. Für eine Rückkehr zum vorherigen Zustand wenden Sie sich bitte an den Administrator.
Änderer/Änderungszeitpunkt — die Spalte Letzte Änderung in der Listenansicht zeigt den zuletzt gespeicherten Zeitpunkt. Der Änderer wird gemäß dem authentifizierten Systembenutzer erfasst.
Empfohlene Erfassung des Änderungsgrunds — im Feld Beschreibung der Flow-Metadaten den Änderungsgrund·Verantwortlichen·zugehörige Ticket-Nummer festzuhalten erhöht die Nachverfolgbarkeit.

Verfolgung von Ausführungsereignissen

Nachrichten-ID-Verfolgung — mit dem Filter Nachrichten-ID im Ausführungshistorie-Bildschirm können Sie den gesamten Ablauf (FLOW_START → alle NODE_IN/NODE_OUT → FLOW_END) einer einzelnen Nachricht chronologisch abfragen.
CSV-Export — mit der Schaltfläche CSV herunterladen im Ausführungshistorie-Bildschirm können Sie das aktuelle Abfrageergebnis exportieren und an ein externes Audit-System übermitteln.

Domänenänderungshistorie

Domänenänderungen, die von Aktions-Nodes des Flows (flow_create_* / flow_update_* / flow_delete_*) durchgeführt werden, werden alle an denselben Domänenservice delegiert und daher gemeinsam in der domänenbezogenen Änderungshistorie im Back-Office-Bildschirm erfasst. Die Bearbeiter-ID wird z. B. als insert_user_id="flow" ausgewiesen, um sie von normalen Benutzeränderungen zu unterscheiden.

Langfristige Aufbewahrung durch externe Speicherung

Wenn eine langfristige Aufbewahrung über die Aufbewahrungsdauer der Ausführungshistorie (7 Tage) hinaus benötigt wird, erstellen Sie einen separaten Flow, um zentrale Ereignisse in ein externes System (Kafka·externe DB usw.) zu laden.

[flow_on_tag_alarm]
   │
   ▼
[flow_kafka_publish: topic=audit.alarm.events]

Checkliste für die Bereitstellung neuer Flows

Prüfpunkte vor der Bereitstellung eines neuen Flows in der Produktionsumgebung.

Graphstruktur

Ist genau ein (oder eine beabsichtigte Mehrzahl von) Trigger-Node(s) verdrahtet
Ist der FAILURE-Ausgang aller Aktions-Nodes verarbeitet (mindestens flow_log)
Ist ein vorhandener Zyklus eine beabsichtigte Verdrahtung wie flow_retry, innerhalb des Schutzes von max_revisit
Ist an externen IO-Nodes retry_count/retry_delay_ms oder flow_retry verdrahtet

Node-Konfiguration

Ist *_pattern des Triggers nicht zu eng oder zu weit
Existiert der dynamische Optionspfad *_field tatsächlich in der Nachricht
Werden alle NOT-NULL-Felder des Create-Nodes befüllt (zusätzlich zur automatischen Ergänzung)
Ist die Teilaktualisierung des Update-Nodes tatsächlich beabsichtigt

Validierung·Test

Ein Durchlauf des SUCCESS-Zweigs mit normaler Payload
Ein Durchlauf des FAILURE-Zweigs mit abnormaler Payload (z. B. fehlendes Pflichtfeld)
Durchlauf Retry → EXHAUSTED-Zweig mit Fehlschlagfall des externen IO (falls zutreffend)
Entspricht die INFO/ERROR-Anzeige im Live-Debug-Panel der Absicht
Werden alle Schritte im Bildschirm /flow/log protokolliert

Betriebssicherheit

Ist eine Sicherung (Export der Graph-JSON) extern aufbewahrt
Sind Änderungsgrund·Verantwortlicher in der Flow-Beschreibung erfasst
Sind bei vorhandenen Benachrichtigungsverdrahtungen (Email/SMS/Push) die Empfänger validiert
Gibt es einen Überwachungsplan für die 5–10 Minuten unmittelbar nach der Bereitstellung

Deployment-Strategien — Canary / Blue-Green / A·B-Test

3 Strategien für eine sichere Einführung neuer Flows. Alle Strategien lassen sich allein mit den Grundfunktionen der Plattform (Trigger-Muster·Nachrichten-Dispatch·Live-Debug) umsetzen.

Strategie 1 — Canary (schrittweise Anwendung auf einige Anlagen)

Wenden Sie den neuen Flow zunächst nur auf einige Anlagen an, beobachten Sie eine Weile und weiten Sie ihn danach auf alle aus.

[1단계 출시]
  새 플로우 v2 — 트리거 패턴: asset_id LIKE 'LINE-1.%'  (1개 라인만)
  기존 플로우 v1 — 트리거 패턴: asset_id LIKE 'LINE-2.%' OR 'LINE-3.%' OR ...

[2단계 확대 (1주일 후 안정 확인)]
  v2 패턴: 'LINE-1.%' OR 'LINE-2.%'
  v1 패턴: 'LINE-3.%' OR 'LINE-4.%'

[3단계 전체 (2주 후)]
  v2 패턴: '%'   ← 모든 자산
  v1 해제·보관

Canary-Fortschrittschecklist

Zeitraum	Überwachungspunkt
Tag 1	Fehlerrate < 1% / durchschnittliche Verarbeitungszeit innerhalb ±20% des bisherigen Werts
1 Woche	Externe Systemanbindung 100% normal / Alarmhäufigkeit angemessen
2 Wochen	Kumulierte Statistik / Nutzerfeedback / Entscheidung über Erweiterung auf weitere Linien

Wählen Sie als Canary-Linie eine Linie mit geringem Einfluss unter den laufenden Linien (z. B. Linien mit hoher Wartungshäufigkeit oder nur nächtlichem Betrieb).

Strategie 2 — Blue-Green (gleichzeitiger Betrieb alt·neu, dann sofortiger Wechsel)

[Blue (현재)]                       [Green (새 버전)]
플로우 v1 — 배포됨                  플로우 v2 — 배포 + 격리된 originator
모든 트리거 처리                    metadata.test_mode=true 인 메시지만 처리

[전환 결정 시점]
   v2 트리거 패턴을 v1 과 동일하게 변경 (1초)
   v1 해제 토글 (1초)

Der Kern von Blue-Green ist der sofortige Wechsel, während beide Versionen gleichzeitig bereitgestellt sind — bei Problemen wird v1 sofort wieder bereitgestellt, um zurückzurollen.

Isoliertes Verdrahtungsmuster für v2

[trigger 모든 메시지]
   ↓
[flow_script_filter — metadata.test_mode === true]
   ↓ TRUE
[새 로직]

Testnachrichten werden über die API /flow/{id}/run unter Angabe von metadata.test_mode=true gesendet.

Strategie 3 — A/B-Test (Vergleich von Performance·Ergebnis)

Beide Versionen empfangen dieselbe Nachricht, führen unterschiedliche Aktionen durch und vergleichen dann die Ergebnisse.

[trigger 메시지]
   ↓
[flow_split (메시지 복제)]
   ├ A 경로 → 기존 v1 액션 → [flow_save_attributes target=stats_v1]
   └ B 경로 → 새 v2 액션  → [flow_save_attributes target=stats_v2]

Anschließend im Tagesstatistik-Bildschirm die kumulierten Ergebnisse von stats_v1 vs stats_v2 vergleichen.

Automatische Analyse des A/B-Vergleichs

-- EQL 으로 두 버전 비교 (예: 알람 생성 누적)
context EVERY_1_HOURS
SELECT
  count(CASE WHEN metadata.flow_version='v1' THEN 1 END) AS v1_count,
  count(CASE WHEN metadata.flow_version='v2' THEN 1 END) AS v2_count
FROM AssetAlarm.win:time(1 hour)

Leitfaden zur Strategieauswahl

Situation	Empfohlene Strategie
Erste Einführung eines neuen Automatisierungsszenarios	Canary — nach Validierung auf einer Linie ausweiten
Große Änderung an bestehender Logik (Strukturreform)	Blue-Green — sofortiges Rollback möglich
Messen, welcher von zwei Algorithmen besser ist	A/B-Test
Einfache Anpassung eines Optionswerts	Direkt ändern — nach Notiz zu `metadata.audit_diff` 1 Woche überwachen

Rollback-Verfahren (gemeinsam)

Bei entdecktem Problem sofortiges Rollback:

Neue Version in der Listenansicht mit dem Schalter Aufheben deaktivieren
(bei Canary/A·B) Trigger-Muster auf 0 Treffer ändern
5 Minuten überwachen, ob die bestehende Version allein normal funktioniert
Prüfen, ob im Diagnoselog kein code=FLOW_NOT_DEPLOYED vorliegt
Ursachenanalyse — fehlgeschlagene Nachrichten in der Ausführungshistorie mit trace_id verfolgen

Vorabcheckliste vor Einführung (Kurzfassung)

Das Wesentliche der Checkliste für die Bereitstellung neuer Flows auf einen Blick:

[ ] 테스트 실행으로 정상·경계·실패 시나리오 모두 통과
[ ] 외부 IO 노드에 timeout_ms / 재시도 정책 설정됨
[ ] 자격 증명은 ${creds.*} 참조 (평문 미포함)
[ ] 트리거 패턴이 의도한 자산만 매칭
[ ] 영향받는 도메인 (자산/태그/주문) 식별됨
[ ] 운영 시간(특히 야간) 영향 검토됨
[ ] 롤백 시점·기준·담당자 결정됨
[ ] [감사 로그](#audit) 에 변경 의도 메모 작성됨

Node-Schnellkonfigurationsreferenz

Eine Cheat-Sheet-Sammlung mit den wichtigsten Konfigurationen häufig genutzter Nodes im Betrieb.

Trigger-Schnellkonfiguration

Node	Zentrale Optionen
`flow_schedule`	`cron` (z. B. `0 /5 * * ?` = alle 5 Minuten), oder `interval_ms`
`flow_on_webhook`	Keine Option — Auslösung von außen über `POST /flow/webhook/{flow_id}`
`flow_on_mqtt_subscribe`	`broker_url`, `topic`, `client_id`, `username`/`password`
`flow_jdbc_poll`	`dsn`, `sql`, `interval_ms`
`flow_on_*` (Domäne)	`_pattern` (Glob — z. B. `MOTOR-`, `SITE-?`)

Transformations-Schnellkonfiguration

Node	Zentrale Optionen
`flow_script_transform`	`language` (`JS`/`EQL`), `script`, `timeout_ms` (Standard 500)
`flow_change_originator`	`entity_type`, `id_field`
`flow_rename_keys`	`mapping` (z. B. `{"old":"new"}`)
`flow_template`	`template` (`${data.x}` / `${metadata.y}`-Ersetzung)
`flow_split`	`path` (Array-Position, Standard `data`)
`flow_to_email`	`subject_template`, `body_template`

Ablaufsteuerungs-Schnellkonfiguration

Node	Zentrale Optionen
`flow_delay`	`delay_ms`
`flow_throttle`	`max_msgs`, `window_ms`, (Verzweigung: SUCCESS/THROTTLED)
`flow_debounce`	`window_ms`
`flow_merge`	`window_ms` (in data.merged-Array kumuliert)
`flow_subflow`	`target_flow_id`
`flow_retry`	`max_attempts` (Standard 3), `backoff_ms` (Standard 1000), `backoff_multiplier` (Standard 2.0)
`flow_log`	`level` (INFO/WARN/ERROR), `prefix`
`flow_noop`	(keine Option)

Externe-Anbindungs-Schnellkonfiguration

Node	Statische Option	Dynamische Option (`*_field`)
`flow_http_request`	`method`, `url`, `headers`, `body_template`, `timeout_ms`, `retry_count`, `retry_delay_ms`	`url_field`, `method_field`, `body_field`
`flow_kafka_publish`	`bootstrap_servers`, `topic`, `value_template`, `headers`	`topic_field`, `key_field`
`flow_mqtt_publish`	`broker_url`, `topic`, `qos`	`topic_field`
`flow_webhook_callback`	`url`, `method`, `headers`	`url_field`
`flow_send_email`	`to`, `cc`, `subject`, `body`	`to_field`, `cc_field`, `subject_field`, `body_field`
`flow_send_sms`	`to`, `text`	`to_field`, `text_field`
`flow_send_push`	`title`, `body`	`title_field`, `body_field`
`flow_jdbc_query`	`dsn`, `sql`, `params_field`	—

Schnellkonfiguration für Aktion — Integration/Speicherung

Node	Zentrale Optionen
`flow_save_tag_point`	`tag_id_field` (Standard `metadata.tag_id`), `value_field`, `timestamp_field`
`flow_save_attributes`	`entity_type_field`, `id_field`, `attributes_field`
`flow_dds_publish`	`type`, `originator_field`, `payload_field`
`flow_publish_asset_event`	`asset_id_field`, `event_type_field`, `severity_field`, `details_field`
`flow_publish_asset_command`	`asset_id_field`, `tag_id`, `cmd_key_field`, `payload_field`

Schnellkonfiguration für Domänen-CRUD

Node	Zentrale Optionen
`flow_create_*`	Domänenspezifische Felder + dynamische Option `*_field`. NOT-NULL-Felder werden teilweise automatisch ergänzt (siehe Tabelle Domänen-CRUD für die automatisch ergänzten Positionen)
`flow_update_*`	Teilaktualisierung: nur eingegebene Felder werden aktualisiert, leere Werte werden ignoriert. Für vollständige Überschreibung Kombination Delete + Create
`flow_delete_*`	`*_id_field`
`flow_start/end/pause/resume_work_order`	`order_id_field` (Standard `data.order_id`)
`flow_abort_work_order`	+ `abort_code_field`, `abort_notes_field`
`flow_update_tag_alarm_band_numeric`	`tag_id_field`, `hi_field` usw. (nur eingegebene Felder werden aktualisiert)

Edge-Schnellkonfiguration

Edge-Nodes verwenden alle denselben Optionssatz.

Option	Beschreibung
`url`	Edge-REST-Endpunkt (z. B. `http://edge.local:60000/opc/server`)
`method`	HTTP-Methode (falls nicht gesetzt, Node-spezifischer Standardwert)
`headers`	JSON-Header (Auth-Token usw.)
`body_template`	Request-Body (falls nicht gesetzt, wird `data` unverändert gesendet)
`timeout_ms`	5000

Flow-REST-API — Flows programmgesteuert bedienen

REST-API zur Verwendung, wenn Sie Flows aus einem externen Automatisierungswerkzeug (Ansible/GitOps/CI-Pipeline) codegesteuert verwalten oder wenn ein externes System einen Flow sofort ausführen lassen möchte.

Authentifizierung

Alle API-Aufrufe hängen ein bei Sicherheit → API-Authentifizierungstoken ausgestelltes Token als Header an.

Authorization: Bearer {api_token}

Liste der Endpunkte

1) Flow-Liste abfragen

GET /flow/list

curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/list

Antwort:

{
  "data": [
    { "flow_id": "flow-abc123", "flow_name": "MES 동기화", "deployed": true,
      "node_count": 12, "exec_count": 9430, "error_count": 2, "last_exec_at": 1746247200000 },
    ...
  ]
}

2) Einzelnen Flow abfragen

GET /flow/get/{flow_id}

Der Antwortkörper enthält alle Graph-Nodes·Relationen·Optionen. Kann direkt für Backup/Versionsverwaltung verwendet werden.

3) Flow erstellen

POST /flow/create
Content-Type: application/json

{
  "flow_name":        "신규 자동화",
  "description":     "외부 알림 → 워크오더 자동 생성",
  "deployed":         false
}

Verwenden Sie die flow_id aus der Antwort für nachfolgende Aufrufe.

4) Flow bearbeiten (Meta)

POST /flow/update
Content-Type: application/json

{
  "flow_id":     "flow-abc123",
  "flow_name":   "수정된 이름",
  "description": "..."
}

5) Graph speichern (Massenersetzung von Nodes·Relationen)

POST /flow/{flow_id}/graph
Content-Type: application/json

{
  "nodes":     [ { "node_id": "n1", "type": "flow_on_tag_point", "options": {...}, "x": 100, "y": 100 }, ... ],
  "relations": [ { "from_node_id": "n1", "to_node_id": "n2", "relation": "TRUE" }, ... ]
}

Das Speichern des Graphen ist eine Transaktion — bei Validierungsfehler wird der gesamte Graph zurückgerollt.

6) Bereitstellen / Aufheben

Ändern Sie das deployed-Feld der Flow-Metadaten, um automatisch bereitzustellen·aufzuheben.

# 배포
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     -d '{"flow_id":"flow-abc123","deployed":true}' \
     https://platform.example.com/flow/update

7) Sofortige Ausführung (manueller Trigger)

POST /flow/{flow_id}/run
Content-Type: application/json

{
  "type":       "WEBHOOK",
  "originator": { "entity_type": "External", "id": "manual-run" },
  "data":       { "test": true }
}

Wird ohne Durchlaufen des Vorabfilters des Trigger-Nodes sofort ab dem ersten Node ausgeführt. Nützlich für manuelle Validierung·Debugging.

8) Dispatch (Auslösung unter Umgehung des Triggers)

POST /flow/dispatch
Content-Type: application/json

{
  "type":       "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data":       { "value": 1500 },
  "metadata":   { "tag_id": "MOTOR-001.SPEED", "ts": 1746247200000 }
}

Führt die Nachricht über den normalen Trigger-Verarbeitungspfad ein — alle Flows, die dieser Nachricht entsprechen, werden gleichzeitig ausgelöst.

9) Statistik zurücksetzen

POST /flow/{flow_id}/stats/reset-errors        — 에러 카운트만 초기화
POST /flow/{flow_id}/stats/reset-all           — 전체 카운트 초기화

10) Export / Import — Graph-Sicherung

# Export — JSON 파일로 다운로드
curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/${FLOW_ID}/export \
     -o flow-backup.json

# Import — 같은 JSON 을 다른 환경에 등록
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     --data @flow-backup.json \
     https://platform.example.com/flow/import

Bei Import wird bei einer ID-Kollision automatisch eine neue ID vergeben. Externe Abhängigkeiten (Zugangsdaten·Anlagen-ID usw.) müssen Sie nach dem Import separat abbilden.

11) Alle Flows neu bereitstellen

POST /flow/redeploy

Wird nach umfangreichen Änderungen oder nach dem Neustart von Nodes zur Massensynchronisation verwendet. Rufen Sie diese Funktion im laufenden Betrieb mit Bedacht auf — es kann zu einer vorübergehenden Verarbeitungsverzögerung kommen.

12) Selbstdiagnose

POST /flow/selftest

Prüft die internen Komponenten der Flow-Engine (Trigger-Warteschlange·Dispatcher·Node-Registry·Skript-Laufzeit) einmal und gibt den Status zurück. Das Ergebnis wird auch im Diagnoselog erfasst.

13) Node-Katalog abfragen

GET /flow/catalog

Gibt alle aktuell registrierten Node-Typen·Optionsschemata zurück. Wird von der UI verwendet, um den Inspector dynamisch zu erzeugen. Auch nützlich, wenn ein externes Werkzeug automatisch einen Graphen erzeugt.

Auslösung von außen über Webhook-Trigger

Flows mit einem flow_on_webhook-Trigger können von externen Systemen direkt über folgende URL ausgelöst werden.

POST /flow/webhook/{flow_id}
Authorization: Bearer {token}    또는    X-API-Key: {token}
Content-Type:  application/json

{
  "order_no":  "PO-001",
  "customer":  "ACME",
  "quantity":  1000
}

Antwort:

{ "status": "ACCEPTED", "trace_id": "tr-..." }

Antwortcode	Bedeutung
200	In die Nachrichtenwarteschlange eingereiht (das tatsächliche Verarbeitungsergebnis ist asynchron)
401	Authentifizierungstoken falsch/fehlend
404	Flow-ID nicht vorhanden oder nicht bereitgestellt
413	Body überschreitet 256 KB
422	Trigger-Node ist nicht `flow_on_webhook`
429	Aufruflimit pro Minute überschritten

Beispiele für die Anbindung externer Automatisierungswerkzeuge

GitHub Actions — Flow-Bereitstellung bei PR-Merge

- name: 플로우 배포
  run: |
    curl -X POST \
      -H "Authorization: Bearer ${{ secrets.PP_TOKEN }}" \
      -H "Content-Type: application/json" \
      --data @flows/mes-sync.json \
      https://platform.example.com/flow/import

Ansible — Batch-Verwaltung von Flows

- name: 모든 플로우 백업
  uri:
    url: "https://platform.example.com/flow/{{ item }}/export"
    headers:
      Authorization: "Bearer {{ pp_token }}"
    dest: "/backups/{{ item }}.json"
  loop: "{{ pp_flow_ids }}"

Jenkins — Selftest bei jedem neuen Build

stage('PlantPulse Flow Selftest') {
  steps {
    sh """
      curl -X POST -H 'Authorization: Bearer ${PP_TOKEN}' \\
           https://platform.example.com/flow/selftest \\
           --fail-with-body
    """
  }
}

API-Rate-Limit

Endpunkt	Limit pro Minute (pro Token)
`GET /flow/list` · `get` · `catalog`	600
`POST /flow/create` · `update` · `delete`	60
`POST /flow/{id}/run` · `dispatch` · `webhook/{id}`	1.000
`POST /flow/redeploy` · `selftest`	10

Bei Überschreiten des Limits Antwort 429 Too Many Requests.

OPC/PLC-Industrieintegrationsmuster

Eine Sammlung typischer Integrationsszenarien im industriellen Vor-Ort-Betrieb. Nutzung der Kombination aus 22 Edge-Node-Arten und 4 Anlagenereignis-Publikations-Arten.

Muster A — Automatische Registrierung eines OPC-Servers

Wenn eine neue Linie in Betrieb geht, wird die Linieninformation vom ERP empfangen und der OPC-Server automatisch im Edge registriert.

[flow_on_webhook]
   ↓ (data = {"line_id":"LINE-7","host":"10.0.7.10","port":4840})
[flow_change_originator]
   ↓ (originator → Edge:EDGE-A)
[flow_edge_opc_create]
   ↓ (path=/api/v1/opc, body={"opc_id":"OPC-${data.line_id}","host":"${data.host}","port":${data.port}})
[flow_edge_opc_start]
   ↓ (수집 시작 명령)
[flow_save_attributes]
   ↓ (등록 이력을 자산 메타에 기록)

Der Edge-Node führt automatisch ein Lookup von api_key in der Mastertabelle mm_edge durch. Der Operator muss keine separate Authentifizierungskonfiguration vornehmen.

Muster B — Automatische Ausgabe eines PLC-Befehls

Bei Auftreten eines Anlagenalarms wird automatisch ein Stoppbefehl an die PLC ausgegeben.

[flow_on_asset_alarm]
   ↓ where priority='ERROR' AND alarm_band='TRIP_HI'
[flow_publish_asset_command]
   ↓ (event_type='STOP', payload={"reason":"trip_high","triggered_by":"flow"})
[flow_log] (감사 로그)

flow_publish_asset_command wird sofort über das Befehls-Topic der Anlage (asset/{id}/cmd/STOP) zugestellt, sodass das Edge einen OPC-Write an die PLC durchführt.

Muster C — Massenregistrierung von OPC-Tags (Import aus CSV/Excel)

Bei einem neuen Anlagen-Setup werden 100–1000 Tags aus einer CSV auf einmal registriert.

[flow_jdbc_poll]            ← 외부 DB 에 적재된 CSV 행
   ↓ (한 행 = 한 메시지)
[flow_script_transform]     ← 태그 정의 가공
   ↓
[flow_create_tag]           ← 도메인 등록 (NOT NULL 자동 보강)
   ↓
[flow_edge_tag_create]      ← 엣지에 동기화
   ↓
[flow_log] (성공 카운트)

Bei der Registrierung von 1.000 Tags wird empfohlen, mit flow_throttle auf höchstens 100 pro Minute zu glätten. Das Edge kann vorübergehend langsamer reagieren.

Muster D — Automatische Wiederherstellung bei OPC-Verbindungsabbruch

Wenn die OPC-Verbindung abbricht und sich erholt, wird automatisch neu gestartet.

[flow_on_opc_status]
   ↓ where prev_status='CONNECTED' AND status='DISCONNECTED'
[flow_delay 30s]            ← 잠시 안정화 대기
   ↓
[flow_edge_opc_start]       ← 수집 재시작 시도
   ↓ SUCCESS: 정상
   └ FAILURE: [flow_retry max=3] → EXHAUSTED: [관리자 알림]

Muster E — Statusprüfung der Linie bei Schichtbeginn

Prüft zum Beginn jeder Schicht alle Linien des Standorts.

[flow_schedule cron='0 0 7,15,23 * * ?']   ← 7시·15시·23시
   ↓
[flow_edge_monitoring]      ← 엣지에서 OPC 상태 전체 조회
   ↓ data.opcs = [...]
[flow_split]                ← 배열을 행별 메시지로
   ↓
[flow_script_filter]        ← status != 'CONNECTED' 만
   ↓
[flow_send_email]           ← 비정상 OPC 만 한 통의 메일에 합쳐 발송

Muster F — Automatische Zuweisung von Mitarbeitern bei Schichtwechsel

Weist gemäß Änderungen im Dienstplan automatisch Mitarbeiter zu Arbeitsaufträgen zu.

[flow_on_entity_event type='ENTITY_UPDATED' Calendar]
   ↓
[flow_script_filter (시프트 시작 시각이 지금인 경우만)]
   ↓
[flow_jdbc_query (그 시프트의 작업자 목록 조회)]
   ↓ data.employees=[...]
[flow_split]
   ↓
[flow_update_work_order (작업자 매핑)]

Muster G — Gleichzeitiger Aufruf mehrerer PLCs (Scatter-Gather)

Ruft Daten mehrerer PLCs gleichzeitig ab und führt die Ergebnisse zusammen.

[flow_on_webhook]
   ↓ data.targets=['PLC-1','PLC-2','PLC-3']
[flow_split]
   ↓ (3개로 분할)
[flow_edge_tag_read]
   ↓ (병렬 호출)
[flow_merge window=5s]
   ↓ data.merged=[...]
[flow_script_transform (data.summary 만들기)]
   ↓
[flow_publish_asset_aggregation]

Muster H — Automatische Anpassung des Alarmbands

Wendet die trainierten LIMIT_MIN/MAX-Werte der Vorhersageanalyse automatisch auf das Alarmband an.

[flow_schedule cron='0 0 4 * * MON']    ← 매주 월요일 4시
   ↓
[flow_jdbc_query (forecast 결과 조회)]
   ↓ data.tags=[{tag_id, limit_min, limit_max}]
[flow_split]
   ↓
[flow_update_tag_alarm_band_numeric]    ← lo=limit_min, hi=limit_max
   ↓ SUCCESS
[flow_log] (적용 결과 기록)

Muster I — Automatische Klassifizierung von Ausfallzeiten

Klassifiziert Anlagenstoppereignisse nach Ursache, um sie präzise in der OEE-Verfügbarkeit widerzuspiegeln.

[flow_on_asset_event event_type='SHUTDOWN']
   ↓
[flow_switch]
   ├ CASE 점심: hour_of_day(ts) BETWEEN 12 AND 13 → [flow_create_calendar (계획정지)]
   ├ CASE 시프트 종료: 시프트 종료 시각 ±5분    → [flow_log (정상 종료)]
   ├ CASE 정기점검: 마지막 정비일 +30일 경과     → [flow_create_calendar (정기점검)]
   └ DEFAULT (비계획)                            → [flow_send_email (긴급)]

Muster J — Bidirektionale Synchronisation von Arbeitsaufträgen mit externem ERP

Bidirektionale Spiegelung PlantPulse ↔ externes ERP.

입력 1: ERP → 플랜트펄스
[flow_on_webhook] → [flow_create_work_order]

입력 2: 플랜트펄스 → ERP
[flow_on_entity_event type='ENTITY_CREATED' WorkOrder]
  → [flow_script_filter (출처가 ERP 가 아닌 경우만)]
  → [flow_http_request (ERP REST PUT)]

Um Endlosschleifen zu vermeiden, kennzeichnen Sie von außen eingehende Nachrichten mit metadata.source='ERP' und filtern Sie diese Nachricht im Flow der Gegenrichtung heraus.

Hinweise bei der Industrieintegration

Punkt	Empfehlung
OPC-Write-Befehl nach Sicherheitsinterlock ausgeben	Erst nach Bestätigung durch den Bediener oder Durchlaufen automatischer Sicherheitsregeln
Nicht zu kurzes PLC-Pollingintervall setzen	Unter 1 Sekunde besteht das Risiko eines starken Anstiegs der PLC-CPU-Auslastung
Ressourcenlimit des Docker-Containers am Edge-Gerät	Vorsicht bei OPC-Erfassung + zusätzlichem Container bei über 80% CPU/Speicher
In der Inbetriebnahmephase alle Befehle im Dry-Run-Modus	Option `dry_run=true` von `flow_edge_tag_write` nutzen
Vorsorge für Stromausfall — nur permanenten Speichertriggern vertrauen	`flow_on_webhook` usw. sind flüchtig, Domänenereignistrigger sind permanent

Cookbook für externe Systemintegration

Sammlung von flow_http_request-Konfigurationen und Payload-Beispielen für häufig aufgerufene externe Systeme.

Slack — Incoming-Webhook-Benachrichtigung

URL:    https://hooks.slack.com/services/T0000/B0000/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body (body_template):
  {
    "text": "${data.title}",
    "blocks": [
      { "type": "header",  "text": { "type": "plain_text", "text": "${data.title}" } },
      { "type": "section", "text": { "type": "mrkdwn",     "text": "*자산:* ${metadata.asset_id}\n*값:* ${data.value}\n*시각:* ${metadata.ts}" } }
    ]
  }

Erfolg: 200 + Antwortkörper ok
Fehlschlag: 400 (fehlerhafte payload) / 404 (fehlerhaftes secret) / 429 (Aufruflimit pro Minute überschritten)
Empfehlung: höchstens 1× pro Minute mit flow_throttle verdrahten

Microsoft Teams — Webhook-Benachrichtigung

URL:    https://{org}.webhook.office.com/webhookb2/{id}/IncomingWebhook/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body:
  {
    "@type":    "MessageCard",
    "@context": "https://schema.org/extensions",
    "themeColor": "FF0000",
    "title": "${data.title}",
    "sections": [{
      "facts": [
        { "name": "자산", "value": "${metadata.asset_id}" },
        { "name": "값",   "value": "${data.value}"       },
        { "name": "시각", "value": "${metadata.ts}"      }
      ]
    }]
  }

Verzweigen Sie themeColor nach FF0000 (rot)/FFA500 (orange)/00C853 (grün), um die Priorität visuell darzustellen.

Jira — Automatische Ticketerstellung

URL:    https://{org}.atlassian.net/rest/api/3/issue
Method: POST
Headers:
  Authorization: Basic {base64(email:apitoken)}
  Content-Type:  application/json
Body:
  {
    "fields": {
      "project":    { "key": "OPS" },
      "summary":    "[자동] ${data.title}",
      "description": {
        "type": "doc",
        "version": 1,
        "content": [{
          "type": "paragraph",
          "content": [{ "type": "text", "text": "자산: ${metadata.asset_id}\n값: ${data.value}" }]
        }]
      },
      "issuetype":  { "name": "Bug" },
      "priority":   { "name": "High" }
    }
  }

Im Antwortkörper data.response_body.key steht ein Ticket-Schlüssel wie OPS-1234, nutzen Sie diesen im nachfolgenden Node
Fehlschlag: 400 (fehlerhaftes Feld) / 401 (Authentifizierung) / 403 (keine Projektberechtigung)

GitHub Issues — Automatische Ticketerstellung

URL:    https://api.github.com/repos/{owner}/{repo}/issues
Method: POST
Headers:
  Authorization: Bearer {pat_token}
  Accept:        application/vnd.github+json
Body:
  {
    "title":  "[자동] ${data.title}",
    "body":   "자산: ${metadata.asset_id}\n값: ${data.value}\n시각: ${metadata.ts}\ntrace: ${metadata.trace_id}",
    "labels": ["automation", "ops"]
  }

ERP (SAP S/4HANA OData) — Arbeitsauftrag ausgeben

URL:    https://{host}/sap/opu/odata/sap/API_MAINTNOTIFICATION/MaintenanceNotification
Method: POST
Headers:
  Authorization:   Basic {base64(user:pass)}
  X-CSRF-Token:    fetch          ← 별도 GET 으로 토큰 받기
  Content-Type:    application/json
Body:
  {
    "NotificationType":         "M2",
    "MaintenanceNotificationType": "M2",
    "TechnicalObject":          "${metadata.asset_id}",
    "NotificationText":         "${data.title}",
    "MalfunctionStartDate":     "${data.start_date}",
    "Priority":                 "${data.priority}"
  }

Sie müssen zuerst per GET das CSRF-Token abrufen und dann mit demselben Session-Cookie POST senden. Verbinden Sie zwei flow_http_request-Nodes und übertragen Sie das Token aus dem Antwortheader des ersten Nodes in den Header des zweiten Nodes.

MES (direktes externes DB-INSERT) — `flow_jdbc_query`

Direkte Zeilenerstellung in der Arbeitsauftragstabelle eines externen MES-Systems.

INSERT INTO mes.work_order
  (order_no, asset_id, product_id, qty, status, due_date, created_by)
VALUES
  (:order_no, :asset_id, :product_id, :qty, 'NEW', :due_date, 'flow')

Node-Option	Wert
`datasource_id`	mes_db (vorab in den Systemeinstellungen registriert)
`query`	obiges SQL
`binds`	`{"order_no":"${data.order_no}","asset_id":"${metadata.asset_id}",...}`

flow_jdbc_query unterstützt neben SELECT auch INSERT/UPDATE/DELETE. Transaktionen werden nodeweise automatisch commit/rollback.

Grafana — Automatische Dashboard-Aktualisierungsbenachrichtigung

URL:    https://{host}/api/annotations
Method: POST
Headers:
  Authorization: Bearer {api_key}
  Content-Type:  application/json
Body:
  {
    "dashboardUID": "...",
    "panelId":      4,
    "time":         ${metadata.ts},
    "tags":         ["alarm", "${metadata.asset_id}"],
    "text":         "${data.title}"
  }

Bei Alarmauslösung wird automatisch ein Marker auf dem Graphen angezeigt — sehr nützlich für nachträgliche Analysen.

Telegram — Bot-Nachricht

URL:    https://api.telegram.org/bot{token}/sendMessage
Method: POST
Body:
  {
    "chat_id": "-1001234567890",
    "text":    "🚨 *${data.title}*\n자산: `${metadata.asset_id}`\n값: ${data.value}",
    "parse_mode": "Markdown"
  }

Schnellübersicht der REST-Authentifizierungsmuster

Authentifizierungsmethode	Header
API Key (Header)	`X-API-Key: {token}`
API Key (Query)	`?api_key={token}` am Ende der URL
Bearer Token	`Authorization: Bearer {token}`
Basic Auth	`Authorization: Basic {base64(user:pass)}`
OAuth2 (Client Credentials)	Token-Ausstellung → Bearer-Header (Erneuerung mit separatem Node)
HMAC-Signatur	Body-/zeitbasierte Signatur per Skript berechnen und in Header `X-Signature` setzen

Muster zur Antwortauswertung

Nach der flow_http_request-Antwort data.response_body im nächsten Node nutzen:

// 응답에서 ID 추출
data.created_id = data.response_body.id;
data.status     = data.response_body.status;

// 응답 배열에서 첫 행만
data.first_item = (data.response_body.items || [])[0] || null;

// 응답이 문자열이면 JSON 파싱
if (typeof data.response_body === 'string') {
  try { data.response_body = JSON.parse(data.response_body); } catch (e) {}
}
msg

Sicherheitsempfehlungen bei externem Systemaufruf

Punkt	Empfehlung
Token nicht im Klartext im Graphen belassen	Nach Registrierung im Zugangsdatenspeicher unter Systemeinstellungen → als `${creds.slack_webhook}` referenzieren
Bei sensiblen Daten im Antwortkörper maskieren	Mit `flow_script_transform` PII/Token entfernen, bevor an den nächsten Node weitergegeben wird
Separate Retry-Richtlinie pro externem System	Bei häufig aufgerufenen Systemen separat `flow_retry` + `flow_throttle` verdrahten
Aufrufergebnis als Audit-Log speichern	Mit `flow_save_attributes` Aufrufhistorie-Metadaten der Anlage hinzufügen

Cookbook für Datentransformation

Eine Sammlung von Transformationsmustern, die Sie direkt in den flow_script_transform-Node übertragen können. Alle Beispiele basieren auf JavaScript und verwenden die Kurzaliase data / metadata.

Einheitenumrechnung

// 섭씨 → 화씨
data.temp_f = data.temp_c * 9 / 5 + 32;

// 바 → kPa
data.pressure_kpa = data.pressure_bar * 100;

// rpm → rad/s
data.angular_velocity = data.rpm * 2 * Math.PI / 60;

// kWh → MJ
data.energy_mj = data.energy_kwh * 3.6;

// 바이트 → MB (소수 1자리)
data.size_mb = Math.round(data.bytes / 1024 / 1024 * 10) / 10;

msg

Datums-/Zeittransformation

const d = new Date(metadata.ts);

// ISO 8601 — 2026-05-12T12:34:56.789Z
data.iso = d.toISOString();

// 사람용 — 2026-05-12 21:34:56 (KST)
data.local = d.toLocaleString('ko-KR', { hour12: false });

// 날짜만 — 2026-05-12
data.date = d.toISOString().slice(0, 10);

// 시간만 — 21:34:56
data.time = d.toTimeString().slice(0, 8);

// 분 단위로 내림 (스파크라인 키)
data.minute_key = Math.floor(metadata.ts / 60000) * 60000;

// 시간 단위로 내림
data.hour_key = Math.floor(metadata.ts / 3600000) * 3600000;

// 한국 시간 (UTC+9) 직접 더하기 (서버 시간이 UTC 인 경우)
data.kst_hour = new Date(metadata.ts + 9 * 3600000).getUTCHours();

msg

Stringnormalisierung

// 공백 trim + 소문자화
data.normalized = (data.text || '').trim().toLowerCase();

// 한글·영문 외 제거 (특수문자/공백 정리)
data.clean = (data.text || '').replace(/[^가-힣a-zA-Z0-9]/g, '');

// camelCase → snake_case
data.snake = (data.text || '').replace(/([A-Z])/g, '_$1').toLowerCase().replace(/^_/, '');

// 전화번호 정규화 (숫자만)
data.phone_digits = (data.phone || '').replace(/\D/g, '');

// 한국 휴대전화 자동 포맷 (010-1234-5678)
const p = (data.phone || '').replace(/\D/g, '');
data.phone_formatted = p.length === 11 ? p.replace(/(\d{3})(\d{4})(\d{4})/, '$1-$2-$3') : p;

msg

Abflachen verschachtelter Objekte

// data.location.address.city → data.city
function flatten(obj, prefix, out) {
  out = out || {};
  for (var k in obj) {
    var v = obj[k];
    var key = prefix ? prefix + '_' + k : k;
    if (v && typeof v === 'object' && !Array.isArray(v)) flatten(v, key, out);
    else out[key] = v;
  }
  return out;
}
data = flatten(data);
msg

Flaches Objekt → Verschachtelt

// data.user_name + data.user_email → data.user = {...}
function unflatten(obj) {
  var out = {};
  for (var k in obj) {
    var parts = k.split('_');
    var cur = out;
    for (var i = 0; i < parts.length - 1; i++) {
      cur[parts[i]] = cur[parts[i]] || {};
      cur = cur[parts[i]];
    }
    cur[parts[parts.length - 1]] = obj[k];
  }
  return out;
}
data = unflatten(data);
msg

CSV-Zeilenerstellung

// 외부 시스템에 CSV 한 줄 보낼 때
function csvEscape(v) {
  if (v === null || v === undefined) return '';
  v = String(v);
  return /[,"\n]/.test(v) ? '"' + v.replace(/"/g, '""') + '"' : v;
}
data.csv_row = [
  csvEscape(metadata.ts),
  csvEscape(metadata.asset_id),
  csvEscape(data.value),
  csvEscape(data.quality)
].join(',');
msg

Array-Aggregation

var arr = data.values || [];

// 합·평균·최소·최대
data.sum = arr.reduce(function(a, b) { return a + b; }, 0);
data.avg = arr.length ? data.sum / arr.length : 0;
data.min = arr.length ? Math.min.apply(null, arr) : null;
data.max = arr.length ? Math.max.apply(null, arr) : null;

// 중앙값
var sorted = arr.slice().sort(function(a, b) { return a - b; });
var mid    = Math.floor(sorted.length / 2);
data.median = sorted.length === 0 ? null
            : sorted.length % 2 ? sorted[mid]
            : (sorted[mid - 1] + sorted[mid]) / 2;

msg

Bedingtes Hinzufügen von Feldern (Schema-Evolution)

// 알람 우선순위에 따라 색상·아이콘 자동 부여
var p = data.priority || 'INFO';
data.color   = { ERROR: '#d32f2f', WARN: '#f57c00', INFO: '#1976d2' }[p] || '#9e9e9e';
data.icon    = { ERROR: '🔴',       WARN: '🟠',       INFO: '🔵' }[p] || '⚪';
data.urgency = p === 'ERROR' ? 3 : p === 'WARN' ? 2 : 1;
msg

Teilweise Maskierung der Payload

function maskEmail(e) {
  if (!e || e.indexOf('@') < 0) return e;
  var parts = e.split('@');
  return parts[0].slice(0, 2) + '***@' + parts[1];
}
function maskPhone(p) {
  return (p || '').replace(/(\d{3})\d{4}(\d{4})/, '$1-****-$2');
}
data.user_email = maskEmail(data.user_email);
data.user_phone = maskPhone(data.user_phone);
msg

Mehrsprachige (i18n) Vorlage

// 메시지 본문을 사용자 로케일에 따라 분기
const tpl = {
  'ko-KR': '🚨 ${asset} 온도 ${value}°C 임계 초과',
  'en-US': '🚨 ${asset} temperature ${value}°C exceeds threshold',
  'ja-JP': '🚨 ${asset} 温度 ${value}°C 閾値超過'
};
const locale = metadata.user_locale || 'ko-KR';
const template = tpl[locale] || tpl['ko-KR'];
data.notification = template
  .replace('${asset}', metadata.asset_id)
  .replace('${value}', data.value);
msg

Sichere JSON-Path-Abfrage

// data.deep.nested.field 처럼 깊은 경로를 안전하게 조회
function get(obj, path, dflt) {
  var keys = path.split('.');
  var cur  = obj;
  for (var i = 0; i < keys.length; i++) {
    if (cur === null || cur === undefined) return dflt;
    cur = cur[keys[i]];
  }
  return cur === undefined ? dflt : cur;
}
data.city = get(data, 'location.address.city', 'Unknown');
msg

Nachrichtenzusammenführung (Nachbearbeitung von `flow_merge`)

// flow_merge 가 data.merged 배열로 모은 메시지들을 1건으로 합침
var items = data.merged || [];
data.summary = {
  count:     items.length,
  first_ts:  items[0]?.metadata?.ts,
  last_ts:   items[items.length - 1]?.metadata?.ts,
  assets:    Array.from(new Set(items.map(function(m) { return m.metadata?.asset_id; }))),
  max_value: Math.max.apply(null, items.map(function(m) { return m.data?.value || 0 }))
};
delete data.merged;
msg

Sammlung wiederverwendbarer Skripte

Eine Skriptbibliothek, die Sie direkt in die Nodes flow_script_filter / flow_script_transform / flow_switch übertragen können. Alle Skripte basieren auf JavaScript und verwenden die Kurzaliase data / metadata.

Transformationsskripte

Einheitenumrechnung·Labeling

// 섭씨 → 화씨 + 등급 라벨
data.temp_f = data.temp * 1.8 + 32;
data.grade  = data.temp > 80 ? 'HIGH' : data.temp < 0 ? 'LOW' : 'OK';
msg

Ergänzung von Metadaten zu Zeit·Schicht·Wochentag

const d = new Date(metadata.ts);
const h = d.getHours();
metadata.shift   = (h >= 6 && h < 14) ? 'A' : (h < 22) ? 'B' : 'C';
metadata.weekday = ['SUN','MON','TUE','WED','THU','FRI','SAT'][d.getDay()];
metadata.is_weekend = (d.getDay() === 0 || d.getDay() === 6);
msg

Berechnung des Alarmbands als Mittelwert ± 3σ

const m = data.mean, s = data.stddev || 1;
data.tag_id = originator.id + '.TEMP';
data.hi     = m + 3*s;  data.lo    = m - 3*s;
data.hi_hi  = m + 4*s;  data.lo_lo = m - 4*s;
data.use_alarm = true;
msg

Zuordnung MES-PO → Arbeitsauftrag

const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty,
  product_id: po.product_code
};
msg

Normalisierung externer Payloads (Vereinheitlichung verschiedener Feldnamen)

// 외부 시스템에 따라 키 이름이 다른 경우 — 한 줄로 정규화
data.value     = data.value ?? data.val ?? data.v;
data.timestamp = data.timestamp ?? data.ts ?? metadata.ts;
data.tag_id    = data.tag_id ?? data.tagId ?? data.id;
msg

Kumulierte Anzahl von Schwellenwertverletzungen (Stateful-Muster)

// 자산 attribute 와 함께 사용 — 상태는 메시지 자체에 적재
data.consecutive_fail = (data.consecutive_fail ?? 0) + (data.passed ? 0 : 1);
data.alert            = data.consecutive_fail >= 5;
msg

Zusammenfassung der Nachrichtenstatistik (Aufbereitung des Merge-Ergebnisses)

// flow_merge 후 data.merged 배열을 받아 요약
const arr = data.merged || [];
data.count   = arr.length;
data.values  = arr.map(m => m.data.value).filter(v => v != null);
data.mean    = data.values.reduce((a,b)=>a+b, 0) / (data.values.length || 1);
data.max     = Math.max(...data.values);
data.min     = Math.min(...data.values);
delete data.merged;
msg

Anwendung von Schwellenwerten nach Tageszeit

const h = new Date(metadata.ts).getHours();
const threshold = (h >= 8 && h < 18) ? 90 : 70;   // 주간 90, 야간 70
data.alarm = data.value > threshold;
data.threshold = threshold;
msg

Filterskripte

Prioritäts-Whitelist

['ERROR', 'CRITICAL'].includes(data.priority)

Zeitfenster (nur Werktage)

const h = new Date(metadata.ts).getHours();
h >= 8 && h < 20

Anlagen-·Tag-ID-Muster

/^MOTOR-.*$/.test(originator.id)
// 또는
metadata.tag_id && metadata.tag_id.startsWith('LINE-A.')

Schwellenwert + Stabilität (N-fach in Folge)

data.value > 100 && (data.consecutive_count ?? 0) >= 5

Prüfung von Geschäftszeiten·Feiertagen

const d = new Date(metadata.ts);
const h = d.getHours();
const wd = d.getDay();
// 평일 09-18시만 통과
wd >= 1 && wd <= 5 && h >= 9 && h < 18

Nur bestimmter Standort

['SITE-01', 'SITE-02'].includes(metadata.site_id)

Prüfung auf Vorhandensein aller Pflichtfelder

data.value != null && data.tag_id && metadata.ts

Blockade selbst publizierter Nachrichten (bidirektionale Synchronisation)

metadata.source !== 'flow'

Switch-Case-Skripte

Nach Prioritätsstufe

// case "Critical"
['CRITICAL', 'EMERGENCY'].includes(data.priority)

// case "High"
data.priority === 'ERROR'

// case "Normal"
['WARN', 'INFO'].includes(data.priority)

Nach Anlagenstatus

// case "Running"
data.status === 'RUN'

// case "Stopped"
['STOP', 'IDLE', 'PAUSED'].includes(data.status)

// case "Faulted"
data.status === 'FAULT' || data.error_count > 0

Nach Lebenszyklus des Arbeitsauftrags

// case "Start"
data.event_type === 'START_REQUEST'

// case "End"
data.event_type === 'COMPLETE' && data.qty_done >= data.qty_planned

// case "Abort"
data.event_type === 'CANCEL'

Die obigen Skripte sind eine Zusammenstellung von Mustern, die im Produktionsbetrieb häufig verwendet werden. Sie funktionieren, wenn Sie sie direkt im Graphen verdrahten, und Sie müssen nur Schwellenwerte·Feldnamen an Ihre Domäne anpassen.

Glossar

Flow-Engine-Begriffe

Begriff	Bedeutung
entity_type	Art der Subjekt-Entität der Nachricht — `Asset`, `Tag`, `Site`, `Order`, `Customer`, `Product`, `Employee`, `Calendar` usw.
originator	Die von der Nachricht referenzierte Subjekt-Entität (`entity_type` + `id`). Beispiel: `Asset/MOTOR-001`
type	Klassifizierungslabel der Nachricht. Identifiziert, welche Art von Ereignis der Trigger empfangen hat
data	Der Inhalt (Payload) der Nachricht — wird hauptsächlich von Transformations-·Aktions-Nodes gelesen und geschrieben
metadata	Kontext der Nachricht (Zeit·Standort·Schicht·Tag-ID usw.) — bleibt auch bei Transformation bis zum Ende des Ablaufs erhalten
relation	Label des Ausgangs-Wires eines Nodes. `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` (alle in Großbuchstaben)
*`_field` dynamische Option**	Eingabemethode, bei der statt eines statischen Werts der Wert aus einem Pfad der Nachrichten-Payload (z. B. `data.tag_id`) gelesen wird
Glob-Muster	Wildcard-Darstellung der Option `_pattern` des Triggers — `` steht für 0 oder mehr Zeichen, `?` für genau 1 Zeichen
Snapshot	Automatisch beim Speichern des Graphen abgelegtes versionsweises Backup. Wird bei Vorfällen zur Wiederherstellung des vorherigen Zustands verwendet
Teilaktualisierung (fetch+merge)	Methode des Update-Nodes, bei der nach Abfrage des bestehenden Datensatzes nur eingegebene Felder zusammengeführt werden. Leere Werte werden ignoriert
SKIPPED	Verarbeitung, bei der Nachrichten, die dem Triggermuster nicht entsprechen, nicht an nachfolgende Nodes fließen und auch vom Zähler ausgeschlossen werden
EXHAUSTED	Zweig, der ausgelöst wird, wenn der `flow_retry`-Node die maximale Anzahl an Wiederholungsversuchen überschritten hat
THROTTLED	Zweig, der ausgelöst wird, wenn der `flow_throttle`-Node eine Nachricht blockiert, die das Limit innerhalb des Fensters überschritten hat

Domänen-·Abkürzungslexikon

Eine Übersicht häufig verwendeter Abkürzungen aus der Betriebsumgebung von Anlagen, damit Sie sie in diesem Handbuch schnell nachschlagen können.

Abkürzung	Bedeutung	In diesem Handbuch
MES	Manufacturing Execution System — Verwaltung von Arbeitsaufträgen·Produktionsergebnissen	Ziel externer Anbindung (HTTP/MQTT/externe DB)
ERP	Enterprise Resource Planning — unternehmensweites Ressourcen-·Planungssystem	Ziel externer Anbindung
SCADA	Supervisory Control and Data Acquisition — Überwachung·Steuerung	Ziel externer Anbindung / Ausgang von `flow_publish_asset_command`
OPC	Open Platform Communications — Industriekommunikationsstandard	Ziel der Kategorie Edge (`flow_edge_opc_*`)
OEE	Overall Equipment Effectiveness — Verfügbarkeit × Leistung × Qualität	Trigger `flow_on_oee_event`
RAM	Reliability·Availability·Maintainability	Trigger `flow_on_ram_event`
EMS	Energy Management System	Trigger `flow_on_ems_event`
EQL	Domänenereignisregel-Ausdruck	Option `language: "EQL"` von `flow_script_filter`/`flow_script_transform`
CEP	Komplexe Ereignisverarbeitung (Complex Event Processing)	EQL-basierte Regelengine. Alarme müssen ausschließlich über den CEP-Pfad entstehen, um Konsistenz zu wahren
PO	Purchase Order — Einkaufs-·Produktionsauftrag	Einheit, die bei MES-Anbindung in einen Arbeitsauftrag umgewandelt wird
WO	Work Order (Arbeitsauftrag)	Node-Gruppe `flow_*_work_order`
CMMS	Computerized Maintenance Management System	Ziel externer Anbindung
MTTF/MTTR	Mean Time To Failure / To Repair	Zentrale Kennzahl von RAM-Ereignissen
HMI	Human–Machine Interface	Bedienbildschirm von SCADA usw.

Versionshinweise

Zeitpunkte der Einführung der wichtigsten Funktionsgruppen, die dieses Handbuch zum aktuellen Zeitpunkt behandelt. Falls Sie eine ältere Version betreiben, können sich einige Funktionen abweichend verhalten.

V2026.05 — Erweiterung der Domänenautomatisierung

11 Arten der Edge-Kategorie neu eingeführt — Automatisierung von OPC-Server-/Tag-CRUD + Monitoring
5 Arten von Statusübergängen des Arbeitsauftrags — flow_start/end/pause/resume/abort_work_order
2 Arten der Teilaktualisierung des Tag-Alarmbands — Aktualisierung nur der eingegebenen Felder für numerisches/boolesches Alarmband
Neue Ablaufsteuerung flow_retry — Backoff + maximale Versuche + EXHAUSTED-Zweig
5 neue Trigger-Arten — flow_on_asset_health_status / _connection_status / _oee_event / _ram_event / _ems_event
Teilaktualisierung des Update-Nodes (fetch+merge) — Zusammenführung nur der eingegebenen Felder nach Abfrage des bestehenden Datensatzes
Standardisierung der Relations auf Großbuchstaben — SUCCESS/FAILURE/... alle in Großbuchstaben, bestehende Graphen werden automatisch konvertiert
SKIPPED-Behandlung — Ausschluss von Nachrichten, die dem Triggermuster nicht entsprechen, aus dem Zähler
Betriebswerkzeug „Alle neu bereitstellen" (redeploy) — Massen-Neuladung aktiver Flows
Alle Zähler zurücksetzen — Massen-Reset der Statistiken auf Node- + Flow-Ebene

V2026.03 — Release-Stabilität

Automatischer Snapshot wird beim Speichern des Graphen abgelegt
Live-Debug-Panel (2-Sekunden-Aktualisierung unten im Inspector) + Node-LED·Anzeige der Dauer
Import/Export (Graph-JSON-Datei)
Einführung eines Lastwächters — Ausgabe eines Diagnoselogs bei anhaltender Last

Davor

M1 — Grundgerüst von Engine·UI, Graph-CRUD, visuelle Canvas
M2 — Domänen-Aktions-Nodes (Anlage·Tag·Arbeitsauftrag·Produktionsdomäne usw.)
M3 — Diversifizierung der Trigger, 8 Arten Filter/Transformation/externe Anbindung, Skript-Engine
M4 — Debug-Speicherung·Ausführungshistorie-Bildschirm, Statistik, Bereitstellen/Aufheben, Import/Export
M5 — Integration von Domänentriggern, Massenvalidierung von Szenarien

Alle in diesem Handbuch behandelten Funktionen und Verhaltensweisen basieren auf dem oben genannten Zeitpunkt V2026.05.

Inhaltsverzeichnis​

Überblick​

Lernpfad — Empfohlene Einstiegsreihenfolge nach Rolle​

👶 Einsteiger (ersten Flow in 1 Stunde erstellen)​

🧑‍🏭 Vor-Ort-Operator (Automatisierungsszenarien erstellen)​

🛠 Systemadministrator (Betrieb·Tuning·Störungsbehebung)​

🔌 Entwickler·Integrationsingenieur (Anbindung externer Systeme)​

🔐 Sicherheitsverantwortlicher (Audit·Authentifizierung)​

📚 Schnellreferenz (bereits vertraute Nutzer)​

Kernkonzepte​

Erste Schritte​

End-to-End-Tutorial​

Schritt 1 — Neuen Flow erstellen​

Schritt 2 — Trigger-Node platzieren​

Schritt 3 — Schwellenwertfilter​

Schritt 4 — Domänenwechsel (Tag → Anlage)​

Schritt 5 — Arbeitsauftrag erstellen​

Schritt 6 — E-Mail-Benachrichtigung (Erfolgszweig)​

Schritt 7 — Behandlung des Fehlerzweigs​

Schritt 8 — Speichern und Testlauf​

Schritt 9 — Validierung und Bereitstellung​

Schritt 10 — Betriebsüberwachung​

Nächste Schritte​

Bildschirmaufbau​

Listenansicht​

Werkzeuge oben​

Übersichtstabelle​

Massenbereitstellung·-aufhebung​

Alle neu bereitstellen​

Bearbeitungsansicht (visuelle Canvas)​

Werkzeugleiste oben​

Testlauf-Nutzung​

Links — Node-Palette​

Mitte — Canvas​

Rechts — Node-Konfiguration​

Zähler zurücksetzen​

Rechts — Live-Debug​

Nachrichtenstruktur​

Nachrichtentypen​

Payload-Beispiele nach Trigger​

flow_on_tag_point — Eingang eines Tag-Points​

flow_on_tag_alarm — Auftreten eines Tag-Alarms​

flow_on_asset_data / flow_on_asset_event — Anlagenereignis​

flow_on_asset_health_status / flow_on_asset_connection_status — Periodische Bewertung​

flow_on_oee_event / flow_on_ram_event / flow_on_ems_event — Plugin-Ereignis​

flow_on_opc_status / flow_on_edge_status — OPC-/Edge-Status​

flow_on_diagnostic — System-Diagnosemeldung​

flow_on_domain_changed — Domänenänderungsereignis​

flow_on_entity_event — Entitäts-Lebenszyklus​

flow_on_webhook — Externer HTTP-Push​

flow_on_mqtt_subscribe — MQTT-Topic-Abonnement​

flow_jdbc_poll — Periodisches externes DB-Polling​

flow_schedule — Zeitbasierte Auslösung​

Hinweise bei der Payload-Transformation​

Node-Katalog​

Trigger (22 Arten)​

Filter (6 Arten)​

Transformation (8 Arten)​

Aktion — Integration·Speicherung (2 Arten)​

Aktion — Anlagenereignis-Publikation (4 Arten)​

flow_asset_command_invoke — Ausführung statt Publikation​

Aktion — Domänen-CRUD (34 Arten)​

Teilweise Aktualisierung des Tag-Alarmbands (2 Arten)​

Statusübergänge des Arbeitsauftrags (WorkOrder) (5 Arten)​

Edge (22 Arten)​

3 Abfrage-Arten​

7 Arten der Docker-App-Steuerung​

Externe Anbindung (9 Arten)​

Ablaufsteuerung (7 Arten)​

Empfohlenes Verdrahtungsmuster für Retry​

Detaillierte Node-Optionsreferenz​

flow_script_filter — Skriptfilter​

Skriptkontext​

Beispiel​

flow_script_transform — Skript-Transformation​

Beispiel​

flow_switch — Mehrfachverzweigung​

Beispiel​

flow_retry — Automatischer Wiederholungsversuch​

Automatische Ergänzung von metadata​

Inhaltsverzeichnis

Überblick

Lernpfad — Empfohlene Einstiegsreihenfolge nach Rolle

👶 Einsteiger (ersten Flow in 1 Stunde erstellen)

🧑‍🏭 Vor-Ort-Operator (Automatisierungsszenarien erstellen)

🛠 Systemadministrator (Betrieb·Tuning·Störungsbehebung)

🔌 Entwickler·Integrationsingenieur (Anbindung externer Systeme)

🔐 Sicherheitsverantwortlicher (Audit·Authentifizierung)

📚 Schnellreferenz (bereits vertraute Nutzer)

Kernkonzepte

Erste Schritte

End-to-End-Tutorial

Schritt 1 — Neuen Flow erstellen

Schritt 2 — Trigger-Node platzieren

Schritt 3 — Schwellenwertfilter

Schritt 4 — Domänenwechsel (Tag → Anlage)

Schritt 5 — Arbeitsauftrag erstellen

Schritt 6 — E-Mail-Benachrichtigung (Erfolgszweig)

Schritt 7 — Behandlung des Fehlerzweigs

Schritt 8 — Speichern und Testlauf

Schritt 9 — Validierung und Bereitstellung

Schritt 10 — Betriebsüberwachung

Nächste Schritte

Bildschirmaufbau

Listenansicht

Werkzeuge oben

Übersichtstabelle

Massenbereitstellung·-aufhebung

Alle neu bereitstellen

Bearbeitungsansicht (visuelle Canvas)

Werkzeugleiste oben

Testlauf-Nutzung

Links — Node-Palette

Mitte — Canvas

Rechts — Node-Konfiguration

Zähler zurücksetzen

Rechts — Live-Debug

Nachrichtenstruktur

Nachrichtentypen

Payload-Beispiele nach Trigger

`flow_on_tag_point` — Eingang eines Tag-Points

`flow_on_tag_alarm` — Auftreten eines Tag-Alarms

`flow_on_asset_data` / `flow_on_asset_event` — Anlagenereignis

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — Periodische Bewertung

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — Plugin-Ereignis

`flow_on_opc_status` / `flow_on_edge_status` — OPC-/Edge-Status

`flow_on_diagnostic` — System-Diagnosemeldung

`flow_on_domain_changed` — Domänenänderungsereignis

`flow_on_entity_event` — Entitäts-Lebenszyklus

`flow_on_webhook` — Externer HTTP-Push

`flow_on_mqtt_subscribe` — MQTT-Topic-Abonnement

`flow_jdbc_poll` — Periodisches externes DB-Polling

`flow_schedule` — Zeitbasierte Auslösung

Hinweise bei der Payload-Transformation

Node-Katalog

Trigger (22 Arten)

Filter (6 Arten)

Transformation (8 Arten)

Aktion — Integration·Speicherung (2 Arten)

Aktion — Anlagenereignis-Publikation (4 Arten)

`flow_asset_command_invoke` — Ausführung statt Publikation

Aktion — Domänen-CRUD (34 Arten)

Teilweise Aktualisierung des Tag-Alarmbands (2 Arten)

Statusübergänge des Arbeitsauftrags (WorkOrder) (5 Arten)

Edge (22 Arten)

3 Abfrage-Arten

7 Arten der Docker-App-Steuerung

Externe Anbindung (9 Arten)

Ablaufsteuerung (7 Arten)

Empfohlenes Verdrahtungsmuster für Retry

Detaillierte Node-Optionsreferenz

`flow_script_filter` — Skriptfilter

Skriptkontext

Beispiel

`flow_script_transform` — Skript-Transformation

Beispiel

`flow_switch` — Mehrfachverzweigung

Beispiel

`flow_retry` — Automatischer Wiederholungsversuch

Automatische Ergänzung von `metadata`