探している情報	セクション
ノードIDと一行説明	ノードカタログ
ノードオプションのデフォルト値	ノードオプション詳細リファレンス
メッセージJSON例	トリガー別ペイロード例
そのまま使えるスクリプト	再利用スクリプト集 · データ変換Cookbook
API呼び出しcurl	フローREST API
実行履歴の解釈	実行履歴で何を確認できるか
トラブルシューティングガイド	段階別デバッグガイド · よくある問題
クイックオプション表	ノードクイック設定リファレンス

すべてのセクションリンクは同一文書内のアンカーです。Ctrl+F でキーワード検索も効果的です。

コア概念

用語	説明
フロー (Flow)	ノードとワイヤー(関係)の集合で構成された一つのオートメーションワークフローです。進入点がある方向グラフであり、配備/解除トグルで有効化の有無を制御します
ノード (Node)	メッセージを受け取り処理した後、次のノードに渡す単位です。7つのカテゴリ(トリガー・フィルター・変換・アクション・外部連携・フロー制御・エッジ)に区分されます
関係 (Relation)	ノード出力から次のノードへ向かうワイヤーのラベルです。ノードが直接 `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` 等のラベルを指定して分岐します。すべてのラベルは大文字に標準化されています
メッセージ (Message)	フロー内部を流れるペイロードです。`type`(分類)、`originator`(主体エンティティ)、`data`(本文)、`metadata`(コンテキスト)を含みます
トリガー (Trigger)	フローの開始点となるノードです。ドメインイベント(タグポイント・アラーム・資産イベント等)、外部進入(Webhook・MQTT・外部DB)、時間(スケジュール)の3種類があります

最初に始める

フローは次の4段階で最も速く慣れることができます。

一覧画面で 새 플로우 ボタンを押して空のフローを作成します(名前と説明のみ入力)。
編集画面で左側パレットのトリガーノード(例: flow_on_tag_alarm) → フィルター → アクション(例: flow_send_email)を順にドラッグして配置し、ノード間をワイヤーで接続します。
各ノードをクリックして右側インスペクターでオプションを入力した後、右上の保存 → テスト実行で動作を一度確認します。
上部の配備トグルをオンにすると、トリガーイベントが入るたびにフローが自動実行され、ライブデバッグパネルと実行履歴画面で結果を確認できます。

オートメーションシナリオの結線パターンは本文書のサンプルフローと活用例を参照してください。実際のフローを最初から最後まで作るステップバイステップチュートリアルはエンドツーエンドチュートリアルで進めることができます。

エンドツーエンドチュートリアル

運用環境で実際に使用可能なオートメーションフローを一つ、最初から最後まで作るステップバイステップチュートリアルです。シナリオ: モーターの温度が80°Cを超えたら自動で緊急整備の作業指示を発行し、担当者にメールで通知する。

ステップ1 — 新しいフローを作成

左メニューで Automation > フロー をクリック → 一覧画面を開く
上部右側の新規フローボタンをクリック
以下の情報を入力して確認

項目	入力値
名前	`모터 과열 자동 정비 발행`
説明	`[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com`

フローが作成されると、自動的に空のキャンバスの編集画面が開きます。

ステップ2 — トリガーノードの配置

資産テレメトリイベントからモーター温度データを受け取ります。

左側パレットのトリガー (Trigger) カテゴリを展開
flow_on_tag_point ノードをキャンバスにドラッグ
ノードをクリック → 右側インスペクターで以下のオプションを入力

オプション	値
表示名	`태그 포인트 인입`
`tag_id_pattern`	`MOTOR-*.TEMP`

tag_id_pattern でモーター温度タグのみ通過させると、後続の処理量が大幅に減ります。他のタグはSKIPPED処理となりカウンターに含まれません。

ステップ3 — 閾値フィルター

80°C超のメッセージのみ通過させるスクリプトフィルターを追加します。

フィルター (Filter) カテゴリから flow_script_filter をドラッグ
トリガーノードの出力ポートから新しいフィルターノードの入力ポートへワイヤーを接続
インスペクターに以下を入力

オプション	値
表示名	`임계값 필터 (80°C 초과)`
`language`	`JS`
`script`	`data.value > 80`

ステップ4 — ドメイン変更 (タグ → 資産)

アラーム・作業指示は資産単位で発行するのが自然なので、メッセージのoriginatorをタグから上位資産へ変換します。

変換 (Transform) カテゴリから flow_change_originator をドラッグ
フィルターノードの TRUE 出力からワイヤーを接続
インスペクター:

オプション	値
表示名	`Tag → Asset 변경`
`entity_type`	`Asset`
`id_field`	`metadata.asset_id`

metadata.asset_id はタグポイント入力時に自動で埋まります。もしメッセージに無ければ、タグIDのprefix部分(例: MOTOR-001.TEMP → MOTOR-001)をスクリプトで抽出しても構いません。

ステップ5 — 作業指示発行

緊急整備の作業指示を自動生成します。

アクション (Action) — ドメインCRUD カテゴリから flow_create_work_order をドラッグ
変換ノードの SUCCESS 出力からワイヤーを接続
インスペクター:

オプション	値
表示名	`긴급 정비 작업지시 생성`
`asset_id_field`	`originator.id`
`order_name_field`	(静的値) `긴급 점검 — 모터 과열`
`master_id_field`	(選択) `data.master_id` (無ければ自動生成)
`description_field`	(静的値) `자동 발행: 임계 온도 초과로 긴급 점검 필요`
`default_priority`	`HIGH`

ステップ6 — メール通知 (成功分岐)

作業指示の発行が成功したら担当者に通知します。

外部連携 (External) カテゴリから flow_send_email をドラッグ
作業指示ノードの SUCCESS 出力からワイヤーを接続
インスペクター:

オプション	値
表示名	`정비 담당자 이메일`
`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}`

ステップ7 — 失敗分岐の処理

作業指示の発行自体が失敗する可能性もあります(例: 資産が消失した、権限問題等)。運用者に即座にプッシュ通知を送ります。

外部連携の flow_send_push をドラッグ
作業指示ノードの FAILURE 出力からワイヤーを接続
インスペクター:

オプション	値
`title`	`자동화 실패`
`body_template`	`${originator.id} 정비 자동 발행 실패: ${data.error}`

ステップ8 — 保存とテスト実行

右上の保存ボタンをクリック。自動スナップショットが積載され、以降ロールバックできます。
右上のテスト実行ボタンをクリック → JSONエディタに以下を入力 → 発行

{
  "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" }
}

ダイアログ下部に ✓ JSON OK (type=POST_TELEMETRY) 表示 → 発行をクリック。

ライブデバッグパネルで以下を確認:
- トリガーノードが緑色点灯 → メッセージ通過
- フィルターノード: 92.5 > 80 なので TRUE 分岐
- 変換ノード: originatorがAsset/MOTOR-001に変更
- 作業指示ノード: SUCCESS、data.work_order_id が自動付与
- メールノード: 送信試行

ステップ9 — 検証と配備

作業指示画面で新しい作業指示が登録されているか確認
メールが正常に届いているか確認(テスト環境のメールボックス)
80°C未満のメッセージで再度テスト(フィルターで遮断されるべき):

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

フィルターノードの分岐は FALSE、後続ノードは灰色 — 正常。

すべての分岐を検証した後、全体カウント初期化で統計ウィンドウをリセット
上部の配備トグルをON

これで実際の運用環境でモーター温度が80°Cを超える度に自動で整備作業指示が発行され、担当者にメールが送られます。

ステップ10 — 運用モニタリング

配備直後の5〜10分間、以下を確認すると安全です。

場所	確認項目
一覧画面	該当フロー行の実行件数が正常範囲で増加しているか(暴走ではないか)
ライブデバッグパネル	ノードの失敗が無いか
実行履歴画面	失敗したメッセージがあれば `NODE_ERROR` の原因を確認
受信メール	意図しない頻度で通知が来ていないか

次のステップ

このフローを発展させるには:

Retry結線の追加 — メール/プッシュ送信失敗時にバックオフ再試行 (例8参照)
バンド自動調整 — 6時間平均±3σで閾値を自動補正 (例6)
ダウンタイム累積 — 過熱履歴を資産attributeに累積 (例14)
品質ライン隔離 — 過熱が連続発生したらライン自動停止 (例15)

画面構成

フローは以下の3画面で構成されます。

画面	用途
一覧	登録されたフローの一覧・検索・一括配備・インポート
編集	ビジュアルキャンバスでノードを配置・接続・設定
実行履歴	ノード単位の実行ログ・タイムライン照会

一覧画面

上部の検索・作成領域とフロー一覧テーブルで構成されます。

上部ツール

項目	説明
状態フィルター	`전체` / `배포` / `해제`
名前・説明検索	キーワードでフローをフィルタリングします
新規フロー	空のフローを作成します(名前・説明入力)
インポート	エクスポートで受け取ったJSONをアップロードしてフローを復元します
すべて再配置	有効化されたすべてのフローを一度に再積載します
更新	一覧を再読込します

一覧テーブル

一覧テーブルは25行単位でページネーションされ、各行に処理トレンドのスパークラインとエラー率のドーナツチャートが一緒に表示されます。ページを切り替えてもチャート状態は維持されます。

カラム	説明
選択	一括配備・解除用チェックボックス
状態	`배포` / `해제` バッジ
フローID	`FLOW_NNNNN` 形式のシーケンスID
名前 \| 説明	運用者が指定したメタ情報
ノード	含まれるノード数
トリガー	トリガーノード数
実行件数	累積メッセージ処理カウント
処理時間	ノードの平均/最近の処理時間
エラー	累積エラーカウント
最終更新	グラフの最後の保存時刻
アクション	`편집` / `삭제` ボタン

一括配備・解除

選択したフローをまとめて配備(有効化)または解除(無効化)します。配備されたフローのみトリガーイベントを受け取ります。

すべて再配置

上部のすべて再配置ボタンは、有効化されたすべてのフローを再積載します。以下の状況で使用してください。

外部からグラフを一括インポートした直後
スケジュール・外部MQTT購読・外部DBポーリング等、自前のスケジューラーがあるトリガーを再登録したい場合
運用中にキャッシュの一貫性の問題が疑われる場合

編集画面 (ビジュアルキャンバス)

キャンバス左側パレットからノードをドラッグして配置後、ノードの出力ポートをクリック&ドラッグして次のノードと接続します。

上部ツールバー

ボタン	動作
名前・説明	フローメタ情報を編集します
配備/解除	現在のフローを即座に有効/無効化します
エクスポート	グラフ全体をJSONファイルとしてダウンロードします
テスト実行	任意のJSONメッセージを注入して一度実行し、結果を確認します(下記テスト実行の使い方参照)
保存	現在のグラフをサーバーに保存します。保存時に自動スナップショットが積載され、以降ロールバックできます

トリガーが無いグラフを保存しようとすると「手動テスト実行のみで開始されます」という警告が表示されます。トリガーを意図的に外して手動実行専用フローとして使用することもできます。

テスト実行の使い方

테스트 실행 ボタンを押すとJSON編集ダイアログが開きます。運用者が直接メッセージを作成して一度発行できるため、トリガーイベントを待たずにグラフの動作を検証できます。

項目	説明
エディター	行番号・構文強調のあるJSONエディター。メッセージ本文を自由に作成できます
検証表示	`type` 必須フィールドがあれば `✓ JSON OK (type=X)`、欠落していれば `⚠ type 필수` 表示
発行	`발행` ボタンでメッセージをディスパッチャーに注入。結果はライブデバッグパネルと実行履歴で確認

基本テンプレート例

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

保存していない変更がある状態でテスト実行を押すと「サーバーは保存済みバージョンを実行します」という案内が表示されます。変更を検証するには先に保存してください。

左側 — ノードパレット

カテゴリ別に折りたたみ・展開が可能で、検索入力で即座にフィルタリングできます。

カテゴリ	色	ノード数
トリガー (Trigger)	グレー	22種
フィルター (Filter)	青	6種
変換 (Transform)	緑	8種
アクション (Action) — 統合・保存・資産発行・命令呼び出し	オレンジ	7種
アクション (Action) — ドメインCRUD	オレンジ	34種
外部連携 (External)	紫	9種
フロー制御 (Control)	グレー	7種
エッジ (Edge)	シアン	22種

中央 — キャンバス

ツール	ショートカット / 操作	動作
拡大/縮小	`Ctrl/⌘ +` / `Ctrl/⌘ -` · マウスホイール	キャンバスズーム
100%	`Ctrl/⌘ 0`	ズームリセット
画面フィット	`Ctrl/⌘ 1`	すべてのノードが見えるよう自動フィット
選択ノード削除	`Del` / `Backspace`	選択したノード/ワイヤーを削除
キャンバス移動 (panning)	空白領域を左クリックしてドラッグ	ノードをクリックしていない状態で空白のキャンバス背景をつかんで引くと、キャンバス全体が追随して動きます

キャンバス右下にはミニマップが表示され、ミニマップをクリックすると該当位置へ即座に移動できます。

💡 panning使用のヒント: ノード上でドラッグするとノードが移動します — キャンバスを移動させるには必ずノード/ワイヤーが無い空白背景領域をつかんでください。大きなフローではミニマップよりpanningの方が速いです。

右側 — ノード設定

キャンバスでノードをクリックすると、右側インスペクターに該当ノードの設定フォームが表示されます。入力フィールドはノード種類に応じて自動生成されます。

入力方式	説明
静的値	フォームに直接入力した値をそのまま使用します
*`_field` 動的値**	メッセージペイロードのパス(例: `data.tag_id`, `metadata.site_id`)から値を抽出します。値が無ければ静的値にフォールバックします

スクリプトノード(フィルター・変換・スイッチ)はインスペクター内で直接コードエディターで編集できるほか、別ダイアログに拡張して大画面で作成することもできます。

コードエディターのフォントは可読性を強化したモノスペースフォントスタック(Cascadia Code · JetBrains Mono · Consolas · Menlo優先)で表示され、日本語コメントも安定して整列されます。

カウント初期化

ノード設定パネル右上には2つの初期化ボタンがアイコンで表示されます。各ボタンにマウスを乗せるとツールチップが案内します。

アイコンボタン	動作
🩹 (絆創膏) — エラーカウント初期化	このフロー内のすべてのノードの累積エラーカウントのみを0にリセットします
🔄 (回転矢印) — 全体カウント初期化	処理・エラー・処理時間すべて + フロー単位の統計をすべて0に戻します。運用検証を終えて新たに統計を始める時に使用してください

どちらのボタンも確認ダイアログ無しで即座に適用されます — 統計のみ初期化され、ノード動作やメッセージ処理には影響ありません。

右側 — ライブデバッグ

ライブデバッグパネルがインスペクターの下に表示されます。

項目	説明
更新周期	2秒
レベルカラー	左側ボーダーにINFO(青)/WARN(黄)/ERROR(赤)を表示
表示情報	ノード表示名・処理時間(ms)・メッセージプレビュー
一時停止	パネル右上のトグルで更新を一時停止します
クリア	積載されたデバッグ項目を画面のみクリアします

キャンバスのノード右上には小さな点灯が表示されます。

色	意味
グレー	待機 — メッセージを受けていない状態
緑	メッセージ通過中
赤	処理中にエラー発生

ノード右下には 평균 X · 최근 Y 形式で処理時間が表示されます。

メッセージ構造

フロー内部を流れるメッセージは以下の4つの領域で構成されます。

{
  "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"
  }
}

領域	意味
`type`	メッセージ分類。フィルターノードの分岐基準
`originator`	メッセージ主体エンティティ(どの資産/タグ/注文についてか)
`data`	ペイロード本文
`metadata`	コンテキスト(時刻・サイト・シフト・タグID等)

メッセージタイプ

タイプ	発生進入点
`POST_TELEMETRY` / `TAG_POINT`	タグポイント入力
`POST_ATTRIBUTES`	タグ/資産メタ更新
`TAG_ALARM`	タグ単位アラーム
`ENTITY_CREATED` / `UPDATED` / `DELETED`	エンティティライフサイクルイベント
`ASSET_DATA` / `ASSET_EVENT` / `ASSET_ALARM` / `ASSET_COMMAND` / `ASSET_AGGREGATION` / `ASSET_CONTEXT`	資産ドメインイベント
`ASSET_HEALTH_STATUS` / `ASSET_CONNECTION_STATUS`	資産周期評価(ヘルス/接続状態)
`OEE_EVENT` / `RAM_EVENT` / `EMS_EVENT`	ISO分析結果イベント
`OPC_STATUS` / `EDGE_STATUS`	OPC/エッジデバイス状態
`DIAGNOSTIC` / `DOMAIN_CHANGED`	診断・ドメイン変更
`ALARM`	アラーム発生
`WEBHOOK`	HTTP Webhook受信
`KAFKA_INBOUND` / `MQTT_INBOUND`	外部トピック受信
`TIMER`	スケジュール発火

トリガー別ペイロード例

スクリプトノード作成時にどのフィールドにアクセスできるかを正確に知る必要があります。以下は各トリガーが生成する実際のメッセージJSON例です。すべてのトリガーメッセージには共通で type, originator, data, metadata が含まれます。

`flow_on_tag_point` — タグポイント入力

タグ1つに値が入るたびに発火します。最も一般的なトリガーです。

{
  "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"
  }
}

フィールド	意味	スクリプトアクセス
`data.value`	受信した値(数値/文字列/ブール)	`msg.data.value`
`data.quality`	OPC品質(GOOD/BAD/UNCERTAIN)	`msg.data.quality`
`data.ts`	受信時刻(epoch ms)	`msg.data.ts`
`metadata.tag_id`	タグID	`msg.metadata.tag_id`

`flow_on_tag_alarm` — タグアラーム発生

タグのアラームバンド(hi/lo/...)を超えた瞬間に発火します。

{
  "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
  }
}

`data.alarm_band` 値	意味
`NORMAL` / `HI` / `LO` / `HI_HI` / `LO_LO` / `TRIP_HI` / `TRIP_LO`	数値型アラーム段階
`BOOL_TRUE` / `BOOL_FALSE`	ブール型アラーム

`flow_on_asset_data` / `flow_on_asset_event` — 資産イベント

資産単位で集計されたイベント(CEP処理結果・プラグイン評価結果等)が発生した時に発火します。

{
  "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 は資産単位の時系列データ(data.values がキー・値マップ)を渡し、flow_on_asset_aggregation は分/時単位の集計値を渡します。

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 周期評価

プラットフォームが1分周期で資産単位のヘルス/接続状態を評価します。

{
  "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 }
}

`data.status` 値	意味
`OK` / `WARN` / `ERROR` / `UNKNOWN`	ヘルス段階
`CONNECTED` / `LATENT` / `ERROR` / `DISCONNECTED` / `UNKNOWN`	接続段階(connection_status)

prev_status と比較して状態が遷移した瞬間にのみ後続アクションを発火させるようフィルタリングするパターンが多く使われます。

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — プラグインイベント

作業指示単位のOEE/RAM/EMS評価結果が更新された時に発火します。発信元(originator)は設備で、計算結果は data.payload にあります。

{
  "type": "OEE_EVENT",
  "originator": { "entity_type": "Asset", "id": "ASSET_DJ_M_0001" },
  "data": {
    "event_type":     "OEE_CALCULATED",
    "plugin":         "OEE",
    "statement_type": "RESULT",
    "asset_id":       "ASSET_DJ_M_0001",
    "timestamp":      1746247200123,
    "payload": {
      "oee":            0.78,
      "availability":   0.95,
      "performance":    0.85,
      "quality":        0.97,
      "good_quantity":  { "quantity_string": "1560", "data_type": "integer", "unit_of_measure": "EA" },
      "scrap_quantity":  { "quantity_string": "42",   "data_type": "integer", "unit_of_measure": "EA" },
      "rework_quantity": { "quantity_string": "6",    "data_type": "integer", "unit_of_measure": "EA" },
      "run_time_ms":    25200000,
      "order_id":       "ORDER_000001"
    }
  }
}

good_quantity・scrap_quantity・rework_quantity はワークオーダー単位の Quantity オブジェクトです(ISA-95 / B2MML)。値は10進の文字列なので、スクリプトで計算するには Number(data.payload.good_quantity.quantity_string) のように変換し、単位は unit_of_measure で確認してください。旧 good_count・bad_count フィールドはありません。フィールドの規則は Order サービス — 数量。
scrap_quantity は廃棄のみで、再作業は rework_quantity です。不良の合計が必要なら両者を足してください(良品 + 廃棄 + 再作業 = 生産)。
計算されなかった比率(«未適用»)と数量は 0 ではなく null です。

`flow_on_opc_status` / `flow_on_edge_status` — OPC/エッジ状態

{
  "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` — システム診断メッセージ

サーバーモジュールで発生した診断メッセージが入ると発火します。

{
  "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`	意味
`INFO` / `WARN` / `ERROR`	診断シビリティ

`flow_on_domain_changed` — ドメイン変更イベント

資産・タグ・サイト・作業指示等のドメインエンティティがCRUDされた時に発火します。

{
  "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` — エンティティライフサイクル

ENTITY_CREATED / ENTITY_UPDATED / ENTITY_DELETED の3タイプに統合発火。単一ノード1つで3種のライフサイクルすべてを受け取ります。

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

`flow_on_webhook` — 外部HTTPプッシュ

外部システムが POST /webhook/flow/{flow_id} に送ったペイロードがそのままメッセージに変換されます。URLパスの {flow_id} とヘッダー X-API-Key で認証。

旧パス /flow/webhook/{flow_id} は410 Goneを返します

そのパスは認証無しで任意の flow_id でフローを発火できたため無効化されました (FlowController が410を返します)。新しい進入点 /webhook/flow/{flow_id} は WebhookController がAPIキー・HMACを検証します。前後が入れ替わったパスなので見落としやすいです — flow/webhook ではなく webhook/flow です。

{
  "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
  }
}

外部システムが送ったJSON本文全体が data にそのまま入ります。HTTPヘッダーは metadata に一部のみ(remote_addr/method/request_id)表示されます。

`flow_on_mqtt_subscribe` — MQTTトピック購読

{
  "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` — 外部DB周期ポーリング

設定されたSELECTクエリを周期的に実行し、各行ごとにメッセージ1件を発火します。

{
  "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
  }
}

行が100件なら100個のメッセージが順次発火されます。同じ行を繰り返し処理しないよう、SELECTクエリ内に処理フラグを一緒に更新するか、processed_at カラムの比較条件を入れてください。

`flow_schedule` — 時間ベースの発火

Cron式または固定周期で発火します。ペイロードは空で metadata.ts のみ埋まります。

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

ペイロード変換時の注意事項

originator.id はドメインID — flow_change_originator ノードで変更すると、その後の flow_save_attributes·flow_publish_asset_* アクションノードが新しいoriginator基準で動作します。
スクリプトで data/metadata を入れ替える時、浅いコピー (Object.assign) ではなく直接割り当てを推奨します — 原本変更時に同じトリガーを購読する他のフローに影響が及ぶ可能性があります。
すべてのepoch時刻フィールドは**ミリ秒(ms)**です。秒単位が必要な場合は Math.floor(msg.metadata.ts / 1000)。

ノードカタログ

詳細なノードIDとオプションは編集画面のインスペクターで確認できます。

トリガー (22種)

すべてのフローの開始点はトリガーノードです。別途の進入点/終端点ノードはありません。

ドメイン自動受信 — システム内部イベントが自動でディスパッチされます。

カテゴリ	トリガーノード
タグ	`flow_on_tag_point` (テレメトリ), `flow_on_tag_alarm`
資産	`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`
プラグイン	`flow_on_oee_event`, `flow_on_ram_event`, `flow_on_ems_event`
OPC/エッジ	`flow_on_opc_status`, `flow_on_edge_status`
診断・ドメイン	`flow_on_diagnostic`, `flow_on_domain_changed`
エンティティ	`flow_on_entity_event`

flow_on_alarm というノードはありません

アラームトリガーは対象によって2つに分かれます — タグアラームは**flow_on_tag_alarm、資産アラームはflow_on_asset_alarm**です。以前の文書の例が flow_on_alarm を使っていましたが、パレットに無い名前なのでそのまま試すとノードが見つかりません。

すべてのトリガーノードは *_pattern オプション(グロブ: *, ?)でメッセージ単位の事前フィルタリングができます。パターン未マッチのメッセージは後続ノードへ渡されず、実行カウントも増加しません(SKIPPED処理)。運用負荷を最小化するためトリガー段階で先にフィルタリングすることを推奨します。

外部進入

ノード	動作
`flow_on_webhook`	外部システムがHTTPでプッシュしたペイロードを受信
`flow_on_mqtt_subscribe`	外部MQTTブローカーのトピックを購読
`flow_jdbc_poll`	外部データベースのSELECT結果を周期的に読み、行ごとに発火

時間ベース

ノード	動作
`flow_schedule`	クーロン/周期スケジュール — `TIMER` メッセージ発火

フィルター (6種)

ノード	説明
`flow_msg_type_filter`	`type` が指定リストに含まれれば `TRUE`
`flow_originator_type_filter`	`originator.entity_type` が指定リストに含まれれば `TRUE`
`flow_script_filter`	スクリプトでboolean評価
`flow_check_existence_field`	`data`/`metadata` 特定フィールドの存在有無
`flow_switch`	多重case分岐(caseごとに異なるrelation)
`flow_check_relation`	前段階のrelation基準の分岐

変換 (8種)

ノード	画面名	説明
`flow_script_transform`	スクリプト変換	スクリプトで `data`/`metadata` 変換
`flow_change_originator`	発信者変更	`originator` を他のエンティティに変更
`flow_rename_keys`	キー名変更	`data` フィールド名を一括変更
`flow_template`	テンプレート	`${path}` 置換テキスト生成
`flow_split`	分割	`data` が配列であれば各要素別メッセージに分割
`flow_merge`	統合	複数のメッセージを時間ウィンドウで統合 — `flow_split` の逆
`flow_flatten`	平坦化	ネストしたオブジェクトの子キーを最上位に引き上げ (`data.data_map.x` → `data.x`)
`flow_to_email`	メール変換	メッセージをメールフォーマットに変換

flow_flatten はタグポイントのように data_map 内に値がもう一段入っているメッセージを後続ノードで ${data.x} で直接参照したい場合に使います。

アクション — 統合・保存 (2種)

ノード	説明
`flow_save_tag_point`	タグポイント積載(通常のインジェスト経路と同一)
`flow_save_attributes`	タグ/資産メタ部分更新

flow_dds_publish (内部メッセージチャンネルへの自由発行)をお探しの場合、ここではなく 外部連携 パレットグループにあります。

アクション — 資産イベント発行 (4種)

CEP(EQL)と同一の処理経路で資産イベントを発行します(永久保存 + キャッシュ + タイムライン + プラグイン + メッセージチャンネルの一貫処理)。

ノード	チャンネル
`flow_publish_asset_event`	資産イベント
`flow_publish_asset_context`	資産コンテキスト
`flow_publish_asset_aggregation`	資産集計
`flow_publish_asset_command`	資産命令

`flow_asset_command_invoke` — 発行ではなく実行

ノード	画面名	行うこと
`flow_asset_command_invoke`	アセット命令呼び出し	資産に定義された命令を同期実行し、結果を待ちます

flow_publish_asset_command と混同しやすいです

flow_publish_asset_command — 資産命令イベントをチャンネルに発行します。発行して終わりです。
flow_asset_command_invoke — 資産に定義された命令を実際に実行し、結果を受け取るまで待ちます。

命令が実行されることを期待して発行ノードを使うと、何も起こらなかったように見えます。

アクション — ドメインCRUD (34種)

ドメイン作業はすべて同一ドメインサービスに委譲され、監査・整合性が維持されます。*_field 動的オプションでメッセージペイロードから値を抽出できます。

ドメイン	Create	Update	Delete
資産 (Asset)	`flow_create_asset`	`flow_update_asset`	`flow_delete_asset`
タグ (Tag)	`flow_create_tag`	`flow_update_tag`	`flow_delete_tag`
サイト/エリア/ライン	`flow_create_site`	`flow_update_site`	`flow_delete_site`
作業指示 (WorkOrder)	`flow_create_work_order`	`flow_update_work_order`	`flow_delete_work_order`
アラーム設定 (EQL)	`flow_create_alarm_config`	`flow_update_alarm_config`	`flow_delete_alarm_config`
顧客 (Customer)	`flow_create_customer`	`flow_update_customer`	`flow_delete_customer`
製品 (Product)	`flow_create_product`	`flow_update_product`	`flow_delete_product`
作業者 (Employee)	`flow_create_employee`	`flow_update_employee`	`flow_delete_employee`
カレンダー(シフト)	`flow_create_calendar`	`flow_update_calendar`	`flow_delete_calendar`

タグアラームバンド部分更新 (2種)

ノード	説明
`flow_update_tag_alarm_band_numeric`	数値型アラームバンド(`hi`/`lo`/`hi_hi`/`lo_lo`/`trip_hi`/`trip_lo`/`band_message`/`use_alarm`)を入力したフィールドのみ部分更新
`flow_update_tag_alarm_band_boolean`	ブール型アラームバンド(`bool_true`/`bool_false`/優先度/メッセージ/`use_alarm`)を入力したフィールドのみ部分更新

作業指示(WorkOrder)状態遷移 (5種)

数値カラムを直接更新せず、ドメインサービスの状態遷移メソッドを呼び出すため、OEE/RAM/EMSの可視性が維持されます。

ノード	遷移
`flow_start_work_order`	`WAIT` → `START`
`flow_pause_work_order`	`START` → `PAUSED`
`flow_resume_work_order`	`PAUSED` → `START`
`flow_end_work_order`	`START` または `PAUSED` → `END`
`flow_abort_work_order`	`START` または `PAUSED` → `ABORTED` (`abort_code`, `notes` オプション)

NOT NULL自動補強 — CreateノードはNOT NULLカラムにdefault値を自動で埋めます。例: 作業指示 status="WAIT" / master_idはMESマスターID(無ければ自動バックフィル)、顧客マネージャー情報 "admin"/"admin@example.com"、作業者 org_id は site_id にフォールバック、すべての行の insert_user_id="flow"。FKカラム(例: customer_id/product_id)で空文字列はNULLに変換されます。

Updateノード — 部分更新 — Customer/Product/Employee/Calendarとアラームバンドの Updateノードは既存レコードを先に照会した後、入力したフィールドのみをマージして保存します。空の文字列・null値は無視され既存値が保持されます。全体上書きが必要な場合はDelete + Createの組み合わせを使用してください。

アラーム直接トリガーノードは意図的に除外されています。アラームは flow_create_alarm_config 経路でのみ発生すべきで、アラーム履歴の整合性が維持されます。

エッジ (Edge, 22種)

エッジデバイス(OPC Agent)のREST APIを呼び出してOPCサーバー登録・タグCRUD・タグ値読み取り/書き込み照会・Dockerアプリ制御を自動化します。すべてのノードが意味的に基本HTTPメソッド(GET/POST/PUT/DELETE)のみ異なる同一動作を共有します。

グループ	ノード
OPCサーバー管理	`flow_edge_opc_create`, `flow_edge_opc_update`, `flow_edge_opc_delete`, `flow_edge_opc_start`, `flow_edge_opc_stop`, `flow_edge_opc_list`
タグ管理	`flow_edge_tag_create`, `flow_edge_tag_update`, `flow_edge_tag_delete`, `flow_edge_tag_read`, `flow_edge_tag_write`, `flow_edge_tag_list`
照会	`flow_edge_monitoring`, `flow_edge_info`, `flow_edge_transfer`
Dockerアプリ制御	`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種

ノード	画面名	行うこと
`flow_edge_monitoring`	モニタリング	エッジモニタリング指標を照会
`flow_edge_info`	エッジ情報	エッジID・バージョン・uptimeを照会
`flow_edge_transfer`	転送ヘルス	MQTT/Sparkplugの転送状態

ノード	画面名	行うこと
`flow_edge_opc_list`	OPC一覧	エッジのOPCサーバー一覧を照会
`flow_edge_tag_list`	タグ一覧	OPCのタグ一覧を照会

Dockerアプリ制御7種

エッジ詳細画面のDockerパネルで手動で行っていた作業をフローで自動化します。

ノード	画面名	行うこと
`flow_edge_app_list`	アプリ一覧	エッジDockerコンテナ一覧
`flow_edge_app_inspect`	アプリ詳細	コンテナ詳細(inspect)
`flow_edge_app_start`	アプリ開始	コンテナ開始
`flow_edge_app_stop`	アプリ停止	コンテナ停止
`flow_edge_app_restart`	アプリ再起動	コンテナ再起動
`flow_edge_app_logs`	アプリログ	コンテナログ (`line=N`)
`flow_edge_app_stats`	アプリ統計	コンテナCPU / メモリ / I/O

アプリ制御ノードは現場機器を実際に操作します

app_stop · app_restart は現場エッジで動くコンテナを本当に停止・再起動させます。トリガーに繋げて自動実行させる時は条件を狭く設定してください — フラッピングするアラームに app_restart を紐付けるとコンテナが繰り返し再起動します。

共通設定

オプション	説明
`url`	エッジRESTエンドポイント。`${data.x}`/`${metadata.y}` テンプレート置換対応
`method`	HTTPメソッド(未設定時はノード別デフォルト値 — 例: create=POST, update=PUT, delete=DELETE, read/monitoring=GET)
`headers`	JSONヘッダー (例: `{"Authorization":"Bearer ${TOKEN}"}`)
`body_template`	リクエスト本文(未設定時は `data` をそのまま送信、GET/DELETEは本文非送信)
`timeout_ms`	タイムアウト(デフォルト5000)

応答・分岐

data.response_status — HTTPステータスコード
data.response — 応答本文(文字列)
data.error — エラーメッセージ(失敗時)
SUCCESS (200~399) / FAILURE (それ以外または例外)

外部連携 (9種)

すべての外部ノードは *_field 動的オプションをサポートします。

ノード	動的オプション
`flow_dds_publish`	内部メッセージチャンネルへの自由発行(外部呼び出しではないがこのグループに含まれます)
`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静的 (SELECT/INSERT/UPDATE/DELETE)

フロー制御 (7種)

ノード	説明
`flow_log`	デバッグログ (level / prefix)
`flow_noop`	通過
`flow_delay`	`delay_ms` 後に次のノードへ渡す
`flow_throttle`	`max_msgs` / `window_ms` 制限(超過時 `THROTTLED` relation)
`flow_debounce`	`window_ms` 安定化後最後のメッセージのみ発火
`flow_merge`	`window_ms` 間に入力を累積し `data.merged` 配列で一度emit
`flow_subflow`	`target_flow_id` — 他のフローを呼び出し
`flow_retry`	`max_attempts`(デフォルト3) / `backoff_ms`(デフォルト1000) / `backoff_multiplier`(デフォルト2.0)。バックオフ待機後 `SUCCESS` 分岐へメッセージ伝達、最大試行到達時 `EXHAUSTED` 分岐。`metadata.retry_count` / `metadata.retry_exhausted` 自動付与

Retry推奨結線パターン

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

ノードオプション詳細リファレンス

複雑なノードのオプションを一行表ではなく、オプション別デフォルト値・例・失敗処理まで詳細に説明します。

`flow_script_filter` — スクリプトフィルター

オプション	タイプ	デフォルト値	説明
`script`	text	(必須)	評価式 — boolean返却。`true` → `TRUE` 分岐 / `false` → `FALSE` 分岐
`script_type`	enum	`javascript`	`javascript` / `eql`
`on_error`	enum	`FALSE`	スクリプト例外時 — `TRUE` / `FALSE` / `FAILURE` 分岐へ送るか

スクリプトコンテキスト

変数	意味
`msg.type`	メッセージタイプ (POST_TELEMETRY等)
`msg.data`	ペイロード(修正可能だがフィルターでは意味を持ちません)
`msg.metadata`	コンテキスト
`msg.originator`	originatorオブジェクト

例

// 온도가 임계 초과 + 야간 시프트만
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` — スクリプト変換

オプション	タイプ	デフォルト値	説明
`script`	text	(必須)	変換式 — `msg` オブジェクトを修正するか新しいオブジェクトを返却
`script_type`	enum	`javascript`	`javascript` / `eql`
`mode`	enum	`mutate`	`mutate`(in-place) / `return`(return値使用)

例

// 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
};

mutate モードでは return 文があっても無視されます。新しいオブジェクトに交換したい場合は mode=return に設定する必要があります。

`flow_switch` — 多重分岐

オプション	タイプ	デフォルト値	説明
`cases`	array	(必須)	`[{expression, relation}]` 配列 — 上から下へ評価、最初のマッチを採用
`default_relation`	string	`DEFAULT`	すべてのcase未マッチ時
`script_type`	enum	`javascript`	—

例

// 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"

後続ノードで4つの分岐ラベル(CRITICAL/WARN/INFO/NORMAL)によりそれぞれ異なる処理が可能です。

`flow_retry` — 自動再試行

外部IOノードの一時的失敗を自動復旧します。

オプション	タイプ	デフォルト値	説明
`max_attempts`	int	3	最大試行回数(この値を超えると `EXHAUSTED`)
`backoff_ms`	long	1000	最初の待機時間 (ms)
`backoff_multiplier`	double	2.0	指数バックオフ倍数 — 1次1秒 → 2次2秒 → 3次4秒
`max_backoff_ms`	long	30000	単一待機の上限
`jitter_pct`	int	0	バックオフに±N%のランダムな揺らぎ(thundering herd回避)

`metadata` 自動補強

フィールド	意味
`metadata.retry_count`	現在までの試行回数
`metadata.retry_exhausted`	trueならEXHAUSTED分岐へ進入
`metadata.retry_last_error`	最後の失敗理由

バックオフの合計が60秒を超えるとトリガー処理キューが詰まる可能性があります。外部システムの応答が一貫して遅い場合は flow_throttle で進入速度自体を制限してください。

`flow_on_webhook` — HTTPトリガー

オプション	タイプ	デフォルト値	説明
`auth_required`	boolean	`true`	`X-API-Key` ヘッダー必須有無 — 切ると誰でも呼び出し可能
`allowed_origins`	csv	`*`	CORS Originホワイトリスト
`max_body_kb`	int	256	本文サイズ上限(超えると413応答)
`payload_pattern`	glob	`*`	メッセージ事前フィルタリンググロブ

呼び出し方法

curl -X POST \
  https://platform.example.com/webhook/flow/{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} は一覧画面でコピー可能
X-API-Key はエッジAPI KeyまたはAPI認証トークン画面で発行されたトークン
応答: 200 OK (メッセージキューにenqueueされる) / 401(認証失敗) / 404(フローが無いか未配備) / 413(本文超過)

`flow_http_request` — 外部HTTP呼び出し

オプション	タイプ	デフォルト値	説明
`url`	string	(必須)	呼び出すURL。`${data.x}` テンプレート置換
`url_field`	string	—	URLをペイロードから動的に取得する時 — `data.endpoint` 等
`method`	enum	`GET`	`GET` / `POST` / `PUT` / `DELETE` / `PATCH`
`method_field`	string	—	メソッドをペイロードから動的に
`headers`	json	`{}`	`{"Authorization": "Bearer ${TOKEN}"}` 形式
`body`	text	—	静的本文(テンプレート置換対応)
`body_field`	string	—	本文をペイロードから取得する時 — 通常 `data`
`timeout_ms`	int	5000	応答待機上限
`follow_redirect`	boolean	`true`	3xxリダイレクトの自動追従
`verify_ssl`	boolean	`true`	TLS証明書検証(テスト用のみ切ること)

応答ペイロード補強

フィールド	意味
`data.response_status`	HTTPステータスコード (200 / 404 / 500 ...)
`data.response_body`	応答本文 (JSONなら自動パース)
`data.response_headers`	応答ヘッダーオブジェクト

分岐

SUCCESS — 2xx/3xx
FAILURE — 4xx/5xxまたは例外/タイムアウト

`flow_send_email` — メール送信

オプション	タイプ	デフォルト値	説明
`to`	string	—	静的受信者(カンマ区切り)
`to_field`	string	—	ペイロードから受信者を抽出 — `data.recipient` 等
`cc` / `cc_field`	string	—	CC
`bcc` / `bcc_field`	string	—	BCC
`subject` / `subject_field`	string	(1つ必須)	件名 — テンプレート置換
`body` / `body_field`	text	(1つ必須)	本文 (HTML可)
`is_html`	boolean	`true`	テキストメールなら切ること
`attachments`	json	`[]`	`[{"url":"...","filename":"..."}]`

SMTP設定はシステム → 設定のメール設定で運用者が予め登録。登録前はすべてのメールノードが FAILURE 分岐に落ちます。

`flow_jdbc_poll` — 外部DBポーリング

オプション	タイプ	デフォルト値	説明
`datasource_id`	string	(必須)	システム → 設定に登録された外部DB識別子
`query`	sql	(必須)	SELECTクエリ — 一度に最大1,000行返却
`poll_interval_ms`	int	60000	ポーリング周期(デフォルト1分)
`marker_column`	string	—	"最後処理時刻"カラム — マーカー以降の行のみSELECT
`marker_initial`	string	`1970-01-01 00:00:00`	初回ポーリング時のマーカー開始値
`row_limit`	int	1000	1ポーリングあたり最大行数(超えても安全)
`on_error_continue`	boolean	`true`	DBエラー時は診断のみ記録し次のポーリングを継続

`marker_column` 使用例

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

→ :marker の位置に最後のポーリングで受け取った最大 created_at 値が自動で入ります。

`flow_kafka_publish` / `flow_mqtt_publish` — 外部発行

オプション	タイプ	デフォルト値	説明
`broker`	string	(必須)	`kafka:9092` または `tcp://mqtt:1883`
`topic`	string	—	静的トピック — `${data.x}` 置換可能
`topic_field`	string	—	ペイロードからトピック抽出 (`data.target_topic` 等)
`key` / `key_field`	string	—	(Kafkaのみ)メッセージキー
`body` / `body_field`	json/text	(1つ必須)	発行本文 — 未設定時 `data` をそのまま
`qos`	int	1	(MQTTのみ) 0/1/2
`retain`	boolean	false	(MQTTのみ) Retainedフラグ

`flow_publish_asset_*` — 資産イベント発行

資産イベント(イベント/コンテキスト/集計/命令)はCEP/プラグイン/タイムラインが同時に認識する統合チャンネルに発行されます。4つのノードの共通オプション:

オプション	タイプ	デフォルト値	説明
`asset_id`	string	—	静的資産ID
`asset_id_field`	string	—	ペイロードから資産ID抽出(通常 `metadata.asset_id`)
`event_type`	string	—	資産イベント分類(例: `STARTUP`, `SHUTDOWN`, `MAINTENANCE`)
`event_type_field`	string	—	ペイロードから分類抽出
`payload`	json	`${data}`	発行本文 — 未設定時 `data` をそのまま

flow_publish_asset_command の場合は資産の命令受信トピック(asset_id/cmd/{event_type})へ即座に伝達されエッジに到達します。

`flow_create_` / `flow_update_` — ドメインCRUD

すべてのCreate/Updateノードは以下のオプションパターンを共有します。

オプション	タイプ	説明
`{컬럼명}`	string	静的値(入力しなければNULL/default)
`{컬럼명}_field`	string	ペイロードから値抽出(`data.foo` / `metadata.bar`)
`id_strategy`	enum	`auto`(システム発給) / `field`(`{도메인}_id_field` から抽出)
`on_duplicate`	enum	`error`(デフォルト) / `skip` / `update` — Createノードのみ

NOT NULL自動補強

CreateノードはNOT NULLカラムにdefault値を自動で埋めます。

ドメイン	自動で埋まるカラム
作業指示	`status="WAIT"` / `master_id`(無ければ自動バックフィル) / `insert_user_id="flow"`
顧客	マネージャー情報 `"admin"` / `"admin@example.com"` (未登録の場合)
作業者	`org_id=site_id`(フォールバック)
共通	`insert_date=now()` / `insert_user_id="flow"`

FK空文字列処理

FKカラム(customer_id/product_id等)に空文字列 "" が入ると自動で NULL に変換されます。JSでは delete msg.data.customer_id より msg.data.customer_id = '' の方が安全です。

`flow_edge_*` — エッジREST呼び出し

エッジデバイスのRESTエンドポイントを呼び出す22種のノードは共通オプションを共有します。

オプション	タイプ	デフォルト値	説明
`edge_id`	string	—	静的エッジID
`edge_id_field`	string	—	ペイロードからエッジID抽出
`path`	string	(ノード別デフォルト)	エッジRESTパス (例: `/api/v1/opc`, `/api/v1/app/grafana/start`)
`body_template`	text	—	リクエスト本文 — 未設定時 `data` をそのまま
`timeout_ms`	int	5000	—

自動認証

エッジノードは呼び出し時に mm_edge マスターテーブルからそのエッジの api_key を自動lookupし、X-API-Key ヘッダーとして添付します。運用者が別途設定する必要はありません。

応答ペイロード

フィールド	意味
`data.edge_response_status`	エッジ応答コード
`data.edge_response`	応答本文
`data.edge_id`	呼び出し対象エッジID(確認用)

分岐は flow_http_request と同一 (SUCCESS / FAILURE)。

スクリプトノード作成

flow_script_filter / flow_script_transform / flow_switch ノードは2種類の表現式をサポートします。

JavaScript (デフォルト推奨)

標準ECMAScript構文。複数行・var/let/const·関数・オブジェクトリテラルをすべてサポートします。

バインディング

変数	説明
`msg`	メッセージ全体。`msg.data.x`, `msg.metadata.topic`, `msg.type`, `msg.originator.id` すべて直接アクセス
`data`	`msg.data` 短縮別名
`metadata`	`msg.metadata` 短縮別名

フィルター例

data.temp > 80

変換例

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

スイッチ例 (case別boolean)

data.t > 100   // case "Critical"

よく使う変換パターン

// 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

よく使うフィルターパターン

// 우선순위 화이트리스트
['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

ユーザースクリプトは安全なサンドボックスで実行され、ファイル・ネットワーク・スレッド・任意クラスアクセスはすべて遮断されます。外部システム呼び出しが必要な場合は flow_http_request のような外部連携ノードを別途結線してください。

EQL表現式

既存EQLユーザー互換用。単一式のみサポート(複数行・セミコロン区切り非対応)され、副作用パターンのみ使用可能です。

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

複数動作が必要な場合はJavaScriptを使用してください。

JavaScript実行環境仕様

スクリプトは分離されたサンドボックスで実行されます。どんな機能が可能/不可能か正確に理解することが安定したスクリプト作成に役立ちます。

使用可能 (✅)

機能	備考
標準ECMAScript	`var`/`let`/`const`·関数・クラス・分割代入・spread·`?.`·`??`等
オブジェクトリテラル	`{ key: value, ... }`
配列メソッド	`map` / `filter` / `reduce` / `forEach` / `find` / `some` / `every` / `flat` / `slice`
文字列メソッド	`split` / `replace` / `includes` / `match` / `padStart` / `repeat`
数学関数	`Math.*` 全体
JSON	`JSON.parse` / `JSON.stringify` (ただし `data` が既にオブジェクトなら再度stringify不要)
Date	`new Date()` / `Date.now()` / `getHours()` / `toISOString()` 等
正規表現	`/pattern/` リテラル + `RegExp` コンストラクタ
エラーthrow	`throw new Error('...')` — `FAILURE` 分岐に自動分岐
try/catch/finally	例外処理

使用不可 (❌)

機能	理由 / 代替
ネットワーク呼び出し (`fetch` / `XMLHttpRequest`)	サンドボックス遮断 — `flow_http_request` ノードを別途結線
ファイルシステム (`require('fs')`)	サンドボックス遮断
スレッド (`setTimeout` / `setInterval` / `Worker`)	同期実行のみ許可 — 遅延が必要なら `flow_delay` ノード
`require` / `import`	外部モジュールロード不可 — 必要な関数は同じスクリプト内に定義
`eval` / `new Function(string)`	セキュリティ上遮断
任意Javaクラス	EQL互換モードでもユーザーコードに露出されません
`process` / `global` / `window`	定義されていません
WebSocket / EventSource	遮断 — メッセージ受信はトリガーノード

実行制限 — ありません

スクリプトには時間・メモリ制限がかかっていません

以前の文書は実行時間500ms・メモリ16MB・スタック1024・出力2MBという表を載せていましたが4つとも適用されていません。

ノード設定に timeout_ms を使えるように見えますが、スクリプトノードはその値を読みません。(ScriptTransformNode · ScriptFilterNode は script と language のみ読みます。)
JS実行コンテキストにも時間・構文数・メモリ制限は設定されていません。

そのため while (true) {} のようなスクリプトは中断されずワーカースレッドを占有します。 フロー全体の限度である30秒(flow_timeout_ms)は実行ループが次の作業を取り出す際に確認されるため、スクリプト内で回る無限ループはその検査に到達しません。

**自ら終了するスクリプトのみ使用してください。**無限ループ・巨大な繰り返し・大きな文字列の累積を避け、ループには必ず上限を設けてください。

実際に遮断されているもの — セキュリティサンドボックス

時間・メモリの代わりに、外部に触れる行為は確実に遮断されています。

項目	状態
Javaクラスアクセス (`Java.type`等)	遮断
ファイル・ネットワークIO	遮断
スレッド生成	遮断
ネイティブアクセス	遮断
実験的オプション	遮断
ホストオブジェクトアクセス	許可リストにあるものだけ

外部呼び出しが必要な場合はスクリプト内で行わず flow_http_request のような専用ノードに分離してください — スクリプトではどうしてもできず、専用ノードには timeout_ms が実際に適用されます。

重い処理はスクリプト1つに集中させず複数のノードに分けてください。

`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
};

標準時刻・ロケール

項目	動作
サーバーシステムタイムゾーン	UTC (epoch ms直接使用推奨)
日本時刻表示	`toLocaleString('ko-KR', { timeZone: 'Asia/Seoul' })`
日本時刻計算	`new Date(ts + 9*3600000).getUTCHours()` または上記ロケール
0秒単位timestamp	`Math.floor(Date.now() / 1000) * 1000`

メモリセーフな作成パターン

スクリプトノードにはメモリ16MBの上限がありますが、頻繁に呼ばれるノードで小さなメモリリークが累積するとエンジンのGC負荷が大きくなります。次のパターンを避けてください。

アンチパターン1 — クロージャが大きなデータを保持

// ❌ 나쁜 예 — 큰 배열을 변환 함수에 캡쳐
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

アンチパターン2 — 大容量配列をそのまま渡す

// ❌ 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

アンチパターン3 — 深いオブジェクトのコピー

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

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

アンチパターン4 — 正規表現の爆発 (Catastrophic Backtracking)

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

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

アンチパターン5 — `let` 累積変数を関数の外に

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

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

アンチパターン6 — 入れ子のtry/catch無限試行

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

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

スクリプト内で throw した例外は FAILURE 分岐へ行き、メッセージは保存されます。外部IOノードは別途自体の分岐を持つのでtry/catchで包まないでください。

メモリ圧迫の診断

NODE_SCRIPT_OOM のようなコードはありません

以前の文書が NODE_SCRIPT_OOM · FLOW_MSG_TRIMMED · ENGINE_GC_PRESSURE をメモリ圧迫の信号として案内していましたが、**製品がそのようなコードを生成しません。**スクリプトメモリの上限もなくペイロード自動切り捨てもありません。

メモリ圧迫はコードではなく指標と症状で判断してください。

見るべきもの	圧迫の信号
`FLOW_EXECUTOR_DROPPED_COUNT`	0から上がり始める — 処理プールが飽和しメッセージを捨てている
`FLOW_EXECUTOR_QUEUE_SIZE`	減らずに増加し続ける
`FLOW_EXECUTION_TIME`	最大値が平常時の数倍に跳ねる
サーバーログ	`work_queue full ... task dropped` 警告、`FlowExecutor timeout` 警告
JVM	GC時間増加・ヒープ使用率上昇

上記の信号が見えたら大きなペイロードを扱うフローから点検し、スクリプトで大容量文字列配列を累積していないか確認してください。

モバイルプッシュ通知チャンネル

flow_send_push ノードで運用者モバイルアプリにプッシュ通知を送信します。

ノードオプション

オプション	タイプ	デフォルト値	説明
`to` / `to_field`	string	(1つ必須)	受信者ユーザーID(コンマ区切り)またはペイロードパス
`title` / `title_field`	string	(1つ必須)	通知タイトル(50文字以内推奨)
`body` / `body_field`	text	(1つ必須)	本文(120文字以内推奨)
`priority`	enum	`NORMAL`	`NORMAL` / `HIGH` — HIGHはロック画面でも表示
`sound`	enum	`default`	`default` / `silent` / カスタム音
`data`	json	`{}`	アプリが受け取って処理する付加データ(ペイロード4KB上限)
`deep_link`	string	—	通知タップ時に開く画面 (例: `pp://asset/MOTOR-001`)
`ttl_sec`	int	86400	未受信時の保管時間(秒) — 期限切れ時自動削除

ユーザー別トークン自動ルーティング

運用者がモバイルアプリを初めてログインすると、デバイストークンがセキュリティ → API認証トークンに自動登録されます。フローではトークンを直接扱わずユーザーIDのみ指定すればOKです。

iOS/Android両方のトークンが登録されていれば両デバイスとも送信
トークンが無効(アプリ削除等)なら自動的に登録解除

単純送信例

[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}"

多言語プッシュ

// 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

以降 flow_send_push の title_field=data.push_title / body_field=data.push_body として使用。

グループ通知まとめ (Inbox-style)

複数のアラームを一度にまとめて一つの通知に(5分単位):

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

ユーザー不在時の自動エスカレーション

プッシュ未確認30分後にSMSまたはメールで自動エスカレーション。

[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}"

送信頻度制限の推奨

優先度	推奨頻度
HIGH (ロック画面表示)	ユーザーあたり時間当たり5件以下 — アラーム疲労防止
NORMAL	ユーザーあたり時間当たり20件以下

flow_throttle ノードを送信前に結線するか、同じ資産のアラームは flow_debounce で安定化後送信してください。

外部認証トークン自動更新

OAuth2のような期限切れになるトークンを外部システム呼び出し時に自動で更新するパターン。

単純更新 — 時間ベース (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}}

以降、他のフローの flow_http_request では:

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

積極的更新 — 401を受け取った時

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

Refresh Token使用

[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]

トークン期限切れ閾値自動アラーム

[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} 만료 예정입니다."

認証情報の保管推奨位置

種類	推奨位置
静的(変更ほぼ無し)	システム → 設定の認証情報保管庫
動的(自動更新)	上記パターンで `flow_save_attributes` を使用し資産メタに保存
ユーザー別OAuth	セキュリティ → API認証トークン画面

認証情報はグラフ内に平文で置かず、必ず外部保管庫参照(${creds.x})形式で作成してください。インポート/エクスポート時にも平文トークンがJSONに含まれません。

language オプションでどの言語を使用するか選択します (JS または EQL)。

実行履歴画面

ノード単位の実行イベントを時系列で照会できます。

検索条件

項目	説明
時間	照会期間を指定します
レベル	`전체` / `INFO` / `WARN` / `ERROR`
フローID	特定のフローのみフィルター
メッセージID	単一メッセージ追跡用
個数	最近200/500/1,000件

クイック時間範囲

タイムラインパネル上部のボタンで即座に移動: 10분전 / 30분전 / 1시간전 / 6시간전 / 12시간전 / 전체기간。

結果カラム

カラム	説明
レベル	INFO / WARN / ERROR
時間	イベント発生時刻
イベント	`FLOW_START` / `NODE_IN` / `NODE_OUT` / `NODE_ERROR` / `FLOW_END`
フローID	どのフローか
ノード	ノード表示名
ノードタイプ	例: `flow_script_transform`
関係	`SUCCESS` / `FAILURE` / `TRUE` / `FALSE` / `MATCH` / `NO_MATCH` / `DEFAULT` / `THROTTLED` / `EXHAUSTED` (すべて大文字)
メッセージ	メッセージタイプ・主体・関係・処理時間・データプレビューの要約

CSVダウンロードボタンで現在の照会結果をエクスポートできます。

実行履歴の保存期間は7日です。長期保管が必要な場合は外部ログシステムへ積載してください。

一般ワークフロー

新規フロー作成 — 一覧画面 새 플로우 ボタン → 名前・説明入力
編集画面へ進入 — 自動で空のキャンバスが開きます
トリガーノード配置 — 左側パレットからトリガーノードをドラッグ
処理ノード追加 — フィルター → 変換 → アクションの順に配置してワイヤーで接続
ノード設定 — 各ノードをクリックして右側インスペクターでオプション入力
保存 — 右上の 저장 ボタン(自動スナップショット積載)
テスト実行 — 任意のメッセージを注入して結果を確認
配備 — 배포 トグルで有効化 → トリガーイベントが入ると自動実行
モニタリング — ライブデバッグパネルと実行履歴画面で確認

サンプルフロー

各例はノード結線ダイアグラム + 核心ノード設定 + 動作説明で構成されます。JSON形式のノード設定は右側インスペクターで入力する値と1:1対応します。

例1: MES作業指示自動生成

MESから新規POを毎分ポーリングし作業指示に変換します。

[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: 실패 알림]

ノード設定

ノード	核心設定
`flow_schedule`	`cron`: `0 * * * * ?` (毎分0秒)
`flow_http_request`	`method`: `GET`, `url`: `https://mes.example.com/api/po/list?status=NEW`, `headers`: `{"Authorization":"Bearer ${MES_TOKEN}"}`
`flow_split`	`path`: `data` (配列に分割)
`flow_script_transform`	`language`: `JS`, 下記スクリプト参照
`flow_create_work_order`	`master_id_field`: `data.master_id`, `asset_id_field`: `data.asset_id`, `order_name_field`: `data.order_name`, `planned_quantity_field`: `data.planned_quantity`, `unit_of_measure_field`: `data.unit_of_measure`, `external_system_field`: `data.external_system`, `external_id_field`: `data.external_id`
`flow_send_email`	`to_field`: `metadata.alert_to`, `subject`: `[MES 동기화 실패] ${data.po}`

Script Transform例

data.master_id = 'WO-MES-' + data.po;
data.order_name = data.product_name + ' (' + data.qty + ' ' + data.uom + ')';
data.asset_id  = data.line_id || 'UNASSIGNED';
data.planned_quantity  = String(data.qty);   // 計画数量 — 0 より大きい10進数
data.unit_of_measure   = data.uom;           // UN/ECE Rec 20 コード(EA・KGM・LTR …) — 必須、既定値なし
data.external_system   = 'MES';              // 外部ワークオーダーは (システム, 番号) のペアで識別
data.external_id       = data.po;
data.due_date  = data.delivery_date;
metadata.alert_to = 'ops@example.com';
msg

例2: 外部Webhookで資産命令発行

外部システムが送ったHTTPペイロードを検証した後、内部資産命令に変換します。

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

呼び出し方法: POST /webhook/flow/{flow_id} でJSON本文を送信すると data 領域にそのまま載った状態でトリガーが発火します(旧 /flow/webhook/… は410)。

Script Filter例

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

例3: アラーム → 緊急作業指示自動発行

ERROR等級以上のアラームを受けると上位資産を対象に緊急整備ワークオーダーを発行します。

[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: 정비 담당자]

Script Filter例

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

例4: 外部DB同期 — 資産一括登録

レガシーDBから新規設備行をポーリングし資産として自動登録します。

[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]

ノード設定

ノード	核心設定
`flow_jdbc_poll`	`dsn`: 外部DB接続, `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`

例5: 作業指示自動状態遷移

資産で発生する稼働/停止イベントで作業指示状態を自動遷移させます。

[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]

Switchノードケース例

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

状態遷移が失敗した場合(不正な現在状態)は自動で FAILURE にルーティングされるので、別途フィルター無しに安全に結線できます。

例6: アラームバンド自動調整

資産集計(例: 6時間平均値)を基準にタグの上限/下限アラームバンドを動的に調整します。

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

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

例7: エッジデバイス自動登録

新しいOPCサーバー情報を受けるとエッジデバイスに一括登録します。

[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: 엣지 디바이스 가동]

呼び出し本文例

{
  "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"  }
    ]
  }
}

例8: 外部API信頼性補強 (Retry)

間欠的に失敗する外部API呼び出しにバックオフ再試行を適用し、最大試行を超えたら運用者に通知します。

[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ノード設定

max_attempts: 5
backoff_ms: 2000
backoff_multiplier: 2.0 → 2秒、4秒、8秒、16秒、32秒間隔

例9: プラグインイベントルーティング (OEE低下通知)

OEE値が閾値以下に落ちるとラインマネージャーにプッシュ通知を送ります。

[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: 라인 매니저]

Template例

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

例10: メッセージ精製パイプライン (Throttle + Debounce)

高頻度タグ変化を1分に1回に制限し、さらに5秒間変化がない場合のみ後続ノードを発火します。

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

例11: 多重トリガー合算 (Merge)

OEE/RAM/EMS 3種のイベントを30秒単位でまとめ、一度のレポートメッセージで送信します。

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

同じフロー内にトリガーノードを複数置くとすべて進入点になります。flow_merge の data.merged 配列にウィンドウ期間に入ったすべてのメッセージが累積されます。

例12: サブフローで共通処理をまとめる

共通メッセージ精製ロジック(重複検査+単位変換+積載)を別のフローに分離し、複数のトリガーから呼び出します。

メインフロー (それぞれ独立)

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

サブフロー FLOW_00099

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

サブフローはトリガーがなくても保存でき、外部呼び出し専用として使用できます(保存時に警告が表示されます)。

例13: シフト変更時の日次レポート自動送信

毎日の夜間シフト終了時刻(例: 06:00)に昨日〜今日にわたる生産・品質・ダウンタイムの要約をメールで送信します。

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

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

例14: ダウンタイム自動集計

資産STOPイベントが入るとシフト別の累積ダウンタイムを更新し、閾値超過時に通知を送ります。

[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분 초과']

例15: 品質不良ライン自動隔離

品質点検で連続5件以上の不良が報告されるとライン資産に停止命令を発行し作業指示を中断します。

[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: 품질 매니저 + 라인 매니저]

例16: エネルギー閾値超過 — ライン一時停止勧告

ラインの時間当たりエネルギー消費が予算を超えると運用者にSMSと共に勧告メッセージを送ります。

[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: 자산에 마지막 경고 시각 기록]

例17: 外部システム双方向同期 — 作業指示状態ミラーリング

外部ERPから入る状態と内部作業指示状態を双方向に同期します。

下方向 (ERP → 内部)

[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]

上方向 (内部 → 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]

双方向同期時に無限ループを避けるには、メッセージの出所を metadata.source のようなキーで表記し、トリガー段階で自己発行メッセージをフィルタリングしてください。

インポート/エクスポート

フロー定義をJSONで直列化し、他の環境へ移植したりバックアップできます。

動作	位置	説明
エクスポート	編集画面上部の `내보내기` ボタン	グラフ+ノード+ワイヤー+ノード設定全体をJSONでダウンロード
インポート	一覧画面上部の `가져오기` ボタン	JSONテキストをペーストまたはアップロード
自動スナップショット	保存時自動	グラフ保存時にバージョン単位で積載(ロールバック対応)

インポート動作

常に新しいフローIDが発行されます(既存ID上書き防止)

ノードIDも新しく付与され、ワイヤーリンクが自動で再マッピングされます

import直後のフローは解除状態で積載され、運用者が検討後直接配備する必要があります

エラーハンドリング

ノード単位

ノード処理中に例外が発生すると自動で FAILURE relationへルーティングされます。
FAILURE 出力に接続されたノードが無ければ、メッセージはdropされエラーログのみ残ります。
トリガーノードの *_pattern 未マッチメッセージは SKIPPED 処理され後続ノードへ渡されず、処理/エラーカウンターにも含まれません。
すべてのエラーはライブデバッグパネルと実行履歴画面に表示されます。

フロー単位

項目	デフォルト値	説明
max_depth	100	1メッセージ処理中の累積ノード訪問数制限
max_revisit	3	同じノードの再訪問回数制限(サイクル無限ループ防止)
flow_timeout_ms	30,000	処理時間超過時強制終了

外部IOノードの再試行

HTTP・Kafka・外部DB等の外部IOノードは retry_count / retry_delay_ms オプションでノード内部で即時再試行が可能で、すべての再試行が失敗すると FAILURE relationへルーティングされます。より精巧なバックオフやEXHAUSTED分岐処理が必要な場合は flow_retry ノードを別途使用してください。

運用診断

管理者はシステムメニューでフローエンジンのディスパッチキュー状態(ワーカー稼働有無、待機キューサイズ、累積処理/失敗/ドロップ件数、最後のエラー)を確認できます。

サーバーはバックグラウンドでフローワーカープールの負荷を周期的に点検し、負荷が一定時間以上持続すると運用ログに一行で状態変化(HEALTHY → DEGRADED → CRITICAL)を記録します。正常に回復すると回復ログがもう一行残ります。

権限

フローは別途の権限モデルを導入せず、既存のシステム認証/権限をそのまま使用します。

機能	必要な権限
一覧照会 / 実行履歴照会	すべての認証済みユーザー
フロー作成・編集・配備・削除	ADMIN
インポート/エクスポート / すべて再配置	ADMIN

運用パターン (Recipes)

よく使う結線形態を集めたライブラリです。各パターンはそのまま複製して新しいフローの出発点として活用できます。

パターン1 — 処理 + 通知分岐

処理結果によりSUCCESSは積載、FAILUREは通知に二重分岐。

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

パターン2 — 安全な再試行

外部IOが一時的失敗に強靭になるよう、バックオフ+EXHAUSTEDハンドラーを構成。

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

パターン3 — 事前フィルターで負荷遮断

トリガー段階の *_pattern でメッセージを予め絞り込み後続処理量を削減。パターン未マッチはSKIPPED処理されカウンターに含まれません。

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

パターン4 — 多重case分岐

状態/タイプにより複数の枝に分岐。

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

パターン5 — ウィンドウ累積 + 一度にemit

高頻度入力を一定時間まとめて一つのメッセージに変換。

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

パターン6 — リトライ + 再試行上限後迂回

再試行が終わればバックアップ経路(他のAPI、通知、DB積載)に迂回。

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

パターン7 — ドメイン変更後の後続アクション

タグ単位のアラームを上位資産単位に変換し資産別処理に委任。

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

パターン8 — スロットル + デバウンス結合

分当たり1回以下にスロットル + 5秒間変化がない場合のみ処理。

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

パターン9 — サブフローで共通ロジックをモジュール化

複数の進入点が同じ後続処理(検証・精製・積載)を共有すべき時、サブフローに分離。

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

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

パターン10 — 直接トリガー迂回

他のフローの結果を次のフローの入力として流すには flow_dds_publish でドメインチャンネルに発行し、他のフローは同一メッセージタイプをトリガーとして受け取る。

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

活用例一覧

シナリオ	構成
MES連携	`flow_schedule` → `flow_http_request` → `flow_script_transform` → `flow_create_work_order`
イベント外部中継	`flow_on_asset_event` → `flow_msg_type_filter` → `flow_mqtt_publish`
データ精製・積載	`flow_on_tag_point` → `flow_script_transform` → `flow_save_tag_point`
アラーム自動化	`flow_on_tag_alarm` → `flow_script_filter` → `flow_send_email` + `flow_create_work_order`
外部DB同期	`flow_jdbc_poll` → `flow_script_transform` → `flow_create_asset`
Webhook受信	`flow_on_webhook` → `flow_script_filter` → `flow_publish_asset_command`
アラームバンド自動調整	`flow_on_asset_aggregation` → `flow_script_transform` → `flow_update_tag_alarm_band_numeric`
作業指示状態自動化	`flow_on_asset_event` → `flow_switch` → `flow_start/end/pause/resume_work_order`
API信頼性補強	`flow_http_request` ─FAILURE→ `flow_retry` ─SUCCESS→ループ / EXHAUSTED→通知
OEE低下通知	`flow_on_oee_event` → `flow_script_filter` → `flow_template` → `flow_send_push`
エッジ一括登録	`flow_on_webhook` → `flow_edge_opc_create` → `flow_split` → `flow_edge_tag_create` → `flow_edge_opc_start`
高頻度精製	`flow_throttle` → `flow_debounce` → 後続
多重イベント合算	多重トリガー → `flow_merge` → 要約変換 → 送信

段階別デバッグガイド

フローが意図どおりに動作しない時に従うべき標準手順です。

ステップ1 — 発火有無の確認

一覧画面で該当フローの実行件数が増加しているか確認します。

観察	意味・次の対応
カウントが0	トリガーが発火していない。フロー配備状態+トリガーノード結線+`*_pattern` オプションを点検
カウントは増加するがエラーも一緒に増加	アクションノードで失敗。ステップ2へ
カウントは増加するが後続処理が実行されない	分岐結線漏れ。`FAILURE`/`THROTTLED`/`EXHAUSTED` 等すべての出力処理を確認

ステップ2 — ライブデバッグでノード別の流れを確認

編集画面を開いてライブデバッグパネルをオンにします(2秒更新)。

観察	意味
特定ノードの点灯がグレーのまま	メッセージが到達していない — 直前ノードで `FAILURE` 分岐またはフィルターで遮断
点灯が赤+ERROR行	ノード処理中の例外。メッセージの `data.error` フィールドと `/flow/log` の `NODE_ERROR` イベントを確認
点灯緑だが次のノードがグレー	出力relationラベル不一致。ワイヤーラベルがノードの出力relation(SUCCESS/TRUE等)と正確に一致するか確認

ステップ3 — 実行履歴画面でメッセージを追跡

/flow/log 画面でメッセージID基準で単一メッセージの全体の流れを追跡します。

メッセージIDフィルターにライブデバッグで見た msg_id を入力
FLOW_START → NODE_IN → NODE_OUT → FLOW_END の順に時系列でソート
NODE_ERROR が見えたらそのノードの error_message が原因

ステップ4 — ノード設定の点検

よくある間違い:

間違い	点検
`*_field` パスのタイプミス	`data.tag_id` で合っているか?`metadata.tag_id` ではないか?ライブデバッグのメッセージプレビューで実際のキーを確認
`${...}` テンプレート変数未置換	変数パスがメッセージに存在するか、タイプミスがないか
外部IOタイムアウト	`timeout_ms` を増やす。外部システム自体の応答時間
権限・認証ヘッダー	`headers` のJSON形式が正確か、トークンが有効か

ステップ5 — 分離テスト

問題のあるノードを新しい臨時フローに単独で移して、テスト実行で1つのメッセージだけ発火させて結果を分離検証します。正常動作すれば元のフローの結線・前段階のメッセージ形態が疑われます。

ステップ6 — カウンター初期化後に再現

全体カウント初期化後に1つのメッセージだけ発火させ、きれいな統計で再現すると問題がより明確になります。

よくある問題

症状	原因・対応
トリガーが発火しない	フローが配備状態か、トリガーノードの `_pattern` が狭すぎないか確認。スケジュール/外部購読ノードはすべて再配置*後に再試行
ライブデバッグが空	メッセージカウンターが増加していないか確認 — トリガーパターン未マッチ(`SKIPPED`)はカウンターに含まれません。パネル上部の一時停止トグルがオンになっていないかも確認
カウンター数値が異常に累積する	ノード設定右上の全体カウント初期化ボタンで統計ウィンドウをリセット
同じメッセージが繰り返し処理される	サイクルを疑ってください。`max_revisit` デフォルト3回以内のみ同一ノードの再訪問が許可されます。`flow_subflow` で分離した後 `flow_throttle`/`flow_debounce` で入力量を調整
外部システム呼び出しが間欠的に失敗	`retry_count`/`retry_delay_ms` ノードオプションまたは `flow_retry` ノードでバックオフ再試行を結線
インポート後にスケジュールが動作しない	一覧上部のすべて再配置を実行
Updateノード実行後一部フィールドがそのまま	意図された動作です。Updateノードは入力したフィールドのみ更新し残りは保持します。全体更新はDelete + Create組み合わせ
Createノードがnot null エラーで失敗	ノード設定で `*_field` 動的オプションパスがメッセージに実際に存在するか確認。一部NOT NULLカラムは自動補強されますがコアID(`asset_id`, `tag_id`等)は直接入力する必要があります
`flow_kafka_publish` / `flow_mqtt_publish` が発行されない	外部ブローカー接続情報(`bootstrap_servers`/`broker_url`)とトピック権限を確認。`flow_log` を横に結線して発行直前のメッセージが到達するか確認
`flow_http_request` が応答が来ない	`timeout_ms`(デフォルト5000)を増やし、本文は `body_template` に `${msg.data.x}` 形式で直接明示。応答は `data.response_status` / `data.response` に積載される
`flow_retry` が再試行されない	結線が誤っているケースです。`flow_retry` の `SUCCESS` 出力を元の失敗ノードに再びループ結線する必要があります(下記パターン2参照)
`flow_jdbc_poll` のような行が毎回また入る	ポーリングSQLの `WHERE` 条件に処理後の状態更新を含める必要があります(例: `WHERE sync_status='NEW'` + 同一フローの終わりで `flow_jdbc_query` で `UPDATE ... sync_status='OK'`)
アラーム直接トリガーノードが見つからない	意図された除外です。アラームは `flow_create_alarm_config` 経路でのみ発生すべきでアラーム履歴の整合性が維持されます
スクリプトノードが `Java class` アクセスエラーで失敗	スクリプトサンドボックスで遮断されます。外部呼び出しは別途外部連携ノードを結線してください
作業指示状態遷移ノードが `FAILURE` のみ出る	現在の状態が遷移可能な開始状態ではありません。例: `flow_pause_work_order` は `START` 状態のみで動作します。事前に `flow_check_existence_field` / `flow_script_filter` で状態を確認
テスト実行が動作はしたがグラフが変更された状態	テスト実行は常に保存済みバージョンを発火します。変更内容を検証するには先に保存 → テスト実行
インポートしたフローが非活性状態	意図された動作です。検討後直接配備トグルをオンにしてください

よくある質問 (FAQ)

運用者が初めてフローを使用する際によく聞かれる質問を集めました。

Q. フロー1つにトリガーを複数置けますか? A. はい。同じフロー内にトリガーノードを複数置くとすべて進入点になり、それぞれ独立して発火します。flow_merge と組み合わせると、複数種類のイベントを一つの後続処理に統合できます。

Q. トリガーが無いフローも作成できますか? A. はい。保存時に警告が表示されますが保存はできます。テスト実行または他のフローの flow_subflow 呼び出しのみで発火させる「ライブラリ」形態として使用できます。

Q. 一つのメッセージが複数のノードを同時に通過するようにできますか? A. はい。一つのノードの出力ポートに複数のワイヤーを接続すると、同じメッセージが同時に分岐しすべての後続ノードへ渡されます。

Q. 変換ノードでメッセージ自体を作って別のメッセージに変えることができますか? A. はい。flow_script_transform で msg.type, msg.originator, msg.data, msg.metadata すべて自由に変更できます。ただしメッセージ自体を別のトリガータイプのように見せても、他のフローを自動で呼び出すわけではありません(元のトリガーマッピングを維持)。他のフローを呼び出すには flow_subflow または flow_dds_publish を使用してください。

Q. 自動スナップショットは何個まで保管されますか?直接復元できますか? A. グラフ保存ごとに自動で積載され、保存ポリシーは運用環境設定に従います。画面から直接復元するUIは提供されず、必要な場合は管理者に依頼して特定バージョンに戻すことができます。最も安全な方法は変更前に 내보내기 でJSONファイルを外部に保管することです。

Q. 実行履歴はどれくらい保管されますか? A. デフォルト7日です。長期保管が必要な場合は flow_kafka_publish または flow_jdbc_query で外部ログシステムに積載してください。

Q. 一つのフローの統計だけ別途リセットしたいです。 A. 編集画面のノード設定右上のエラーカウント初期化(エラーのみ)または全体カウント初期化(全体)を使用してください。他のフローには影響しません。

Q. 別の環境(開発/ステージング/運用)にフローを移したいです。 A. 編集画面の 내보내기 でJSONをダウンロードし、対象環境の一覧画面の 가져오기 でアップロードすればよいです。インポートしたフローは常に新しいID + 解除状態で積載されるので、運用者が検討後に直接配備できます。

Q. 同じペイロードが2回処理されることがあります。どう防ぎますか? A. トリガー段階で *_pattern を絞るか、flow_throttle/flow_debounce で頻度を制限してください。外部入力(Webhook・MQTT)なら発信側に冪等キーを設け、flow_check_existence_field またはメッセージID基準のフィルターで重複を除去できます。

Q. サイクル(ループ)結線は安全ですか? A. flow_retry のような意図されたサイクルは安全です。それ以外のサイクルは max_revisit(デフォルト3回)に引っかかり自動で遮断されます。ただし運用中には flow_log で流れを可視化し、意図しないサイクルを早期に発見することをお勧めします。

Q. フロー変更はリアルタイムに反映されますか? A. グラフ保存時に即座に反映されます。ただしスケジュール・外部購読・外部DBポーリングのような自前のスケジューラーを持つトリガーはすべて再配置またはフロー解除→配備トグルで再登録してください。

Q. 既に登録されたノードのIDを変更できますか? A. ノードIDはシステムが自動付与します。表示名はノード設定フォームで変更でき、ライブデバッグ・実行履歴に表示名が使用されます。

Q. 権限が無いユーザーがフローを誤って変更できますか? A. フロー作成・編集・配備・削除にはADMIN権限が必要です。一般ユーザーは一覧照会と実行履歴照会のみ可能です。

運用ベストプラクティス

フローをプロダクションで安全に運用する際に守っていただきたい原則です。

負荷管理

トリガー段階で *_pattern により事前フィルタリングして後続処理量を減らしてください。毎分数万件入るタグポイントは最もコストが大きい進入点です。
外部システム呼び出し(flow_http_request/flow_jdbc_query等)は flow_throttle / flow_debounce と組み合わせて外部負荷を制御してください。
同じ入力に複数の後続ノードが接続される必要がある場合は flow_subflow で分離してグラフ全体のノード数を削減してください。ノードが多いほどライブデバッグのポーリングコストも増加します。
デバッグモードは運用検証段階のみオンにしてください。積載量が多いほど実行履歴保存7日ウィンドウが早く一杯になります。

安全な作成

すべてのアクションノードには必ず FAILURE 分岐を結線しておいてください(最低 flow_log でも)。FAILURE が空だとメッセージが静かにdropされ事故追跡が困難になります。
外部IOノードはできる限り flow_retry と組み合わせて一時的障害を吸収してください。
Updateノードは入力したフィールドのみ更新します(部分更新)。全体上書きはDelete + Create組み合わせを使用してください。
トリガーノードが無いグラフは手動テスト実行専用です。誤ってトリガーを漏らさないよう保存時の警告を確認してください。
サイクル(同じノードに戻る結線)は必ず flow_retry 等の意図された形態のみで作成し、それ以外の場合は max_revisit が保護しますが、疑わしいグラフは flow_log で流れを可視化してください。

変更手順 (Change Management)

運用中のフローを変更する際の推奨順序です。

バックアップ — 編集画面の 내보내기 で現在のグラフをJSONファイルとしてダウンロードして保管します(自動スナップショットも積載されますが外部保管が安全)。
複製または新しいフローで作業 — 運用中のフローを即座に修正するより、新しいフロー(またはインポートで作った複製)で変更後に検証してください。
テスト実行 — JSONメッセージを直接注入してすべての分岐(SUCCESS/FAILURE/EXHAUSTED等)を一度ずつ発火させ、実行履歴で結果を確認します。
運用反映 — 検証されたフローのグラフJSONを 내보내기 → 運用環境で 가져오기 → 運用者が検討後配備トグル。
問題発生時ロールバック — 即座に解除トグルで無効化。自動スナップショットが積載されているので開発者に依頼すれば以前のバージョンに戻せます。

カウンター・統計の活用

一覧画面の処理トレンドスパークラインが急に平坦になったら、トリガー未発火またはSKIPPED処理の可能性を確認してください。
エラー率のドーナツチャートが異常であれば、ノード設定の *_field パスを疑ってみてください。メッセージペイロードにキーが無いと頻繁に失敗します。
運用検証後には全体カウント初期化で統計ウィンドウをリセットし、平常時のベースラインを再測定してください。

緊急対応手順

運用中に問題が発生した際に迅速に使用するステップバイステップの手順です。

シナリオ1 — 特定のフローが暴走(メッセージ暴増)

症状: 一つのフローの実行件数が正常値に対し数十〜数百倍急増、エラーも一緒に増加

対応

即座に該当フローの解除トグルで無効化 (一覧画面で1秒)
ライブデバッグパネル・実行履歴でどのトリガーが暴走を引き起こしたか確認
トリガーノードの *_pattern オプションを狭めるか、直後に flow_throttle / flow_debounce を結線
必要に応じて flow_check_existence_field または flow_msg_type_filter でメッセージタイプを限定
修正後テスト実行で再現→正常確認後配備再開

シナリオ2 — 外部システム障害による一括失敗

症状: HTTP/Kafka/外部DB等の外部IOノードのエラーが同時に累積

対応

影響範囲が広い場合は関連フローを一括解除 (一覧画面で一括選択後一括解除)
外部システムの復旧確認
flow_retry 結線が無い外部IOノードなら結線追加
外部システムの応答が遅くなったら timeout_ms を調整
運用環境に応じてすべて再配置後配備再開

シナリオ3 — サイクルによる無限ループ

症状: 単一メッセージが同じノードを繰り返し通過、処理時間が累積

対応

max_revisit 保護で最大3回まで再訪問され自動で遮断されますが、運用安全のため解除後点検
グラフでサイクルを視覚的に追跡 (編集画面でワイヤーを辿る)
意図されたループ(flow_retry)なら max_attempts が適正か確認
意図されていないサイクルなら結線を削除するか flow_check_relation で分岐を追加
修正後全体カウント初期化 → 再配備

シナリオ4 — データ損傷疑い(誤った自動更新)

症状: オートメーションでドメインデータが意図と異なる更新をされる

対応

即座に該当フローを解除
外部に保管中の直前のグラフJSONバックアップまたは自動スナップショットから以前のバージョンを確認(管理者協力)
グラフ分析: 意図しないUpdateノード結線・誤った *_field パス・スクリプト変換エラーを確認
影響を受けたドメインデータは別途手順でバックオフィス画面から修正
修正したグラフをテスト実行で検証後再配備

シナリオ5 — システム点検・保守中の停止

症状: 外部システムの点検時間中に外部IO呼び出しを一時停止したい

対応

影響を受けるフローを一括解除 (一覧から複数選択)
点検終了後配備再開
自前スケジューラー保有トリガー(スケジュール・外部MQTT購読・外部DBポーリング)はすべて再配置を1回実行して正常登録を確認

シナリオ6 — ライブデバッグ積載キューがいっぱい

症状: 運用診断ページの積載キューサイズが閾値近く、ドロップ件数が増加

対応

デバッグモードがオンのフロー数と積載量を減らしてください(検証済みのフローはデバッグモードOFF)
トリガー段階の *_pattern で後続処理量自体を減らしてください
システムが自動回復モードに入ると負荷ウォッチドッグに DEGRADED/CRITICAL ログが一行残ります — 管理者に共有してください

すべてのシナリオで優先順位は即座に解除 → 原因把握 → 修正 → 検証 → 再配備の順序です。変更手順(変更安全手順)も併せて参照してください。

セキュリティ・機密情報の取り扱い

フローは外部API呼び出し・メール/SMS送信・Webhook受信等の機密性のある入出力を扱うので、次の原則を守ってください。

トークン・認証情報

APIトークン・パスワードをノード設定に平文で直接入力しないでください。${ENV_VAR} 形式の環境変数置換を使用し、運用環境の秘密保管庫から注入してください。
運用環境でトークンを露出させないよう headers JSON のトークン位置をできる限り環境変数に分離します。

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

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

Webhook進入点の保護

flow_on_webhook は認証されたユーザーが呼び出す内部進入点ですが、外部に露出する際はペイロード検証フィルターを必ず結線してください。

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

機密データのマスキング

ライブデバッグパネルと実行履歴にはメッセージプレビューが表示されます。個人情報(名前・連絡先・アカウント)とトークンは積載直前にマスキングしてください。

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

外部IOの応答本文処理

flow_http_request の data.response は応答本文全体が積載されます。応答に機密データが含まれる場合は即座に後続の変換ノードで必要なキーだけを抽出し原本を除去してください。

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

スクリプトノードの分離

スクリプトノードは安全なサンドボックスで実行され、ファイル・ネットワーク・任意クラスアクセスが遮断されます。外部呼び出しが必要な場合は必ず別途外部連携ノードを結線してください。

フローメトリクスとアラーム

フローが運用中に自動で収集される指標と、その指標を見る場所・アラームを設定する方法。

フロー単位KPI

一覧画面(/flow/index)の各行に次のKPIが表示されます。

指標	意味	異常信号
全体実行件数	トリガーが発火してグラフを一度通過した件数(成功/失敗合計)	平常時比急減→トリガーが死んでいる / 急増→暴走
エラー件数	グラフ中のどのノードでも `FAILURE` 分岐に落ちた件数	全体の5%以上なら点検
エラー率	`에러 / 전체 × 100`	一定閾値超過時アラームを設定してください
最終実行	最も直近の発火時刻	「5分以上発火無し」が正常な場合のみOK
平均処理時間	メッセージ1件がグラフ全体を通過する平均ms	外部IOノード追加/削除時に大きく変化
最近の処理時間推移	5分スパークライン — ライブデバッグパネル	spike発生時は外部システムの応答を点検

ノード単位KPI

編集画面でノードをクリックすると、ノード右下に以下が表示されます。

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

位置	表示
上部点灯	グレー(待機) / 緑(処理中) / 赤(エラー)
下部ラベル	`처리 N · 에러 M · 평균 Xms · 최근 Yms`

ノード統計4種

カウンター	意味
処理	そのノードに入り正常分岐へ出たメッセージ数
エラー	`FAILURE` 分岐または例外発生数
平均処理時間	ノード自体の処理ms(外部IO含む)
最近の処理時間	最後の1件の処理ms

システム単位メトリクス

フローエンジン全体のメトリクスはシステム → モニタリング画面で確認できます。

メトリクスはJMXドメイン plantpulse.core.engine 配下に次の5個として公開されます。

メトリクス	種類	意味	見方
`FLOW_EXECUTOR_QUEUE_SIZE`	Gauge	フロー実行プールの待機キューサイズ	増え続けると処理が入力に追いついていない
`FLOW_EXECUTOR_ACTIVE_COUNT`	Gauge	実行プールのアクティブスレッド数	プールサイズに付いていると飽和
`FLOW_EXECUTOR_DROPPED_COUNT`	Gauge (累積)	プール飽和により捨てられたtaskの累積数	0以外ならメッセージが失われている — 最初に見るべき値
`FLOW_DEBUG_QUEUE_SIZE`	Gauge	デバッグイベント積載待機キューサイズ	デバッグモードをオンにしたフローが多いと大きくなります
`FLOW_EXECUTION_TIME`	Timer	フローごとの処理時間分布	平均・最大追跡

flow.engine.* という名前のメトリクスはありません

以前の文書が flow.engine.queue_depth · in_flight · dispatch_lag_ms · exec_p95_ms failed_per_min · script_timeout_per_min を表に載せていましたが、**そのような名前のメトリクスは存在しません。**上記5つが全てです。

(flow.engine.enabled はメトリクスではなく engine.properties の設定キーです — フローエンジン全体を停止するマスターゲートです。)

メトリクスベースのアラーム登録パターン

フロー自体のメトリクスを高度なアラームまたはCEP → トリガーで監視するパターン4種。

パターンA — エラー率閾値超過

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

パターンB — 処理キューバックプレッシャー発生

flow.engine.queue_depth > 500 が1分以上持続する時アラーム — トリガー発火速度が処理速度を超える状況。

パターンC — 特定フローの発火途絶

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) )
]

正常に分当たりN件発火していたフローが15分以上沈黙するとアラーム。

パターンD — 外部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

アラーム出力チャンネルの推奨

アラーム種類	推奨チャンネル
エラー率・発火途絶 (運用直接)	メール+プッシュ通知
キューバックプレッシャー・タイムアウト暴増 (システム)	Slack/Teams webhook
自動診断 (参考)	診断ログのみ

⚠️ アラーム用フロー自体には絶対にアラームを設定するメトリクスに依存しないでください — アラームフローが死ぬとアラーム自体が来なくなります。アラーム用フローはシステム → モニタリングの外部ヘルスチェックで監視してください。

メッセージ処理の意味論とバックプレッシャー

フローエンジンがメッセージを処理する際の保証レベルとバックプレッシャー動作。

配信保証 — At-Least-Once

フローエンジンはat-least-once配信を保証します。

ケース	動作
正常処理	一度発火→一度グラフ通過→一度完了
処理中のエンジン再起動	トリガーキューに残っていたメッセージは次の起動後に再処理される
ノード単位の例外	`FAILURE` 分岐にのみ遷移。メッセージは消えない
外部IOタイムアウト	`FAILURE` 分岐 + `flow_retry` で自動再試行可能

重複の可能性 — 正確に一度(exactly-once)ではありません。flow_create_* が途中で切れ再試行されると同じIDが2度入る可能性があるので、on_duplicate=skip または外部システムの冪等性キーを使用してください。

冪等性キーパターン

フローで外部システムを呼び出す際に冪等性を保証するパターン。

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

以降外部呼び出し時にヘッダーとして添付:

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

処理順序 — トリガー単位FIFO

同じoriginatorのメッセージ	異なるoriginator
トリガー単位FIFO(到着順)	並行処理(順序無関係)

資産Aのイベント2件がほぼ同時に発生しても、Aのイベントは発生順に処理されます。資産Aと資産Bのイベントは別々のワーカーで並行処理される可能性があります。

バックプレッシャー

処理速度が発火速度に追いつかない場合の動作。

状況	動作
キュー深さ < 80%	正常 — 新規トリガー即時enqueue
キュー深さ80%~100%	診断WARNログ発生 — キューは継続受容
キュー深さ=100% (一杯)	新規トリガーをdrop + 診断ERRORログ発生

キューが一杯の時運用者が行うべきこと

システム → モニタリングで flow.engine.queue_depth 確認
一覧画面で実行件数が急増したフローを識別
そのフローのトリガー事前フィルター(*_pattern)を狭めて負荷減少
またはそのフローの配備を一時解除して非常処理

サーキットブレーカー (外部IO)

外部IOノード(HTTP/Kafka/MQTT/メール)は次の条件が満たされると30秒間一括遮断されます。

条件	閾値
最近1分内の連続失敗	≥ 10件
平均応答時間	≥ 10秒

遮断中はそのノードに入るすべてのメッセージが即座に FAILURE へ分岐します。外部システムが応答を回復すると自動で遮断解除。この動作は外部障害がフローエンジン自体を麻痺させないよう隔離する役割です。

サーキットが開いた時点は診断ログに code = CIRCUIT_OPENED として記録されます。外部システムが早く回復したら実行履歴の metadata.retry_count を見て漏れたメッセージを手動で再実行できます。

サイクル(循環)防止

フロー内に次のようなサイクルができると無限ループが発生する可能性があります。

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

フローエンジンは2つの防御線でサイクルを遮断します。

防御線	動作
`max_depth=100`	ノード100個を経ると強制終了
`max_revisit=3`	同じノードを4回目に訪問する瞬間メッセージ廃棄+診断ERROR

flow_retry のSUCCESS分岐→元のノードループは意図的なサイクルであり、metadata.retry_count が一緒に増加するのでmax_revisitが発動する前に正常終了します。

実行履歴で何を確認できるか

以前の文書の「エラーコードカタログ」は実在しませんでした

NODE_SCRIPT_TIMEOUT · FLOW_TIMEOUT · FLOW_MAX_DEPTH · TRIGGER_QUEUE_FULL WEBHOOK_AUTH_FAIL · DOMAIN_DUP_KEY のようなコードが表に載っていましたが、**製品のどこにもその文字列は生成されません。**ログでそのコードを検索すると永遠に0件です。

フローエンジンが実際に残すのは以下のイベントタイプ5種です。

フロー実行履歴(mm_flow_log, 左メニューAutomation > フロー実行履歴)に積載されるイベントは5種類です。

イベントタイプ	いつ	積載条件
`FLOW_START`	フロー進入時	常に
`FLOW_END`	フロー処理終了(成功・タイムアウトともに)	常に
`NODE_IN`	ノードがメッセージを受け取る時	デバッグモード時のみ
`NODE_OUT`	ノードがメッセージを送り出す時	デバッグモード時のみ
`NODE_ERROR`	ノード処理中の例外	常に

各行が共通で持つフィールドです。

フィールド	内容
`level`	`INFO` / `WARN` / `ERROR`
`flow_id` · `flow_node_id` · `node_name` · `node_type`	どのフローのどのノードか
`msg_id`	メッセージ単位追跡用 — 一つのメッセージの流れを繋いで見るにはこの値でグルーピングします
`message` · `error_message`	人が読める説明、例外メッセージ
`duration_ns`	ノード実行時間(ナノ秒) — `NODE_OUT` · `NODE_ERROR` でのみ意味があります

普段はノード単位のログが残りません

NODE_IN · NODE_OUT はデバッグモードでのみ積載されます。普段の実行履歴に FLOW_START · FLOW_END のみ見えるのは正常です。ノードを一つずつ追って見るにはそのフローのデバッグモードをオンにしてください — 代わりにログ量が大幅に増えます。

症状から探す

コードの代わりに症状から出発します。

症状	最初に見る場所
トリガーは発火するがどのノードも動いていないようだ	フローが配備(トグルON)されているか・トリガーの `*_pattern` がメッセージを絞り込んでいないか
処理が途中で切れる	1メッセージ30秒(`flow_timeout_ms`)超過。サーバーログの `FlowExecutor timeout` 警告で確認します
同じメッセージが繰り返し処理される	サイクル。`max_revisit`(3回)で自動遮断されますが結線を点検してください
グラフが長すぎて終わらない	`max_depth`(100)超過。サブフローに分割します
ノードが例外を投げる	実行履歴の `NODE_ERROR` 行の `error_message` を見ます
外部呼び出しが失敗する	該当ノードの `data.response_status` · `data.error` を後続ノードで確認します

限度値(100・3・30,000ms)は ExecutionState のデフォルト値です。

クラスター・高可用性(HA)動作

プラットフォームがクラスター環境にインストールされている場合、フローエンジンは次のルールで分散動作します。

ノードの役割分離

役割	動作
リーダー(Leader)	フローグラフの変更(保存/配備/解除)を直列処理する単一ノード
ワーカー(Worker)	トリガー発火・メッセージ処理を並行実行する一般ノード(すべてのノードがワーカー役割を兼務)
スケジューラー(Scheduler)	`flow_schedule` cron評価担当 — リーダーと同一ノード

リーダーノードはプラットフォーム起動時に自動選出され、リーダーがダウンすると他のノードのうち一つが自動でリーダーに昇格します。運用者が直接指定する必要はありません。

メッセージ分配 — 資産単位の一貫ルーティング

分配キー	動作
`originator.id` (資産/タグ/注文ID)	同じ資産のメッセージは常に同じワーカーへルーティング — シーケンス保証
外部進入(`flow_on_webhook`等)	ラウンドロビン分配

このルールで資産Aのイベントが複数のワーカーで同時処理され順序が崩れることが防止されます。結果として資産単位では単一ワーカー処理保証、資産間では並行処理が同時に成立します。

リーダー障害時の動作 — フェイルオーバー

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

段階	影響
0〜3秒	`flow_schedule` 発火一時停止 / トリガー処理は影響なし
3〜5秒	新しいリーダー確定、スケジュール再開
5秒以降	正常

cron発火は**misfire ポリシーに従い一度漏れた発火を即座に再実行**します。運用中には flow_schedule の評価単位が5秒未満だと短期の漏れが見える可能性があります。

トリガーキューの永続性

項目	動作
トリガーキュー位置	インメモリキュー+永久保存庫(トランザクションログ)
ノード再起動時	キューに残っていた未処理メッセージは次の起動後に再ディスパッチ
処理中のノードダウン	そのメッセージは再処理されますがat-least-once保証により外部システムに冪等性キーが必要

クラスター配備時運用者チェックリスト

すべてのノードの時間同期(NTP) — トリガー発火時刻がノード間で一致
外部システム(MQTT/Kafkaブローカー)をすべてのノードが到達できるネットワーク位置に配置
flow_send_email のSMTP設定はシステム設定に一度だけ登録 — すべてのノードが共有
ワーカープールサイズ(flow.engine.workers)はノード別CPUコア数に合わせて調整

単一ノードモード (開発・小規模)

リーダー・ワーカー・スケジューラーすべて1プロセスで動作
フェイルオーバー無し — ノードがダウンするとフロー自体が中断
トリガーキューはインメモリ+ディスクで保証され再起動後に復旧

エンドツーエンドトレース

一つのメッセージがグラフ全体をどのように通過したかを事後追跡する方法。

自動相関IDの付与

フローエンジンはすべてのトリガー発火メッセージに**相関ID(correlation ID)**を自動で付与します。

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

フィールド	意味
`metadata.trace_id`	一つのトリガー発火全体に固有のID — グラフのすべてのノードを通過するメッセージが共有
`metadata.span_id`	ノード別固有ID — ノードを経る度に更新
`metadata.parent_id`	直前ノードの `span_id`

実行履歴画面でtrace_idで検索

実行履歴画面で trace_id 入力欄にIDをペーストすると、そのメッセージが経たすべてのノードの時系列ログが表示されます。

🔍 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伝播

flow_subflow ノードで他のフローを呼び出すと同じ trace_id がそのまま繋がります。つまりメインフロー+呼び出されたサブフローのすべてのノードログが同じtraceで検索されます。

外部システムへの伝播

外部HTTP呼び出し時に自動で X-Trace-Id ヘッダーが添付されます。

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

外部システムがこのヘッダーを受信してログに一緒に記録すれば、両システムのログを同じIDでマッチングできます。

アラーム・メールでtrace_idを露出

アラーム本文またはメールテンプレートに ${metadata.trace_id} を含めると、運用者がアラームを受け取った直後に履歴画面で即座にその事件を追跡できます。

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

トレース保存期間

データ	保存期間
trace_idとノード別span ログ	7日(実行履歴と同一)
外部積載 (長期保管)	監査・履歴追跡の外部システムミラーリング使用

trace_idが長くメッセージサイズが負担になる場合は flow_script_transform で最後4桁(tr-...d4のような)だけを露出する短い形式を作ってアラーム・メールに使っても構いません。ただし検索時は全体IDが必要です。

グラフパターンカタログ

フローグラフでよく使われる結線パターン10種。

パターン1 — Pipeline (単純直列)

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

| 特徴 | メッセージがノードを順番に通過 | | 使用例 | 閾値超過アラーム→メール送信 |

パターン2 — Fan-out (1 to N)

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

| 特徴 | 一つのメッセージを複数のアクションが同時処理 | | 使用例 | アラーム発生→メール+SMS+Slack+ワークオーダー作成 |

同じメッセージの複製が複数のノードに分配されます。各分岐は独立して処理され、一方の失敗が他方に影響しません。

パターン3 — Fan-in (N to 1) — Merge

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

| 特徴 | 複数のトリガーのメッセージを時間ウィンドウ内にまとめて一度に処理 | | 使用例 | 5分間に発生したすべてのアラームを合算し日次レポート1通 |

パターン4 — Scatter-Gather (分散→収集)

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

| 特徴 | 配列を要素別に分割処理後結果を再統合 | | 使用例 | 100件の外部注文を並行検証した後結果を一括報告 |

パターン5 — Switch (条件分岐)

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

| 特徴 | 一つのメッセージを条件別に異なる経路へ | | 使用例 | アラーム優先度別処理チャンネル分離 |

パターン6 — Retry with Fallback

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

| 特徴 | 一時的障害を自動再試行、永久的障害は代替処理 | | 使用例 | 外部API失敗時3回再試行、それでも失敗すれば人に通知 |

パターン7 — Circuit Breaker (サーキット遮断活用)

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

| 特徴 | flow_throttle で呼び出し速度制限+サーキットブレーカーで一括遮断 | | 使用例 | 外部システム過負荷保護 |

パターン8 — Dead Letter Queue (DLQ)

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

| 特徴 | 永久的に失敗したメッセージを別途保管庫へ | | 使用例 | 外部ERP同期失敗メッセージ収集(手動再処理対象) |

パターン9 — Sliding Window Aggregation

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

| 特徴 | 最近N件ウィンドウ内の状態で意思決定 | | 使用例 | 最近5分アラーム30件超過時サイト非常モード |

パターン10 — Saga (多段階トランザクション)

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

| 特徴 | 複数の外部システムに渡る作業の部分失敗を補償で処理 | | 使用例 | ワークオーダー作成→資産予約→ERP同期→作業者通知(一段階失敗時前段階すべて取り消し) |

パターン選択ガイド

要件	推奨パターン
単純閾値アラーム	Pipeline (1)
一つの事件→複数チャンネル	Fan-out (2)
複数の事件→一つの要約	Fan-in / Merge (3)
配列一括処理	Scatter-Gather (4)
分岐処理	Switch (5)
外部API信頼性	Retry (6)
外部システム保護	Circuit (7)
失敗メッセージ保管	DLQ (8)
最近N件累積意思決定	Sliding (9)
複数外部システムの一貫性	Saga (10)

フローテストのベストプラクティス

フローを安全に変更・配備するためのテスト戦略。

3段階テスト — 単位→統合→シミュレーション

段階	ツール	検証対象
①単位	編集画面 ▶ テスト実行	一つのノード単独動作(スクリプト式、外部呼び出し応答)
②統合	同じ画面、臨時ペイロード+ライブデバッグON	グラフ全体のシーケンス・分岐
③シミュレーション	配備+実際のトリガー待機(ステージング環境)	実メッセージの流れ・外部システムとの結合

テスト実行ペイロードライブラリ

テスト実行ダイアログにペーストできるトリガー別ペイロード例はトリガー別ペイロード例セクション参照。運用環境で頻繁に発生する境界ケースも予め作っておいてください。

境界ケース例

ケース	ペイロード
null フィールド	`{"value": null}` — スクリプトがnull安全か
空文字列	`{"value": ""}` — 空文字列を0と解釈するか検証
負数	`{"value": -1}` — 閾値検査が±符号意図どおりか
巨大な数値	`{"value": 1e20}` — オーバーフロー/精度
Unicode	`{"name": "한글-Émoji-🚀"}` — 外部システムのエンコーディング

変更安全手順

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

回帰テストシナリオの保管

よく使われるシナリオをテキストファイルとして保管し、変更後は必ず再実行することを推奨します。

# 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)パターン

ステージング環境で外部システムの応答をモックするには:

方法	説明
flow_http_requestのurlをモックサーバーへ	運用URLをステージングURL変数に分離
flow_jdbc_pollのdatasourceを変更	運用DB→ステージングDBのみ変更
flow_send_emailのto_fieldをfake@example.comに	誤送信防止

カウンター初期化後の負荷テスト

新バージョンフローのカウント初期化(全体)
5分間正常なトリガーを発火させる
一覧画面で処理/エラーカウント、平均処理時間を確認
p95処理時間が既存対比20%以上増加すれば原因分析後ロールバック検討

パフォーマンス限界とチューニング

フローエンジンの処理限界とチューニングのヒントです。

デフォルト限界

項目	デフォルト値	備考
1メッセージのノード訪問数 (`max_depth`)	100	ノード100個を経る間に終了しなければ強制中断
同じノードの再訪問回数 (`max_revisit`)	3	サイクル無限ループ防止
フロー処理時間 (`flow_timeout_ms`)	30,000ms	1メッセージ処理に30秒超過時強制終了
外部IOノードタイムアウト	5,000ms (`timeout_ms`)	HTTP/Kafka/MQTT等
スクリプト実行タイムアウト	500ms (`timeout_ms`)	ノード単位
実行履歴保存	7日	それ以降自動失効

よく使うウィンドウサイズの推奨

ノード	推奨ウィンドウ	備考
`flow_throttle`	1,000~60,000ms	外部システムAPI限度に合わせて
`flow_debounce`	500~5,000ms	安定した値のみ通過させたい時
`flow_merge`	5,000~60,000ms	短すぎると断片化、長すぎるとlatency増加
`flow_retry` バックオフ	開始1,000ms × 2倍	5回再試行なら1・2・4・8・16秒

処理量を増やす方法

トリガー事前フィルター — *_pattern で必要なメッセージのみ進入させる(最も効果的)
サブフローでグラフを単純化 — メインのグラフは分岐・ルーティングのみ、重い処理はサブフローに
外部IOを非同期に結線 — flow_delay / flow_throttle で外部API負荷を平坦化
デバッグモードは検証段階のみ — 運用安定後デバッグモードOFF
必要なカテゴリだけ結線 — Edge・外部連携等の重いノードをすべての分岐に無理に結線しない

メッセージサイズ

メッセージの data / metadata はJSON直列化され実行履歴に積載されます。巨大なペイロード(数MB以上の応答本文等)はできる限り変換ノードで必要なキーだけを抽出しておいてください。履歴保存・デバッグ表示・サブフロー呼び出しすべてでメッセージサイズが処理コストに比例します。

監査・履歴追跡

フローの変更・実行履歴を追跡する際に確認すべき場所です。

グラフ変更履歴

自動スナップショット — グラフ保存ごとにバージョン単位で積載されます。以前の状態に戻すには管理者に依頼してください。
変更者/変更時刻 — 一覧画面の最終更新カラムが最も直近の保存時刻を表示します。変更者はシステム認証ユーザー基準で記録されます。
変更理由の記録推奨 — フローメタの説明フィールドに変更理由・担当者・関連チケット番号を一緒に記録すると追跡性が高まります。

実行イベント追跡

メッセージID追跡 — 実行履歴画面のメッセージIDフィルターで単一メッセージの全体の流れ(FLOW_START → すべての NODE_IN/NODE_OUT → FLOW_END)を時系列で照会できます。
CSVエクスポート — 実行履歴画面のCSVダウンロードボタンで現在の照会結果をエクスポートし、外部監査システムに渡すことができます。

ドメイン変更履歴

フローのアクションノード(flow_create_* / flow_update_* / flow_delete_*)が実行したドメイン変更はすべて同一ドメインサービスに委譲されるので、バックオフィス画面のドメイン別変更履歴に一緒に記録されます。作業者IDは insert_user_id="flow" 等で表記され、一般ユーザー変更と区別できます。

外部積載による長期保管

実行履歴の保存期間(7日)を超える長期保管が必要な場合は、別途フローを作って核心イベントを外部システム(Kafka・外部DB等)に積載してください。

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

新規フロー配備チェックリスト

運用環境に新しいフローを配備する前に確認する項目です。

グラフ構造

トリガーノードが正確に1つ(または意図された複数)結線されているか
すべてのアクションノードの FAILURE 出力が処理されているか(最低 flow_log)
サイクルがある場合 flow_retry 等の意図された結線か、max_revisit 保護内にあるか
外部IOノードに retry_count/retry_delay_ms または flow_retry が結線されているか

ノード設定

トリガーの *_pattern が狭すぎたり広すぎたりしないか
*_field 動的オプションパスが実際のメッセージに存在するか
CreateノードのNOT NULLフィールドがすべて埋まるか(自動補強以外)
Updateノードは部分更新を意図したものか

検証・テスト

正常ペイロードでSUCCESS分岐1回通過
異常ペイロード(必須フィールド欠落等)でFAILURE分岐通過
外部IO失敗ケースでRetry→EXHAUSTED分岐通過(該当時)
ライブデバッグパネルでINFO/ERROR表示が意図と一致
/flow/log 画面にすべての段階が記録されているか

運用安全

バックアップ(グラフJSONエクスポート)が外部に保管されているか
変更理由・担当者がフロー説明に記録されているか
通知(Email/SMS/Push)結線がある場合受信者が検証されているか
配備直後5〜10分間の実行履歴モニタリング計画があるか

配備戦略 — Canary / Blue-Green / A・Bテスト

新しいフローを安全に出荷するための3つの戦略。すべての戦略はプラットフォームの基本機能(トリガーパターン・メッセージdispatch・ライブデバッグ)のみで実装可能です。

戦略1 — Canary (一部の資産だけ段階適用)

新しいフローを一部の資産にのみ先に適用し、一定期間観察後全体に拡大。

[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進行チェックリスト

期間	モニタリング項目
1日目	エラー率<1% / 平均処理時間既存±20%以内
1週間	外部システム連携100%正常 / アラーム発生頻度適正
2週間	累積統計 / ユーザーフィードバック / 次のライン拡大決定

Canaryラインは運用中のラインの中でも影響が少ないラインを選んでください(整備頻度が高いか夜間のみ稼働するライン等)。

戦略2 — Blue-Green (旧・新同時運用後即時切替)

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

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

Blue-Greenの核心は2つのバージョンが同時に配備された状態で即時切替 — 問題発見時にv1を即座に再配備してロールバック。

v2隔離結線パターン

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

テスト用メッセージは/flow/{id}/runAPIで metadata.test_mode=true を明示して送ります。

戦略3 — A/Bテスト (性能・結果比較)

2つのバージョンが同じメッセージを受け取り異なるアクション後、結果を比較。

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

以降日次統計画面で stats_v1 vs stats_v2 累積結果比較。

A/B比較自動分析

-- 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)

戦略選択ガイド

状況	推奨戦略
新しいオートメーションシナリオ初回出荷	Canary — 一つのラインで検証後拡大
既存ロジックの大きな変更(構造改編)	Blue-Green — 即時ロールバック可能
2つのアルゴリズムのどちらが優れているか測定	A/Bテスト
単純オプション値の調整	直接変更 — `metadata.audit_diff` メモ後1週間モニタリング

ロールバック手順(共通)

問題発見時即座にロールバック:

一覧画面で新バージョン解除トグル
(Canary/A・Bの場合)トリガーパターンを0件マッチに変更
既存バージョンが単独で動作するか5分モニタリング
診断ログに code=FLOW_NOT_DEPLOYED が無いか確認
原因分析 — 実行履歴で trace_id で失敗メッセージを追跡

出荷事前チェックリスト(圧縮版)

新規フロー配備チェックリストの核心だけ一画面:

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

ノードクイック設定リファレンス

運用中によく使うノードの核心設定だけを集めたチートシートです。

トリガークイック設定

ノード	核心オプション
`flow_schedule`	`cron` (例: `0 /5 * * ?` = 5分毎)、または `interval_ms`
`flow_on_webhook`	オプション無し — 外部から `POST /webhook/flow/{flow_id}` で発火
`flow_on_mqtt_subscribe`	`broker_url`, `topic`, `client_id`, `username`/`password`
`flow_jdbc_poll`	`dsn`, `sql`, `interval_ms`
`flow_on_*` (ドメイン)	`_pattern` (グロブ — `MOTOR-`, `SITE-?`等)

変換クイック設定

ノード	核心オプション
`flow_script_transform`	`language` (`JS`/`EQL`), `script`, `timeout_ms` (デフォルト500)
`flow_change_originator`	`entity_type`, `id_field`
`flow_rename_keys`	`mapping` (例: `{"old":"new"}`)
`flow_template`	`template` (`${data.x}` / `${metadata.y}` 置換)
`flow_split`	`path` (配列位置、デフォルト `data`)
`flow_to_email`	`subject_template`, `body_template`

フロー制御クイック設定

ノード	核心オプション
`flow_delay`	`delay_ms`
`flow_throttle`	`max_msgs`, `window_ms`, (分岐: SUCCESS/THROTTLED)
`flow_debounce`	`window_ms`
`flow_merge`	`window_ms` (data.merged配列に累積)
`flow_subflow`	`target_flow_id`
`flow_retry`	`max_attempts` (デフォルト3), `backoff_ms` (デフォルト1000), `backoff_multiplier` (デフォルト2.0)
`flow_log`	`level` (INFO/WARN/ERROR), `prefix`
`flow_noop`	(オプション無し)

外部連携クイック設定

ノード	静的オプション	動的オプション(`*_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`	—

アクション — 統合/保存クイック設定

ノード	核心オプション
`flow_save_tag_point`	`tag_id_field` (デフォルト `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`

ドメインCRUDクイック設定

ノード	核心オプション
`flow_create_*`	ドメイン別フィールド+`*_field` 動的オプション。NOT NULLフィールドは一部自動補強(自動補強項目はドメインCRUD表参照)
`flow_update_*`	部分更新: 入力したフィールドのみ更新、空値は無視。全体上書きはDelete + Create組み合わせ
`flow_delete_*`	`*_id_field`
`flow_start/end/pause/resume_work_order`	`order_id_field` (デフォルト `data.order_id`)
`flow_abort_work_order`	+ `abort_code_field`, `abort_notes_field`
`flow_update_tag_alarm_band_numeric`	`tag_id_field`, `hi_field`等 (入力フィールドのみ更新)

エッジクイック設定

エッジノードはすべて同一のオプションセットを使用します。

オプション	説明
`url`	エッジRESTエンドポイント (例: `http://edge.local:60000/opc/server`)
`method`	HTTPメソッド(未設定時ノード別デフォルト値)
`headers`	JSONヘッダー(認証トークン等)
`body_template`	リクエスト本文(未設定時 `data` そのまま送信)
`timeout_ms`	5000

フロー REST API — プログラムからフローを操作

フローを外部オートメーションツール(Ansible/GitOps/CIパイプライン)からコードで管理したり、外部システムがフローを即座に実行させたい時に使うREST API。

認証

すべてのAPI呼び出しはセキュリティ → API認証トークンで発行したトークンをヘッダーとして添付します。

Authorization: Bearer {api_token}

エンドポイント一覧

1) フロー一覧照会

GET /flow/list

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

応答:

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

2) 単一フロー照会

GET /flow/get/{flow_id}

応答本文にグラフノード・関係・オプションがすべて含まれます。バックアップ/バージョン管理にそのまま使用可能です。

3) フロー作成

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

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

応答から flow_id を受け取り後続呼び出しに使用します。

4) フロー修正(メタ)

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

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

5) グラフ保存(ノード・関係一括交換)

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" }, ... ]
}

グラフ保存はトランザクション — 検証失敗時にグラフ全体がロールバックされます。

6) 配備 / 解除

フローメタの deployed フィールドを変更すると自動で配備・解除されます。

# 배포
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) 即時実行(手動トリガー)

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

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

トリガーノードの事前フィルターを経ずに最初のノードから即座に実行されます。手動検証・デバッグ時に有用。

8) Dispatch(トリガー迂回発火)

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 }
}

正常なトリガー処理経路にメッセージを進入させます — 該当メッセージにマッチするすべてのフローが同時に発火します。

9) 統計初期化

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

10) Export / Import — グラフバックアップ

# 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

Import時に衝突IDがあれば新しいIDが自動発行されます。外部依存性(認証情報・資産ID等)はimport後別途マッピングしてください。

11) すべてのフロー再配備

POST /flow/redeploy

大量変更後または、ノード再起動後の一括同期に使用。運用中の呼び出しは慎重に — 一時的な処理遅延が発生します。

12) 自己診断

POST /flow/selftest

フローエンジンの内部コンポーネント(トリガーキュー・ディスパッチャー・ノードレジストリ・スクリプトランタイム)を一度点検し状態を返します。結果は診断ログにも記録されます。

13) ノードカタログ照会

GET /flow/catalog

現在登録されたすべてのノードタイプ・オプションスキーマを返します。UIがインスペクターを動的生成するのに使用。外部ツールがグラフを自動生成する際も参考。

Webhookトリガーで外部から発火

flow_on_webhook トリガーが含まれるフローは次のURLで外部システムが直接発火できます。

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

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

応答:

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

応答コード	意味
200	メッセージキューにenqueueされた(実際の処理結果は非同期)
401	認証トークン誤り/欠落
404	フローIDが無いか未配備
413	本文256KB超過
422	トリガーノードが `flow_on_webhook` でない
429	分当たり呼び出し限度超過

外部オートメーションツール連携例

GitHub Actions — PRマージ時にフロー配備

- 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 — フローバッチ管理

- 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

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

エンドポイント	分当たり限度(トークン毎)
`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

限度超過時 429 Too Many Requests 応答。

OPC/PLC産業統合パターン

産業現場特有の統合シナリオ集。エッジノード22種と資産イベント発行4種の組み合わせ活用。

パターンA — OPCサーバー自動登録

新しいラインが稼働する時ERPからライン情報を受けエッジにOPCサーバーを自動登録。

[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]
   ↓ (등록 이력을 자산 메타에 기록)

エッジノードは自動で mm_edge マスターテーブルから api_key をlookupします。運用者が別途認証設定をする必要はありません。

パターンB — PLC命令自動発行

資産アラームが発生するとPLCに停止命令を自動発行。

[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 は資産の命令トピック(asset/{id}/cmd/STOP)に即座に伝達されエッジがPLCにOPC Writeします。

パターンC — OPCタグ一括登録 (CSV/Excel入力)

新規設備セットアップ時にCSVのタグ100〜1000件を一度に登録。

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

1,000件登録時は flow_throttle で分当たり100件以下に平坦化することを推奨します。エッジが一時的に応答が遅くなる可能性があります。

パターンD — OPC切断自動復旧

OPC接続が切れた後復旧すると自動で再起動。

[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: [관리자 알림]

パターンE — シフト開始時ライン状態点検

各シフト開始時刻にサイトのすべてのラインを点検。

[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 만 한 통의 메일에 합쳐 발송

パターンF — シフト変更時作業者自動マッピング

勤務表変更に応じて自動で作業指示に作業者をマッピング。

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

パターンG — 多重PLC同時呼び出し (Scatter-Gather)

複数のPLCのデータを同時呼び出し後結果統合。

[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]

パターンH — アラームバンド自動調整

予測分析の学習されたLIMIT_MIN/MAX値をアラームバンドに自動適用。

[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] (적용 결과 기록)

パターンI — ダウンタイム自動分類

資産停止イベントを理由別に分類しOEE可用性に正確に反映。

[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 (긴급)]

パターンJ — 外部ERPとの双方向作業指示同期

プラントパルス↔外部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)]

無限ループ防止のため外部から入ったメッセージには metadata.source='ERP' を表示し、逆方向のフローでそのメッセージをフィルタリングしてください。

産業統合時の注意事項

事項	推奨
OPC Write命令は安全インターロック後発行	運転員確認または自動安全ルール通過後にのみ
PLCポーリング周期を短く設定しすぎない	1秒以下はPLC CPU占有率急増の危険
エッジデバイスのDockerコンテナ資源限度	OPC収集+追加コンテナ時CPU/メモリ80%以上注意
試運転段階ではすべての命令をdry-runモードで	`flow_edge_tag_write` の `dry_run=true` オプション活用
停電対策 — 永久保存トリガーのみ信頼	`flow_on_webhook` 等は揮発性、ドメインイベントトリガーは永久

外部システム統合Cookbook

よく呼び出す外部システム別 flow_http_request 設定とペイロード例集。

Slack — Incoming Webhook通知

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}" } }
    ]
  }

成功: 200 + ok 本文
失敗: 400(不正なpayload) / 404(不正なsecret) / 429(分当たり呼び出し限度超過)
推奨: 分当たり1回以下に flow_throttle 結線

Microsoft Teams — Webhook通知

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}"      }
      ]
    }]
  }

themeColor を FF0000(赤)/FFA500(オレンジ)/00C853(緑)に分岐して優先度を視覚化してください。

Jira — イシュー自動生成

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" }
    }
  }

応答本文 data.response_body.key に OPS-1234 のようなイシューキーが入るので後続ノードで活用
失敗: 400(不正なフィールド) / 401(認証) / 403(プロジェクト権限無し)

GitHub Issues — イシュー自動生成

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) — 作業指示発行

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}"
  }

CSRFトークンを先にGETで取得した後、同じセッションクッキーでPOSTする必要があります。flow_http_request ノードを2つ接続して最初のノードの応答ヘッダーのトークンを2番目のノードのヘッダーに伝播させてください。

MES (外部DB直接INSERT) — `flow_jdbc_query`

外部MESシステムの作業指示テーブルに直接行を挿入。

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')

ノードオプション	値
`datasource_id`	mes_db (システム設定に事前登録)
`query`	上記SQL
`binds`	`{"order_no":"${data.order_no}","asset_id":"${metadata.asset_id}",...}`

flow_jdbc_query はSELECT以外にINSERT/UPDATE/DELETEもサポートします。トランザクションはノード単位で自動commit/rollbackされます。

Grafana — ダッシュボード自動更新通知

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}"
  }

アラーム発生時グラフ上にマーカーが自動表示されます — 事後分析時に非常に有用。

Telegram — ボットメッセージ

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"
  }

REST認証パターンクイック表

認証方式	ヘッダー
API Key (ヘッダー)	`X-API-Key: {token}`
API Key (クエリ)	URL末尾に `?api_key={token}`
Bearer Token	`Authorization: Bearer {token}`
Basic Auth	`Authorization: Basic {base64(user:pass)}`
OAuth2 (Client Credentials)	トークン発行→Bearerヘッダー(別途ノードで更新)
HMAC署名	本文/時刻ベースの署名をスクリプトで計算して `X-Signature` ヘッダー

応答パースパターン

flow_http_request 応答後 data.response_body を次のノードで活用:

// 응답에서 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

外部システム呼び出し時のセキュリティ推奨

事項	推奨
トークンをグラフ内に平文で置かない	システム設定→認証情報保管庫に登録後 `${creds.slack_webhook}` のように参照
応答本文に機密情報があればマスキング	`flow_script_transform` でPII/トークンを除去後次のノードへ
外部システムごとに別途retryポリシー	呼び出し頻度が高いところは別途 `flow_retry` + `flow_throttle` 結線
呼び出し結果を監査ログとして保存	`flow_save_attributes` で資産に呼び出し履歴メタを追加

データ変換Cookbook

flow_script_transform ノードにそのまま移して使用できる変換パターン集。すべての例はJavaScript基準で data / metadata 短縮別名を使用します。

単位変換

// 섭씨 → 화씨
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

日付・時刻変換

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

文字列正規化

// 공백 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

ネストしたオブジェクトの平坦化

// 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

平坦なオブジェクト→ネスト

// 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行生成

// 외부 시스템에 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

配列集計

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

条件付きフィールド追加(スキーマ進化)

// 알람 우선순위에 따라 색상·아이콘 자동 부여
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

ペイロードの一部マスキング

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

多言語(i18n)テンプレート

// 메시지 본문을 사용자 로케일에 따라 분기
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

JSON Path安全照会

// 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

メッセージ統合(`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

再利用スクリプト集

flow_script_filter / flow_script_transform / flow_switch ノードにそのまま移して使用できるスクリプトライブラリです。すべてのスクリプトはJavaScript基準で data / metadata 短縮別名を使用します。

変換スクリプト

単位変換・ラベリング

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

時刻・シフト・曜日メタ補強

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

平均±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

MES PO→作業指示マッピング

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

外部ペイロード正規化(様々なフィールド名統一)

// 외부 시스템에 따라 키 이름이 다른 경우 — 한 줄로 정규화
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

閾値違反回数累積 (Statefulパターン)

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

メッセージ統計要約 (Merge結果加工)

// 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

時間帯別閾値適用

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

フィルタースクリプト

優先度ホワイトリスト

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

時間ウィンドウ(平日のみ)

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

資産・タグIDパターン

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

閾値+安定性(連続N回)

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

営業時間・休日チェック

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

特定サイトのみ

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

必須フィールドすべて存在検査

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

自己発行メッセージ遮断(双方向同期)

metadata.source !== 'flow'

スイッチケーススクリプト

優先度等級別

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

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

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

資産状態別

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

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

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

作業指示ライフサイクル別

// 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'

上記スクリプトはすべて運用環境で頻繁に使われるパターンを集めたものです。グラフ自体に直接結線しても動作し、ドメインに合わせて閾値・フィールド名だけ調整すればよいです。

用語集

フローエンジン用語

用語	意味
entity_type	メッセージの主体エンティティ種類 — `Asset`, `Tag`, `Site`, `Order`, `Customer`, `Product`, `Employee`, `Calendar` 等
originator	メッセージが指す主体エンティティ(`entity_type` + `id`)。例: `Asset/MOTOR-001`
type	メッセージ分類ラベル。トリガーがどんな種類のイベントを受け取ったかを識別します
data	メッセージの本文(ペイロード) — 変換・アクションノードが主に読み書きします
metadata	メッセージのコンテキスト(時刻・サイト・シフト・タグID等) — 変換されても流れの最後まで維持されます
relation	ノード出力ワイヤーのラベル。`SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` (すべて大文字)
*`_field` 動的オプション**	静的値の代わりにメッセージペイロードのパス(例: `data.tag_id`)から値を読み込む入力方式
グロブパターン	トリガーの `_pattern` オプションのワイルドカード表現 — `` は0文字以上、`?` は正確に1文字にマッチ
スナップショット	グラフ保存時に自動積載されるバージョン単位のバックアップ。事故時に以前の状態に戻す際に使用
部分更新(fetch+merge)	Updateノードが既存レコード照会後入力したフィールドのみマージする方式。空値は無視
SKIPPED	トリガーパターン未マッチメッセージが後続ノードへ流れずカウンターからも除外される処理
EXHAUSTED	`flow_retry` ノードが最大再試行回数を超えた時に発火する分岐
THROTTLED	`flow_throttle` ノードがウィンドウ内の限度を超えたメッセージを遮断する時に発火する分岐

ドメイン・略語辞典

プラント運用環境で頻繁に使われる略語を、本マニュアル内で素早く参照できるよう整理しました。

略語	意味	本マニュアルでは
MES	Manufacturing Execution System — 作業指示・生産実績管理	外部連携対象 (HTTP/MQTT/外部DB)
ERP	Enterprise Resource Planning — 全社資源・計画システム	外部連携対象
SCADA	Supervisory Control and Data Acquisition — 監視・制御	外部連携対象 / `flow_publish_asset_command` の出口
OPC	Open Platform Communications — 産業通信標準	エッジカテゴリ(`flow_edge_opc_*`)の対象
OEE	Overall Equipment Effectiveness — 稼働率×性能×品質	`flow_on_oee_event` トリガー
RAM	Reliability・Availability・Maintainability	`flow_on_ram_event` トリガー
EMS	Energy Management System	`flow_on_ems_event` トリガー
EQL	ドメインイベントルール表現式	`flow_script_filter`/`flow_script_transform` の `language: "EQL"` オプション
CEP	複合イベント処理 (Complex Event Processing)	EQLベースのルールエンジン。アラームはCEP経路でのみ発生すべきで整合性維持
PO	Purchase Order — 購買・生産注文	MES連携時に作業指示に変換される単位
WO	Work Order (作業指示)	`flow_*_work_order` ノード群
CMMS	Computerized Maintenance Management System	外部連携対象
MTTF/MTTR	Mean Time To Failure / To Repair	RAMイベントのコア指標
HMI	Human–Machine Interface	SCADA等の運用画面

バージョンノート

マニュアルが現時点で扱う主要機能群の導入時点です。以前のバージョンを運用中の場合、一部機能が異なる動作をする可能性があります。

V2026.05 — ドメインオートメーション拡張

エッジ(Edge)カテゴリ11種新設 — OPCサーバー/タグCRUD+モニタリング自動化
作業指示状態遷移5種 — flow_start/end/pause/resume/abort_work_order
タグアラームバンド部分更新2種 — 数値型/ブール型アラームバンドを入力フィールドのみ更新
flow_retry フロー制御新規 — バックオフ+最大試行+EXHAUSTED分岐
新しいトリガー5種 — flow_on_asset_health_status / _connection_status / _oee_event / _ram_event / _ems_event
Updateノード部分更新(fetch+merge) — 既存レコード照会後入力フィールドのみマージ
Relations大文字標準化 — SUCCESS/FAILURE/... すべて大文字、既存グラフは自動変換
SKIPPED処理 — トリガーパターン未マッチメッセージをカウンターから除外
すべて再配置(redeploy)運用ツール — アクティブフロー一括リロード
全体カウント初期化 — ノード+フロー単位の統計一括リセット

V2026.03 — リリース安定性

グラフ保存時に自動スナップショット積載
ライブデバッグパネル (インスペクター下部2秒更新) + ノード点灯・duration表示
import/export (グラフJSONファイル)
負荷ウォッチドッグ導入 — 負荷持続時に診断ログ一行発行

それ以前

M1 — エンジン・UI骨格、グラフCRUD、ビジュアルキャンバス
M2 — ドメインアクションノード(資産・タグ・作業指示・生産ドメイン等)
M3 — トリガー多様化、フィルター/変換/外部連携8種、スクリプトエンジン
M4 — デバッグ積載・実行履歴画面、統計、配備/解除、インポート/エクスポート
M5 — ドメイントリガー統合、シナリオ一括検証

本マニュアルが扱うすべての機能・動作は上記V2026.05時点を基準としています。

目次​

概要​

学習ロードマップ — 役割別推奨進入順序​

👶 初めてのユーザー (1時間以内に最初のフローを作る)​

🧑‍🏭 現場運用者 (オートメーションシナリオ作成)​

🛠 システム管理者 (運用・チューニング・障害対応)​

🔌 開発者・統合エンジニア (外部システム連携)​

🔐 セキュリティ担当 (監査・認証)​

📚 クイックリファレンス (既に使い慣れたユーザー)​

コア概念​

最初に始める​

エンドツーエンドチュートリアル​

ステップ1 — 新しいフローを作成​

ステップ2 — トリガーノードの配置​

ステップ3 — 閾値フィルター​

ステップ4 — ドメイン変更 (タグ → 資産)​

ステップ5 — 作業指示発行​

ステップ6 — メール通知 (成功分岐)​

ステップ7 — 失敗分岐の処理​

ステップ8 — 保存とテスト実行​

ステップ9 — 検証と配備​

ステップ10 — 運用モニタリング​

次のステップ​

画面構成​

一覧画面​

上部ツール​

一覧テーブル​

一括配備・解除​

すべて再配置​

編集画面 (ビジュアルキャンバス)​

上部ツールバー​

テスト実行の使い方​

左側 — ノードパレット​

中央 — キャンバス​

右側 — ノード設定​

カウント初期化​

右側 — ライブデバッグ​

メッセージ構造​

メッセージタイプ​

トリガー別ペイロード例​

flow_on_tag_point — タグポイント入力​

flow_on_tag_alarm — タグアラーム発生​

flow_on_asset_data / flow_on_asset_event — 資産イベント​

flow_on_asset_health_status / flow_on_asset_connection_status — 周期評価​

flow_on_oee_event / flow_on_ram_event / flow_on_ems_event — プラグインイベント​

flow_on_opc_status / flow_on_edge_status — OPC/エッジ状態​

flow_on_diagnostic — システム診断メッセージ​

flow_on_domain_changed — ドメイン変更イベント​

flow_on_entity_event — エンティティライフサイクル​

flow_on_webhook — 外部HTTPプッシュ​

flow_on_mqtt_subscribe — MQTTトピック購読​

flow_jdbc_poll — 外部DB周期ポーリング​

flow_schedule — 時間ベースの発火​

ペイロード変換時の注意事項​

ノードカタログ​

トリガー (22種)​

フィルター (6種)​

変換 (8種)​

アクション — 統合・保存 (2種)​

アクション — 資産イベント発行 (4種)​

flow_asset_command_invoke — 発行ではなく実行​

アクション — ドメインCRUD (34種)​

タグアラームバンド部分更新 (2種)​

作業指示(WorkOrder)状態遷移 (5種)​

エッジ (Edge, 22種)​

照会3種​

Dockerアプリ制御7種​

外部連携 (9種)​

フロー制御 (7種)​

Retry推奨結線パターン​

ノードオプション詳細リファレンス​

flow_script_filter — スクリプトフィルター​

スクリプトコンテキスト​

例​

flow_script_transform — スクリプト変換​

例​

flow_switch — 多重分岐​

例​

flow_retry — 自動再試行​

metadata 自動補強​

目次

概要

学習ロードマップ — 役割別推奨進入順序

👶 初めてのユーザー (1時間以内に最初のフローを作る)

🧑‍🏭 現場運用者 (オートメーションシナリオ作成)

🛠 システム管理者 (運用・チューニング・障害対応)

🔌 開発者・統合エンジニア (外部システム連携)

🔐 セキュリティ担当 (監査・認証)

📚 クイックリファレンス (既に使い慣れたユーザー)

コア概念

最初に始める

エンドツーエンドチュートリアル

ステップ1 — 新しいフローを作成

ステップ2 — トリガーノードの配置

ステップ3 — 閾値フィルター

ステップ4 — ドメイン変更 (タグ → 資産)

ステップ5 — 作業指示発行

ステップ6 — メール通知 (成功分岐)

ステップ7 — 失敗分岐の処理

ステップ8 — 保存とテスト実行

ステップ9 — 検証と配備

ステップ10 — 運用モニタリング

次のステップ

画面構成

一覧画面

上部ツール

一覧テーブル

一括配備・解除

すべて再配置

編集画面 (ビジュアルキャンバス)

上部ツールバー

テスト実行の使い方

左側 — ノードパレット

中央 — キャンバス

右側 — ノード設定

カウント初期化

右側 — ライブデバッグ

メッセージ構造

メッセージタイプ

トリガー別ペイロード例

`flow_on_tag_point` — タグポイント入力

`flow_on_tag_alarm` — タグアラーム発生

`flow_on_asset_data` / `flow_on_asset_event` — 資産イベント

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 周期評価

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — プラグインイベント

`flow_on_opc_status` / `flow_on_edge_status` — OPC/エッジ状態

`flow_on_diagnostic` — システム診断メッセージ

`flow_on_domain_changed` — ドメイン変更イベント

`flow_on_entity_event` — エンティティライフサイクル

`flow_on_webhook` — 外部HTTPプッシュ

`flow_on_mqtt_subscribe` — MQTTトピック購読

`flow_jdbc_poll` — 外部DB周期ポーリング

`flow_schedule` — 時間ベースの発火

ペイロード変換時の注意事項

ノードカタログ

トリガー (22種)

フィルター (6種)

変換 (8種)

アクション — 統合・保存 (2種)

アクション — 資産イベント発行 (4種)

`flow_asset_command_invoke` — 発行ではなく実行

アクション — ドメインCRUD (34種)

タグアラームバンド部分更新 (2種)

作業指示(WorkOrder)状態遷移 (5種)

エッジ (Edge, 22種)

照会3種

Dockerアプリ制御7種

外部連携 (9種)

フロー制御 (7種)

Retry推奨結線パターン

ノードオプション詳細リファレンス

`flow_script_filter` — スクリプトフィルター

スクリプトコンテキスト

例

`flow_script_transform` — スクリプト変換

例

`flow_switch` — 多重分岐

例

`flow_retry` — 自動再試行

`metadata` 自動補強