> For the complete documentation index, see [llms.txt](https://kopens.gitbook.io/plantpulse-platform/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://kopens.gitbook.io/plantpulse-platform/user/flow.md).
# 플로우 (Flow)
## 목차
**시작하기**
* [개요](#overview) · [학습 로드맵](#learning-roadmap) · [핵심 개념](#concepts) · [처음 시작하기](#quickstart) · [엔드 투 엔드 튜토리얼](#tutorial) · [화면 구성](#layout)
**화면별 가이드**
* [목록 화면](#list) · [편집 화면 (시각 캔버스)](#edit) · [실행 이력 화면](#log)
**메시지·노드 레퍼런스**
* [메시지 구조](#message) · [트리거별 페이로드 예제](#trigger-payloads) · [노드 카탈로그](#node-catalog)
* [트리거](#triggers) · [필터](#filters) · [변환](#transforms)
* [액션 — 통합/저장](#actions-integration) · [자산 이벤트 발행](#actions-asset) · [도메인 CRUD](#actions-crud)
* [엣지](#edge-nodes) · [외부 연동](#externals) · [흐름 제어](#controls)
* [노드 옵션 상세 레퍼런스](#node-reference) · [스크립트 노드 작성](#scripting) · [JS 실행 환경 사양](#js-runtime) · [메모리 안전 작성 패턴](#memory-safety) · [모바일 푸시 알림 채널](#mobile-push) · [외부 인증 토큰 자동 갱신](#auth-refresh)
**예시·패턴**
* [예시 플로우 (17종)](#examples) · [운영 패턴 (Recipes 10종)](#patterns) · [활용 예시](#use-cases)
**운영**
* [일반 워크플로우](#workflow) · [가져오기/내보내기](#import-export) · [에러 핸들링](#error-handling) · [운영 진단](#diagnostic) · [권한](#permissions)
**프로덕션 운영**
* [자주 묻는 질문 (FAQ)](#faq) · [운영 모범 사례](#best-practices) · [보안·민감 정보 처리](#security)
* [플로우 메트릭과 알람](#metrics) · [메시지 처리 의미론과 백압](#semantics) · [에러 코드 카탈로그](#error-codes)
* [클러스터·HA 동작](#cluster) · [종단간 트레이스](#tracing) · [그래프 패턴 카탈로그](#graph-patterns) · [플로우 테스트 모범 사례](#testing)
* [성능 한계와 튜닝](#performance) · [감사·이력 추적](#audit) · [신규 플로우 배포 체크리스트](#checklist) · [배포 전략 (Canary/Blue-Green/A·B)](#deployment-strategies) · [긴급 대응 절차](#emergency)
* [노드 빠른 설정 레퍼런스](#cheatsheet) · [플로우 REST API](#rest-api) · [Webhook 트리거](#webhook-trigger) · [OPC/PLC 산업 통합 패턴](#industrial-patterns)
* [외부 시스템 통합 Cookbook](#integration-cookbook) · [데이터 변환 Cookbook](#transform-cookbook) · [재사용 스크립트 모음](#script-library)
**문제 해결**
* [단계별 디버깅 가이드](#debug-guide) · [자주 겪는 문제](#troubleshooting)
**기타**
* [용어 사전](#glossary) · [버전 노트](#changelog) · [관련 화면](#related)
***
## 개요
플로우는 외부 시스템(MES/ERP/SCADA 등)과의 데이터 연동, 도메인 객체(태그·자산·작업지시·작업자 등)의 자동 생성·갱신을 **시각적 그래프**로 정의하는 자동화 메뉴입니다. 코딩 없이 운영자가 직접 자동화 시나리오를 설계·배포·운영하실 수 있습니다.
100여 종의 노드를 캔버스 위에 드래그 앤 드롭으로 배치하고 와이어로 연결하여 데이터 처리 파이프라인을 구성합니다.
**경로**: 왼쪽 메뉴 > **Automation > 플로우**
***
## 학습 로드맵 — 역할별 추천 진입 순서
본 매뉴얼은 4,000 라인 이상의 종합 가이드입니다. 자신의 역할과 목적에 맞는 섹션부터 읽으시면 효율적입니다.
### 👶 처음 사용자 (1시간 내 첫 플로우 만들기)
1. [핵심 개념](#concepts) — 5분
2. [처음 시작하기](#quickstart) — 5분
3. [엔드 투 엔드 튜토리얼](#tutorial) — 30분 (단계 1\~10 따라가기)
4. [화면 구성](#layout) + [편집 화면](#edit) — 10분
5. [예시 플로우 1·2·3](#examples) 따라하기 — 10분
→ 첫 플로우 배포 완료. 이후 필요한 노드를 [노드 카탈로그](#node-catalog) 에서 검색.
### 🧑🏭 현장 운영자 (자동화 시나리오 작성)
1. [트리거별 페이로드 예제](#trigger-payloads) — 실제 데이터 구조 이해
2. [노드 옵션 상세 레퍼런스](#node-reference) — 자주 쓰는 노드 옵션 숙지
3. [데이터 변환 Cookbook](#transform-cookbook) — 흔한 변환 패턴 복사 사용
4. [그래프 패턴 카탈로그](#graph-patterns) — 결선 패턴 선택
5. [예시 플로우 (17종)](#examples) — 시나리오별 완성품 참고
### 🛠 시스템 관리자 (운영·튜닝·장애 대응)
1. [플로우 메트릭과 알람](#metrics) — 어떤 지표를 봐야 하나
2. [에러 코드 카탈로그](#error-codes) — 진단 로그 해석
3. [클러스터·HA 동작](#cluster) — 다중 노드 환경 이해
4. [종단간 트레이스](#tracing) — 문제 메시지 추적
5. [성능 한계와 튜닝](#performance) + [긴급 대응 절차](#emergency)
### 🔌 개발자·통합 엔지니어 (외부 시스템 연동)
1. [외부 시스템 통합 Cookbook](#integration-cookbook) — Slack/Teams/Jira/SAP 예제
2. [플로우 REST API](#rest-api) — 프로그램으로 플로우 조작
3. [Webhook 트리거](#webhook-trigger) — 외부에서 플로우 발화
4. [OPC/PLC 산업 통합 패턴](#industrial-patterns) — 산업 현장 시나리오
5. [JS 실행 환경 사양](#js-runtime) + [재사용 스크립트 모음](#script-library)
### 🔐 보안 담당 (감사·인증)
1. [보안·민감 정보 처리](#security) — 자격 증명 보관
2. [외부 인증 토큰 자동 갱신](#auth-refresh) — OAuth2 토큰 운영
3. [권한](#permissions) — 역할별 가능 동작
4. [감사·이력 추적](#audit) — 변경/실행 이력 보존
### 📚 빠른 참조 (이미 익숙한 사용자)
| 찾는 정보 | 섹션 |
| -------------- | ----------------------------------------------------------------------- |
| 노드 ID 와 한 줄 설명 | [노드 카탈로그](#node-catalog) |
| 노드 옵션 기본값 | [노드 옵션 상세 레퍼런스](#node-reference) |
| 메시지 JSON 예제 | [트리거별 페이로드 예제](#trigger-payloads) |
| 그대로 쓸 스크립트 | [재사용 스크립트 모음](#script-library) · [데이터 변환 Cookbook](#transform-cookbook) |
| API 호출 curl | [플로우 REST API](#rest-api) |
| 에러 코드 의미 | [에러 코드 카탈로그](#error-codes) |
| 문제 해결 가이드 | [단계별 디버깅 가이드](#debug-guide) · [자주 겪는 문제](#troubleshooting) |
| 빠른 옵션 표 | [노드 빠른 설정 레퍼런스](#cheatsheet) |
> 모든 섹션 링크는 동일 문서 내 앵커입니다. `Ctrl+F` 로 키워드 검색도 효과적입니다.
***
## 핵심 개념
| 용어 | 설명 |
| ----------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **플로우 (Flow)** | 노드와 와이어(관계)의 집합으로 구성된 하나의 자동화 워크플로우입니다. 진입점이 있는 방향 그래프이며, **배포/해제** 토글로 활성화 여부를 제어합니다 |
| **노드 (Node)** | 메시지를 받아 처리한 뒤 다음 노드로 전달하는 단위입니다. 7개 카테고리(트리거·필터·변환·액션·외부 연동·흐름 제어·엣지)로 구분됩니다 |
| **관계 (Relation)** | 노드 출력에서 다음 노드로 가는 와이어의 레이블입니다. 노드가 직접 `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` 등의 라벨을 지정해 분기합니다. 모든 라벨은 **대문자로 표준화**되어 있습니다 |
| **메시지 (Message)** | 플로우 내부를 흐르는 페이로드입니다. `type`(분류), `originator`(주체 엔티티), `data`(본문), `metadata`(컨텍스트)를 포함합니다 |
| **트리거 (Trigger)** | 플로우의 시작점이 되는 노드입니다. 도메인 이벤트(태그 포인트·알람·자산 이벤트 등), 외부 진입(웹훅·MQTT·외부 DB), 시간(스케줄) 세 종류가 있습니다 |
***
## 처음 시작하기
플로우는 다음 4단계로 가장 빨리 익숙해지실 수 있습니다.
1. **목록 화면**에서 `새 플로우` 버튼을 눌러 빈 플로우를 만듭니다 (이름과 설명만 입력).
2. **편집 화면**에서 좌측 팔레트의 트리거 노드(예: `flow_on_alarm`) → 필터 → 액션(예: `flow_send_email`)을 차례로 드래그해 배치하고, 노드 사이를 와이어로 연결합니다.
3. 각 노드를 클릭해 우측 인스펙터에서 옵션을 입력한 뒤 우상단 **저장** → **테스트 실행**으로 동작을 한 차례 확인합니다.
4. 상단 **배포** 토글을 켜면 트리거 이벤트가 들어올 때마다 플로우가 자동 실행되며, 라이브 디버그 패널과 실행 이력 화면에서 결과를 확인하실 수 있습니다.
> 자동화 시나리오의 결선 패턴은 본 문서의 [예시 플로우](#examples)와 [활용 예시](#use-cases)를 참고해 주세요. 실제 플로우를 처음부터 끝까지 만드는 단계별 튜토리얼은 [엔드 투 엔드 튜토리얼](#tutorial)에서 따라하실 수 있습니다.
***
## 엔드 투 엔드 튜토리얼
운영 환경에서 실제로 사용 가능한 자동화 플로우 하나를 처음부터 끝까지 만들어 보는 단계별 튜토리얼입니다. **시나리오: 모터의 온도가 80°C 를 넘으면 자동으로 긴급 정비 작업지시를 발행하고 담당자에게 이메일로 알리기.**
### 단계 1 — 새 플로우 만들기
1. 왼쪽 메뉴에서 **Automation > 플로우** 클릭 → 목록 화면 열기
2. 상단 우측 **새 플로우** 버튼 클릭
3. 아래 정보를 입력하고 **확인**
| 항목 | 입력 값 |
| -- | ----------------------------------------------------------------- |
| 이름 | `모터 과열 자동 정비 발행` |
| 설명 | `[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com` |
플로우가 생성되면 자동으로 빈 캔버스의 편집 화면이 열립니다.
### 단계 2 — 트리거 노드 배치
자산 텔레메트리 이벤트로부터 모터 온도 데이터를 받습니다.
1. 좌측 팔레트의 **트리거 (Trigger)** 카테고리를 펼침
2. `flow_on_tag_point` 노드를 캔버스에 드래그
3. 노드 클릭 → 우측 인스펙터에서 다음 옵션 입력
| 옵션 | 값 |
| ---------------- | -------------- |
| **표시명** | `태그 포인트 인입` |
| `tag_id_pattern` | `MOTOR-*.TEMP` |
> `tag_id_pattern` 으로 모터 온도 태그만 통과시키면 후속 처리량이 크게 줄어듭니다. 다른 태그는 SKIPPED 처리되어 카운터에 포함되지 않습니다.
### 단계 3 — 임계값 필터
80°C 초과 메시지만 통과시키도록 스크립트 필터를 추가합니다.
1. **필터 (Filter)** 카테고리에서 `flow_script_filter` 드래그
2. 트리거 노드의 출력 포트에서 새 필터 노드의 입력 포트로 와이어 연결
3. 인스펙터에 다음 입력
| 옵션 | 값 |
| ---------- | ------------------ |
| **표시명** | `임계값 필터 (80°C 초과)` |
| `language` | `JS` |
| `script` | `data.value > 80` |
### 단계 4 — 도메인 변경 (태그 → 자산)
알람·작업지시는 자산 단위로 발행하는 게 자연스러우므로, 메시지의 originator 를 태그에서 상위 자산으로 변환합니다.
1. **변환 (Transform)** 카테고리에서 `flow_change_originator` 드래그
2. 필터 노드의 `TRUE` 출력에서 와이어 연결
3. 인스펙터:
| 옵션 | 값 |
| ------------- | ------------------- |
| **표시명** | `Tag → Asset 변경` |
| `entity_type` | `Asset` |
| `id_field` | `metadata.asset_id` |
> `metadata.asset_id` 는 태그 포인트 인입 시 자동 채워집니다. 만약 메시지에 없으면 태그 ID 의 prefix 부분(예: `MOTOR-001.TEMP` → `MOTOR-001`)을 스크립트로 추출하셔도 됩니다.
### 단계 5 — 작업지시 발행
긴급 정비 작업지시를 자동 생성합니다.
1. **액션 (Action) — 도메인 CRUD** 카테고리에서 `flow_create_work_order` 드래그
2. 변환 노드의 `SUCCESS` 출력에서 와이어 연결
3. 인스펙터:
| 옵션 | 값 |
| ------------------- | ---------------------------------- |
| **표시명** | `긴급 정비 작업지시 생성` |
| `asset_id_field` | `originator.id` |
| `title_field` | (정적 값) `긴급 점검 — 모터 과열` |
| `master_id_field` | (선택) `data.master_id` (없으면 자동 생성) |
| `description_field` | (정적 값) `자동 발행: 임계 온도 초과로 긴급 점검 필요` |
| `default_priority` | `HIGH` |
### 단계 6 — 이메일 알림 (성공 분기)
작업지시 발행이 성공하면 담당자에게 알립니다.
1. **외부 연동 (External)** 카테고리에서 `flow_send_email` 드래그
2. 작업지시 노드의 `SUCCESS` 출력에서 와이어 연결
3. 인스펙터:
| 옵션 | 값 |
| ------------------ | ------------------------------------------------------------------------------------------------------- |
| **표시명** | `정비 담당자 이메일` |
| `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 — 실패 분기 처리
작업지시 발행 자체가 실패할 수도 있습니다(예: 자산이 사라졌거나 권한 문제). 운영자에게 즉시 푸시 알림을 보냅니다.
1. **외부 연동** 의 `flow_send_push` 드래그
2. 작업지시 노드의 `FAILURE` 출력에서 와이어 연결
3. 인스펙터:
| 옵션 | 값 |
| --------------- | --------------------------------------------- |
| `title` | `자동화 실패` |
| `body_template` | `${originator.id} 정비 자동 발행 실패: ${data.error}` |
### 단계 8 — 저장과 테스트 실행
1. 우상단 **저장** 버튼 클릭. 자동 스냅샷이 적재되어 이후 롤백할 수 있습니다.
2. 우상단 **테스트 실행** 버튼 클릭 → JSON 편집기에 다음 입력 → **발행**
```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)` 표시 → **발행** 클릭.
3. 라이브 디버그 패널에서 다음을 확인:
* 트리거 노드 점등 녹색 → 메시지 통과
* 필터 노드: `92.5 > 80` 이므로 `TRUE` 분기
* 변환 노드: originator 가 Asset/MOTOR-001 으로 변경
* 작업지시 노드: SUCCESS, `data.work_order_id` 자동 부여
* 이메일 노드: 발송 시도
### 단계 9 — 검증과 배포
1. 작업지시 화면에서 새 작업지시가 등록되었는지 확인
2. 이메일이 정상 도착했는지 확인 (테스트 환경의 메일함)
3. **80°C 미만** 메시지로 한 번 더 테스트 (필터에서 차단되어야 함):
```json
{ "type": "POST_TELEMETRY", "originator": {"entity_type":"Tag","id":"MOTOR-001.TEMP"},
"data": {"value": 70}, "metadata": {"asset_id":"MOTOR-001"} }
```
필터 노드의 분기는 `FALSE`, 후속 노드 회색 — 정상.
4. 모든 분기를 검증한 뒤 **전체 카운트 초기화** 로 통계 윈도우를 리셋
5. 상단 **배포** 토글 ON
이제 실제 운영 환경에서 모터 온도가 80°C 를 넘는 즉시 자동으로 정비 작업지시가 발행되고 담당자에게 이메일이 갑니다.
### 단계 10 — 운영 모니터링
배포 직후 5\~10분간 다음을 확인하시면 안전합니다.
| 위치 | 확인 항목 |
| ---------- | --------------------------------------- |
| 목록 화면 | 해당 플로우 행의 실행 건수가 정상 범위에서 증가하는지 (폭주 아닌지) |
| 라이브 디버그 패널 | 노드 실패가 없는지 |
| 실행 이력 화면 | 실패한 메시지가 있다면 `NODE_ERROR` 의 원인 확인 |
| 받는 이메일 | 의도하지 않은 빈도로 알림이 가지 않는지 |
### 다음 단계
이 플로우를 발전시키시려면:
* **Retry 결선 추가** — 이메일/푸시 발송 실패 시 백오프 재시도 ([예시 8](#example-retry) 참고)
* **밴드 자동 조정** — 6시간 평균 ± 3σ 로 임계값 자동 보정 ([예시 6](#example-alarm-band))
* **다운타임 누적** — 과열 이력을 자산 attribute 에 누적 ([예시 14](#example-downtime))
* **품질 라인 격리** — 과열이 연속 발생하면 라인 자동 정지 ([예시 15](#example-quality))
***
## 화면 구성
플로우는 다음 세 화면으로 구성됩니다.
| 화면 | 용도 |
| --------- | ------------------------ |
| **목록** | 등록된 플로우 일람·검색·일괄 배포·가져오기 |
| **편집** | 시각 캔버스로 노드를 배치·연결·설정 |
| **실행 이력** | 노드 단위 실행 로그·타임라인 조회 |
***
## 목록 화면
상단 검색·생성 영역과 플로우 일람 테이블로 구성됩니다.
### 상단 도구
| 항목 | 설명 |
| ------------ | ------------------------------ |
| **상태 필터** | `전체` / `배포` / `해제` |
| **이름·설명 검색** | 키워드로 플로우를 필터링합니다 |
| **새 플로우** | 빈 플로우를 생성합니다 (이름·설명 입력) |
| **가져오기** | 내보내기로 받은 JSON을 업로드해 플로우를 복원합니다 |
| **모두 재배치** | 활성화된 모든 플로우를 한 번에 다시 적재합니다 |
| **새로고침** | 목록을 다시 불러옵니다 |
### 일람 테이블
목록 테이블은 25행 단위로 페이지네이션되며, 각 행에 처리 추세 스파크라인과 에러 비율 도넛 차트가 함께 표시됩니다. 페이지를 넘겨도 차트 상태가 유지됩니다.
| 컬럼 | 설명 |
| ------------ | ----------------------- |
| **선택** | 일괄 배포·해제용 체크박스 |
| **상태** | `배포` / `해제` 뱃지 |
| **플로우 ID** | `FLOW_NNNNN` 형식의 시퀀스 ID |
| **이름 \| 설명** | 운영자가 지정한 메타 정보 |
| **노드** | 포함된 노드 개수 |
| **트리거** | 트리거 노드 개수 |
| **실행 건수** | 누적 메시지 처리 카운트 |
| **처리 시간** | 노드 평균/최근 처리 시간 |
| **에러** | 누적 에러 카운트 |
| **최종 수정** | 그래프 마지막 저장 시각 |
| **액션** | `편집` / `삭제` 버튼 |
### 일괄 배포·해제
선택한 플로우들을 한꺼번에 **배포**(활성화) 또는 **해제**(비활성화)합니다. 배포된 플로우만 트리거 이벤트를 받습니다.
### 모두 재배치
상단 **모두 재배치** 버튼은 활성화된 모든 플로우를 다시 적재합니다. 다음 상황에서 사용하세요.
* 외부에서 그래프를 일괄 가져오기 한 직후
* 스케줄·외부 MQTT 구독·외부 DB 폴링 등 자체 스케줄러가 있는 트리거를 다시 등록하고 싶을 때
* 운영 중 캐시 일관성에 문제가 의심될 때
***
## 편집 화면 (시각 캔버스)
캔버스 좌측 팔레트에서 노드를 드래그해 배치한 후, 노드의 출력 포트를 클릭&드래그하여 다음 노드와 연결합니다.
### 상단 툴바
| 버튼 | 동작 |
| ---------- | --------------------------------------------------------------------- |
| **이름·설명** | 플로우 메타 정보를 편집합니다 |
| **배포/해제** | 현재 플로우를 즉시 활성/비활성화합니다 |
| **내보내기** | 그래프 전체를 JSON 파일로 다운로드합니다 |
| **테스트 실행** | 임의의 JSON 메시지를 주입해 한 차례 실행하며 결과를 확인합니다 (아래 [테스트 실행 사용법](#test-run) 참고) |
| **저장** | 현재 그래프를 서버에 저장합니다. 저장 시 자동 스냅샷이 적재되어 이후 롤백할 수 있습니다 |
> **트리거가 없는 그래프**를 저장하려고 하면 "수동 테스트 실행으로만 시작됩니다" 경고가 표시됩니다. 트리거를 의도적으로 빼고 수동 실행 전용 플로우로 사용하실 수 있습니다.
#### 테스트 실행 사용법
`테스트 실행` 버튼을 누르면 JSON 편집 다이얼로그가 열립니다. 운영자가 직접 메시지를 작성해 한 차례 발행할 수 있어, 트리거 이벤트를 기다리지 않고 그래프 동작을 검증할 수 있습니다.
| 항목 | 설명 |
| --------- | ----------------------------------------------------------- |
| **편집기** | 줄 번호·문법 강조가 있는 JSON 에디터. 메시지 본문을 자유롭게 작성합니다 |
| **검증 표시** | `type` 필수 필드가 있으면 `✓ JSON OK (type=X)`, 누락이면 `⚠ type 필수` 표시 |
| **발행** | `발행` 버튼으로 메시지를 디스패처에 주입. 결과는 라이브 디버그 패널과 실행 이력에서 확인 |
**기본 템플릿 예시**
```json
{
"type": "POST_TELEMETRY",
"originator": { "entity_type": "Asset", "id": "MOTOR-001" },
"data": { "speed": 1500, "temp": 75.3 },
"metadata": { "ts": 1746247200000, "site_id": "SITE-01" }
}
```
> 저장하지 않은 변경이 있을 때 테스트 실행을 누르면 "서버는 저장된 버전을 실행합니다" 안내가 뜹니다. 변경을 검증하려면 먼저 저장하세요.
### 좌측 — 노드 팔레트
카테고리별로 접고 펼칠 수 있으며, 검색 입력으로 즉시 필터링이 가능합니다.
| 카테고리 | 색상 | 노드 수 |
| ----------------------------- | -- | ---- |
| **트리거 (Trigger)** | 회색 | 23종 |
| **필터 (Filter)** | 파랑 | 6종 |
| **변환 (Transform)** | 녹색 | 6종 |
| **액션 (Action) — 통합/저장/자산 발행** | 주황 | 7종 |
| **액션 (Action) — 도메인 CRUD** | 주황 | 34종 |
| **외부 연동 (External)** | 보라 | 8종 |
| **흐름 제어 (Control)** | 회색 | 7종 |
| **엣지 (Edge)** | 청록 | 11종 |
### 중앙 — 캔버스
| 도구 | 단축키 / 조작 | 동작 |
| -------------------- | ------------------------------- | ------------------------------------------------- |
| **확대/축소** | `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 우선)으로 표시되며, 한글 코멘트도 안정적으로 정렬됩니다.
#### 카운트 초기화
노드 설정 패널 우상단에는 두 개의 초기화 버튼이 아이콘으로 표시됩니다. 각 버튼에 마우스를 올리면 툴팁이 안내합니다.
| 아이콘 버튼 | 동작 |
| ------------------------------ | ----------------------------------------------------------------------- |
| **🩹 (반창고)** — *에러 카운트 초기화* | 이 플로우 내 모든 노드의 누적 **에러 카운트만** 0으로 리셋합니다 |
| **🔄 (회전 화살표)** — *전체 카운트 초기화* | 처리·에러·처리 시간 모두 + 플로우 단위 통계를 모두 0으로 되돌립니다. 운영 검증을 끝내고 새로 통계를 시작할 때 사용하세요 |
> 두 버튼 모두 확인 다이얼로그 없이 즉시 적용됩니다 — 통계만 초기화될 뿐 노드 동작이나 메시지 처리에는 영향이 없습니다.
### 우측 — 라이브 디버그
라이브 디버그 패널이 인스펙터 아래에 표시됩니다.
| 항목 | 설명 |
| --------- | ------------------------------------- |
| **갱신 주기** | 2초 |
| **레벨 컬러** | 좌측 보더에 INFO(파랑)/WARN(노랑)/ERROR(빨강) 표시 |
| **표시 정보** | 노드 표시명 · 처리 시간(ms) · 메시지 미리보기 |
| **일시정지** | 패널 우상단 토글로 갱신을 일시정지합니다 |
| **비우기** | 누적된 디버그 항목을 화면에서만 비웁니다 |
캔버스의 노드 우상단에는 작은 점등이 표시됩니다.
| 색상 | 의미 |
| -- | ------------------ |
| 회색 | 대기 — 메시지를 받지 않은 상태 |
| 녹색 | 메시지가 통과 중 |
| 빨강 | 처리 중 에러 발생 |
노드 우하단에는 `평균 X · 최근 Y` 형태로 처리 시간이 표시됩니다.
***
## 메시지 구조
플로우 내부를 흐르는 메시지는 다음 4개 영역으로 구성됩니다.
```json
{
"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 웹훅 수신 |
| `KAFKA_INBOUND` / `MQTT_INBOUND` | 외부 토픽 수신 |
| `TIMER` | 스케줄 발화 |
### 트리거별 페이로드 예제
스크립트 노드 작성 시 어떤 필드에 접근할 수 있는지 정확히 알아야 합니다. 아래는 각 트리거가 만들어 내는 실제 메시지 JSON 예제입니다. 모든 트리거 메시지에는 공통으로 `type`, `originator`, `data`, `metadata` 가 포함됩니다.
#### `flow_on_tag_point` — 태그 포인트 인입
태그 한 개에 값이 들어올 때마다 발화합니다. 가장 흔한 트리거입니다.
```json
{
"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/...)를 넘기는 순간 발화합니다.
```json
{
"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 처리 결과·플러그인 평가 결과 등)가 발생할 때 발화합니다.
```json
{
"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분 주기로 자산 단위 헬스/연결 상태를 평가합니다.
```json
{
"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 평가 결과가 갱신될 때 발화합니다.
```json
{
"type": "OEE_EVENT",
"originator": { "entity_type": "WorkOrder", "id": "WO-20260512-001" },
"data": {
"oee": 0.78,
"availability": 0.95,
"performance": 0.85,
"quality": 0.97,
"good_count": 1560,
"bad_count": 42,
"target_count": 2000
},
"metadata": {
"order_id": "WO-20260512-001",
"asset_id": "LINE-1.PRESS",
"shift_id": "DAY-A",
"ts": 1746247200123
}
}
```
#### `flow_on_opc_status` / `flow_on_edge_status` — OPC/엣지 상태
```json
{
"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` — 시스템 진단 메시지
서버 모듈에서 발생한 진단 메시지가 들어오면 발화합니다.
```json
{
"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 될 때 발화합니다.
```json
{
"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` 세 타입으로 통합 발화. 단일 노드 하나로 세 가지 라이프사이클을 모두 받습니다.
```json
{
"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 /api/v4/flow/webhook/{flow_id}` 로 보낸 페이로드가 그대로 메시지로 변환됩니다. URL 경로의 `{flow_id}` 와 헤더 `X-API-Key` 로 인증.
```json
{
"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 토픽 구독
```json
{
"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 쿼리를 주기적으로 실행해 **각 행마다** 메시지 한 건씩 발화합니다.
```json
{
"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` 만 채워집니다.
```json
{
"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와 옵션은 편집 화면의 인스펙터에서 확인하실 수 있습니다.
### 트리거 (23종)
> **모든 플로우의 시작점은 트리거 노드**입니다. 별도의 진입점/종단점 노드는 없습니다.
**도메인 자동 수신** — 시스템 내부 이벤트가 자동으로 디스패치됩니다.
| 카테고리 | 트리거 노드 |
| ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| 태그 | `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_alarm`, `flow_on_entity_event` |
> 모든 트리거 노드는 `*_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 기준 분기 |
### 변환 (6종)
| 노드 | 설명 |
| ------------------------ | -------------------------- |
| `flow_script_transform` | 스크립트로 `data`/`metadata` 변환 |
| `flow_change_originator` | `originator`를 다른 엔티티로 변경 |
| `flow_rename_keys` | `data` 필드명 일괄 변경 |
| `flow_template` | `${path}` 치환 텍스트 생성 |
| `flow_split` | `data`가 배열이면 각 원소별 메시지로 분할 |
| `flow_to_email` | 메시지를 이메일 포맷으로 변환 |
### 액션 — 통합·저장 (3종)
| 노드 | 설명 |
| ---------------------- | -------------------------- |
| `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` | 자산 명령 |
### 액션 — 도메인 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, 11종)
엣지 디바이스(OPC Agent)의 REST API를 호출하여 OPC 서버 등록·태그 CRUD·태그 값 읽기/쓰기·모니터링 조회를 자동화합니다. 모든 노드가 의미상 기본 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_tag_create`, `flow_edge_tag_update`, `flow_edge_tag_delete`, `flow_edge_tag_read`, `flow_edge_tag_write` |
| 모니터링 | `flow_edge_monitoring` |
**공통 설정**
| 옵션 | 설명 |
| --------------- | ----------------------------------------------------------------------------------------- |
| `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` (그 외 또는 예외)
### 외부 연동 (8종)
모든 외부 노드는 `*_field` 동적 옵션을 지원합니다.
| 노드 | 동적 옵션 |
| ----------------------- | -------------------------------------------------------- |
| `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 객체 |
#### 예시
```javascript
// 온도가 임계 초과 + 야간 시프트만
msg.data.value > 80 && msg.metadata.shift === 'NIGHT'
```
```javascript
// 사이트별 임계 분기
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 값 사용) |
#### 예시
```javascript
// data 에 계산 필드 추가
msg.data.fahrenheit = msg.data.value * 9/5 + 32;
msg.metadata.processed_at = Date.now();
```
```javascript
// 페이로드 통째로 교체 (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` | — |
#### 예시
```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 | `*` | 메시지 사전 필터링 글롭 |
#### 호출 방법
```bash
curl -X POST \
https://platform.example.com/api/v4/flow/webhook/{flow_id} \
-H "X-API-Key: {edge_or_token_key}" \
-H "Content-Type: application/json" \
-d '{"order_no":"PO-001","customer":"ACME","quantity":1000}'
```
* 경로의 `{flow_id}` 는 목록 화면에서 복사 가능
* `X-API-Key` 는 엣지 [API Key](/plantpulse-platform/user/connection.md#edge-view) 또는 [API 인증 토큰](/plantpulse-platform/user/security.md#api-token) 화면에서 발급한 토큰
* 응답: `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 | — | 참조 |
| `bcc` / `bcc_field` | string | — | 숨은 참조 |
| `subject` / `subject_field` | string | (필수 1개) | 제목 — 템플릿 치환 |
| `body` / `body_field` | text | (필수 1개) | 본문 (HTML 허용) |
| `is_html` | boolean | `true` | 텍스트 메일이면 끄세요 |
| `attachments` | json | `[]` | `[{"url":"...","filename":"..."}]` |
> SMTP 설정은 [시스템 → 설정](/plantpulse-platform/user/system.md) 의 메일 설정에서 운영자가 미리 등록. 등록 전에는 모든 이메일 노드가 `FAILURE` 분기로 떨어집니다.
### `flow_jdbc_poll` — 외부 DB 폴링
| 옵션 | 타입 | 기본값 | 설명 |
| ------------------- | ------- | --------------------- | --------------------------------------------------------------- |
| `datasource_id` | string | (필수) | [시스템 → 설정](/plantpulse-platform/user/system.md) 에 등록된 외부 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 | 한 폴링당 최대 행 (이를 초과해도 안전) |
| `on_error_continue` | boolean | `true` | DB 오류 시 진단만 기록하고 다음 폴링 진행 |
#### `marker_column` 사용 예
```sql
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 엔드포인트를 호출하는 11종 노드는 공통 옵션을 공유합니다.
| 옵션 | 타입 | 기본값 | 설명 |
| --------------- | ------ | -------- | ---------------------------------------------------------- |
| `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` 노드는 두 가지 표현식을 지원합니다.
### JavaScript (기본 권장)
표준 ECMAScript 문법. 다중 라인·`var`/`let`/`const`·함수·객체 리터럴을 모두 지원합니다.
**바인딩**
| 변수 | 설명 |
| ---------- | ------------------------------------------------------------------------------------ |
| `msg` | 메시지 전체. `msg.data.x`, `msg.metadata.topic`, `msg.type`, `msg.originator.id` 모두 직접 접근 |
| `data` | `msg.data` 단축 별칭 |
| `metadata` | `msg.metadata` 단축 별칭 |
**필터 예시**
```js
data.temp > 80
```
**변환 예시**
```js
data.temp_f = data.temp * 1.8 + 32;
data.alert = data.temp > 80 ? 'HIGH' : 'OK';
msg
```
**스위치 예시 (case별 boolean)**
```js
data.t > 100 // case "Critical"
```
**자주 쓰는 변환 패턴**
```js
// 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
```
**자주 쓰는 필터 패턴**
```js
// 우선순위 화이트리스트
['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 | `NODE_SCRIPT_TIMEOUT` 에러 |
| **메모리** | 16MB | `NODE_SCRIPT_OOM` 에러 |
| **스택 깊이** | 1024 | 재귀 폭주 차단 |
| **출력 페이로드 크기** | 2MB | 자동 잘림 + 진단 WARN |
> 무거운 처리는 스크립트 1개로 통합하지 마시고 여러 노드로 분리하세요. 각 노드는 별도로 500ms 한도를 갖습니다.
#### `msg` 반환 규약
```javascript
// 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 — 클로저가 큰 데이터 유지
```javascript
// ❌ 나쁜 예 — 큰 배열을 변환 함수에 캡쳐
const heavy = data.records || []; // 1만 건
const summarize = (item) => heavy.find(r => r.id === item.id);
data.matched = data.targets.map(summarize);
msg
```
```javascript
// ✅ 좋은 예 — 인덱스 미리 만들고 함수 안에서만 사용
const index = {};
(data.records || []).forEach(r => { index[r.id] = r; });
data.matched = data.targets.map(t => index[t.id]);
data.records = undefined; // 변환 후 큰 원본 제거
msg
```
### 안티패턴 2 — 대용량 배열 그대로 전달
```javascript
// ❌ data.records 가 10,000 건이면 모든 후속 노드에서 메모리 차지
msg
```
```javascript
// ✅ 필요한 통계만 남기고 원본 제거
data.summary = {
count: (data.records || []).length,
total: (data.records || []).reduce((s, r) => s + r.value, 0)
};
delete data.records;
msg
```
### 안티패턴 3 — 깊은 객체 복사
```javascript
// ❌ JSON.parse(JSON.stringify(obj)) 는 큰 객체에서 매우 느림
data.copy = JSON.parse(JSON.stringify(data.original));
```
```javascript
// ✅ 얕은 복사 또는 필요한 필드만 직접 선택
data.summary = { id: data.original.id, name: data.original.name };
```
### 안티패턴 4 — 정규식 폭발 (Catastrophic Backtracking)
```javascript
// ❌ (a+)+b 형태의 중첩 그룹은 입력에 따라 지수 시간
const re = /^(a+)+b$/;
if (re.test(data.text)) ...
```
```javascript
// ✅ 비포획 그룹 + atomic 그룹 패턴 또는 단순 매칭
const re = /^a+b$/;
if (re.test(data.text)) ...
```
### 안티패턴 5 — `let` 누적 변수를 함수 밖에서
```javascript
// ❌ 함수 밖 let 은 매 노드 호출마다 0으로 리셋되지만, 의도와 다르게 동작 가능
let counter = 0;
data.items.forEach(() => counter++);
data.counter = counter;
```
```javascript
// ✅ 명시적으로 함수 안에서 선언
data.counter = data.items.length; // 같은 결과, 더 명확
```
### 안티패턴 6 — 중첩 try/catch 무한 시도
```javascript
// ❌ 실패해도 다시 던지지 않으면 그래프가 잘못된 분기로 이동
try {
riskyCall();
} catch (e) { /* 무시 */ }
msg
```
```javascript
// ✅ 실패 의도면 throw, 정상 의도면 명시적으로 기록
try {
riskyCall();
data.status = 'OK';
} catch (e) {
data.status = 'ERROR';
data.error_msg = e.message;
}
msg
```
> 스크립트 안에서 `throw` 한 예외는 `FAILURE` 분기로 가며 메시지는 보존됩니다. 외부 IO 노드는 따로 자체 분기를 가지므로 try/catch 로 감싸지 마세요.
### 메모리 압박 진단
진단 로그(`Log` 스트림)의 `code` 가 다음 중 하나면 메모리 압박 신호:
| 코드 | 의미 |
| -------------------- | ---------------------------- |
| `NODE_SCRIPT_OOM` | 스크립트가 16MB 초과 |
| `FLOW_MSG_TRIMMED` | 출력 페이로드 2MB 초과로 자동 잘림 |
| `ENGINE_GC_PRESSURE` | 엔진 전체 GC 빈도 급증 (대용량 페이로드 다발) |
운영 중 위 코드가 분당 5건 이상 나오면 즉시 가장 무거운 플로우의 스크립트를 점검하세요.
***
## 모바일 푸시 알림 채널
`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 인증 토큰](/plantpulse-platform/user/security.md#api-token) 에 자동 등록됩니다. 플로우에서는 토큰을 직접 다루지 않고 **사용자 ID** 만 지정하시면 됩니다.
* 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}"
```
### 다국어 푸시
```javascript
// 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} 만료 예정입니다."
```
### 자격 증명 보관 권장 위치
| 종류 | 권장 위치 |
| ------------- | -------------------------------------------------------------------- |
| 정적 (변경 거의 없음) | [시스템 → 설정](/plantpulse-platform/user/system.md) 의 자격 증명 저장소 |
| 동적 (자동 갱신) | 위 패턴으로 `flow_save_attributes` 사용해 자산 메타에 저장 |
| 사용자별 OAuth | [보안 → API 인증 토큰](/plantpulse-platform/user/security.md#api-token) 화면 |
> 자격 증명은 그래프 안에 평문으로 두지 마시고 반드시 외부 저장소 참조 (`${creds.x}`) 형태로 작성하세요. [내보내기/가져오기](#import-export) 시에도 평문 토큰이 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일입니다. 장기 보관이 필요하면 외부 로그 시스템으로 적재해 주세요.
***
## 일반 워크플로우
1. **새 플로우 생성** — 목록 화면 `새 플로우` 버튼 → 이름·설명 입력
2. **편집 화면 진입** — 자동으로 빈 캔버스가 열립니다
3. **트리거 노드 배치** — 좌측 팔레트에서 트리거 노드를 드래그
4. **처리 노드 추가** — 필터 → 변환 → 액션 순으로 배치하고 와이어로 연결
5. **노드 설정** — 각 노드를 클릭해 우측 인스펙터에서 옵션 입력
6. **저장** — 우상단 `저장` 버튼 (자동 스냅샷 적재)
7. **테스트 실행** — 임의 메시지를 주입해 결과를 확인
8. **배포** — `배포` 토글로 활성화 → 트리거 이벤트가 들어오면 자동 실행
9. **모니터링** — 라이브 디버그 패널과 실행 이력 화면으로 확인
***
## 예시 플로우
각 예시는 노드 결선 다이어그램 + 핵심 노드 설정 + 동작 설명으로 구성됩니다. 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`, `title_field`: `data.title` |
| `flow_send_email` | `to_field`: `metadata.alert_to`, `subject`: `[MES 동기화 실패] ${data.po}` |
**Script Transform 예시**
```js
data.master_id = 'WO-MES-' + data.po;
data.title = data.product_name + ' (' + data.qty + ')';
data.asset_id = data.line_id || 'UNASSIGNED';
data.due_date = data.delivery_date;
metadata.alert_to = 'ops@example.com';
msg
```
### 예시 2: 외부 웹훅으로 자산 명령 발행
외부 시스템이 보낸 HTTP 페이로드를 검증한 뒤 내부 자산 명령으로 변환합니다.
```
[flow_on_webhook]
│
▼ WEBHOOK
[flow_script_filter: payload 검증]
│
├── TRUE
▼
[flow_publish_asset_command]
│
├── SUCCESS → [flow_log]
└── FAILURE → [flow_webhook_callback: 외부에 실패 통보]
```
**호출 방법**: `POST /flow/webhook/{flow_id}` 로 JSON 본문을 전송하면 `data` 영역에 그대로 실린 채 트리거가 발화합니다.
**Script Filter 예시**
```js
// 인증 토큰 일치 + 필수 필드 존재 검사
if (data.token !== 'EXPECTED_TOKEN') return false;
if (!data.asset_id || !data.cmd_key) return false;
true
```
### 예시 3: 알람 → 긴급 작업지시 자동 발행
ERROR 등급 이상의 알람을 받으면 상위 자산을 대상으로 긴급 정비 워크오더를 발행합니다.
```
[flow_on_alarm]
│
▼ ALARM
[flow_script_filter: priority = ERROR/CRITICAL]
│
├── TRUE
▼
[flow_change_originator: Tag → 상위 Asset]
│
▼
[flow_create_work_order: 긴급 정비]
│
└── SUCCESS → [flow_send_email: 정비 담당자]
```
**Script Filter 예시**
```js
['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`로 라우팅되므로 별도의 Filter 없이 안전하게 결선할 수 있습니다.
### 예시 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 예시**
```js
// 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: 엣지 디바이스 가동]
```
**호출 본문 예시**
```json
{
"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 예시**
```js
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 | 한 메시지 처리 중 누적 노드 방문 수 제한 |
| **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_alarm` → `flow_script_filter` → `flow_send_email` + `flow_create_work_order` |
| **외부 DB 동기화** | `flow_jdbc_poll` → `flow_script_transform` → `flow_create_asset` |
| **웹훅 수신** | `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단계 — 격리 테스트
문제 노드를 새 임시 플로우에 단독으로 옮겨, **테스트 실행** 으로 한 메시지만 발화시켜 결과를 격리 검증합니다. 정상 동작하면 원본 플로우의 결선·이전 단계 메시지 형태가 의심됩니다.
### 6단계 — 카운터 초기화 후 재현
전체 카운트 초기화 후 한 메시지만 발화시켜 깨끗한 통계로 재현하면 문제가 더 분명해집니다.
***
## 자주 겪는 문제
| 증상 | 원인·조치 |
| --------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------ |
| 트리거가 발화하지 않는다 | 플로우가 **배포** 상태인지, 트리거 노드의 `*_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](#patterns) 참고) |
| `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. 플로우 하나에 트리거를 여러 개 둘 수 있나요?** 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. 같은 페이로드가 두 번 처리되는 일이 발생합니다. 어떻게 막나요?** A. 트리거 단계에서 `*_pattern` 으로 좁히거나, `flow_throttle`/`flow_debounce` 로 빈도를 제한하세요. 외부 입력(웹훅·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)
운영 중인 플로우를 변경할 때 권장 순서입니다.
1. **백업** — 편집 화면 `내보내기` 로 현재 그래프를 JSON 파일로 다운로드해 보관합니다 (자동 스냅샷도 적재되지만 외부 보관이 안전).
2. **복제 또는 새 플로우로 작업** — 운영 중인 플로우를 즉시 수정하기보다 새 플로우(또는 가져오기로 만든 사본)에서 변경 후 검증하세요.
3. **테스트 실행** — JSON 메시지를 직접 주입해 모든 분기(SUCCESS/FAILURE/EXHAUSTED 등)를 한 번씩 발화시켜 실행 이력에서 결과를 확인합니다.
4. **운영 반영** — 검증된 플로우의 그래프 JSON을 `내보내기` → 운영 환경에서 `가져오기` → 운영자가 검토 후 **배포** 토글.
5. **문제 발생 시 롤백** — 즉시 **해제** 토글로 비활성화. 자동 스냅샷이 적재되어 있으므로 개발자에게 의뢰하면 이전 버전으로 되돌릴 수 있습니다.
### 카운터·통계 활용
* 목록 화면의 처리 추세 스파크라인이 갑자기 평탄해진다면 트리거 미발화 또는 SKIPPED 처리 가능성을 확인하세요.
* 에러 비율 도넛 차트가 비정상이면 노드 설정의 `*_field` 경로를 의심해 보세요. 메시지 페이로드에 키가 없으면 자주 실패합니다.
* 운영 검증 후에는 **전체 카운트 초기화** 로 통계 윈도우를 리셋해 평상시 베이스라인을 다시 측정하세요.
***
## 긴급 대응 절차
운영 중 문제가 발생했을 때 신속하게 사용하실 단계별 절차입니다.
### 시나리오 1 — 특정 플로우가 폭주(메시지 폭증)
**증상**: 한 플로우의 실행 건수가 정상치 대비 수십\~수백 배 급증, 에러도 동반 증가
**조치**
1. 즉시 해당 플로우의 **해제** 토글로 비활성화 (목록 화면에서 1초)
2. 라이브 디버그 패널·실행 이력에서 어떤 트리거가 폭주를 일으켰는지 확인
3. 트리거 노드의 `*_pattern` 옵션을 좁히거나, 직후에 `flow_throttle` / `flow_debounce` 결선
4. 필요 시 `flow_check_existence_field` 또는 `flow_msg_type_filter` 로 메시지 타입 한정
5. 수정 후 **테스트 실행** 으로 재현 → 정상 확인 후 **배포** 재개
### 시나리오 2 — 외부 시스템 장애로 일괄 실패
**증상**: HTTP/Kafka/외부 DB 등 외부 IO 노드의 에러가 동시에 누적
**조치**
1. 영향 범위가 넓다면 관련 플로우들을 일괄 **해제** (목록 화면 일괄 선택 후 일괄 해제)
2. 외부 시스템 복구 확인
3. `flow_retry` 결선이 없는 외부 IO 노드라면 결선 추가
4. 외부 시스템 응답이 느려졌다면 `timeout_ms` 조정
5. 운영 환경에 따라 **모두 재배치** 후 **배포** 재개
### 시나리오 3 — 사이클로 인한 무한 루프
**증상**: 단일 메시지가 같은 노드를 계속 통과, 처리 시간 누적
**조치**
1. `max_revisit` 보호로 최대 3회까지만 재방문되어 자동 차단되지만, 운영 안전을 위해 **해제** 후 점검
2. 그래프에서 사이클을 시각적으로 추적 (편집 화면에서 와이어 따라가기)
3. 의도된 루프(`flow_retry`)면 `max_attempts` 가 적정한지 확인
4. 의도되지 않은 사이클이면 결선 제거 또는 `flow_check_relation` 으로 분기 추가
5. 수정 후 **전체 카운트 초기화** → 재배포
### 시나리오 4 — 데이터 손상 의심 (잘못된 자동 갱신)
**증상**: 자동화로 도메인 데이터가 의도와 다르게 갱신됨
**조치**
1. 즉시 해당 플로우 **해제**
2. 외부에 보관 중인 직전 그래프 JSON 백업 또는 자동 스냅샷에서 이전 버전 확인 (관리자 협조)
3. 그래프 분석: 의도하지 않은 Update 노드 결선·잘못된 `*_field` 경로·스크립트 변환 오류 확인
4. 영향 받은 도메인 데이터는 별도 절차로 백오피스 화면에서 정정
5. 수정한 그래프를 **테스트 실행** 으로 검증 후 재배포
### 시나리오 5 — 시스템 점검·유지보수 중 중지
**증상**: 외부 시스템 점검 시간 동안 외부 IO 호출을 잠시 멈추고 싶음
**조치**
1. 영향받는 플로우들을 일괄 **해제** (목록에서 다중 선택)
2. 점검 종료 후 **배포** 재개
3. 자체 스케줄러 보유 트리거(스케줄·외부 MQTT 구독·외부 DB 폴링)는 **모두 재배치** 1회 실행으로 정상 등록 확인
### 시나리오 6 — 라이브 디버그 적재 큐 가득참
**증상**: 운영 진단 페이지의 적재 큐 크기가 임계 근처, 드롭 건수 증가
**조치**
1. 디버그 모드가 켜진 플로우 수와 적재량을 줄이세요 (검증 끝난 플로우는 디버그 모드 OFF)
2. 트리거 단계의 `*_pattern` 으로 후속 처리량 자체를 줄이세요
3. 시스템이 자동 회복 모드로 들어가면 부하 워치독에 `DEGRADED`/`CRITICAL` 로그가 한 줄 남습니다 — 관리자에게 공유하세요
> 모든 시나리오에서 우선순위는 **즉시 해제 → 원인 파악 → 수정 → 검증 → 재배포** 순서입니다. 변경 절차([변경 안전 절차](#best-practices))도 함께 참고하세요.
***
## 보안·민감 정보 처리
플로우는 외부 API 호출·이메일/SMS 발송·웹훅 수신 등 민감한 입출력을 다루므로 다음 원칙을 지켜 주세요.
### 토큰·자격 증명
* API 토큰·비밀번호를 노드 설정에 **평문**으로 직접 입력하지 마세요. `${ENV_VAR}` 형태의 환경 변수 치환을 사용해 운영 환경의 비밀 저장소에서 주입받으세요.
* 운영 환경에서 토큰을 노출시키지 않도록 `headers` JSON 의 토큰 위치를 가능한 한 환경 변수로 분리합니다.
```json
// 권장
{ "Authorization": "Bearer ${MES_TOKEN}" }
// 비권장
{ "Authorization": "Bearer eyJhbGciOi..." }
```
### 웹훅 진입점 보호
`flow_on_webhook` 은 인증된 사용자가 호출하는 내부 진입점이지만, 외부에 노출하실 때는 페이로드 검증 필터를 반드시 결선하세요.
```js
// flow_script_filter — 토큰 + 필수 필드 검사
if (data.token !== '${WEBHOOK_TOKEN}') return false;
if (!data.asset_id || !data.cmd_key) return false;
true
```
### 민감 데이터 마스킹
라이브 디버그 패널과 실행 이력에는 메시지 미리보기가 표시됩니다. 개인정보(이름·연락처·계정)와 토큰은 적재 직전에 마스킹하세요.
```js
// 디버그/로그 적재 직전 변환
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` 는 응답 본문 전체가 적재됩니다. 응답에 민감 데이터가 포함된 경우 즉시 후속 변환 노드로 필요한 키만 추출하고 원본을 제거하세요.
```js
// 응답에서 필요한 필드만 보존
data = { id: data.response_obj.id, status: data.response_obj.status };
msg
```
### 스크립트 노드 격리
스크립트 노드는 안전한 샌드박스에서 실행되며 파일·네트워크·임의 클래스 접근이 차단됩니다. 외부 호출이 필요하면 반드시 별도의 외부 연동 노드를 결선하세요.
***
## 플로우 메트릭과 알람
플로우가 운영 중일 때 자동으로 수집되는 지표와 그 지표를 보는 곳·알람을 거는 방법.
### 플로우 단위 KPI
목록 화면(`/flow/index`) 의 각 행에 다음 KPI 가 표시됩니다.
| 지표 | 의미 | 비정상 신호 |
| --------------- | ----------------------------------- | ---------------------------- |
| **전체 실행 건수** | 트리거가 발화하여 그래프를 한 번 통과한 건수 (성공/실패 합) | 평소 대비 급감 → 트리거가 죽음 / 급증 → 폭주 |
| **에러 건수** | 그래프 중 어느 노드에서든 `FAILURE` 분기로 빠진 건수 | 전체 대비 5% 이상이면 점검 |
| **에러율** | `에러 / 전체 × 100` | 일정 임계 초과 시 알람 거세요 |
| **마지막 실행** | 가장 최근 발화 시각 | "5분 이상 발화 없음" 이 정상 인 경우만 OK |
| **평균 처리 시간** | 메시지 한 건이 그래프 전체를 통과하는 평균 ms | 외부 IO 노드 추가/제거 시 변화 큼 |
| **최근 처리 시간 추이** | 5분 스파크라인 — 라이브 디버그 패널 | spike 발생 시 외부 시스템 응답 점검 |
### 노드 단위 KPI
편집 화면에서 노드를 클릭하면 노드 우하단에 다음이 표시됩니다.
```
[Node Name]
처리 999 · 에러 3 · 평균 12ms · 최근 18ms
```
| 위치 | 표시 |
| --------- | ------------------------------- |
| **상단 점등** | 회색(대기) / 녹색(처리 중) / 빨강(에러) |
| **하단 라벨** | `처리 N · 에러 M · 평균 Xms · 최근 Yms` |
#### 노드 통계 4 종
| 카운터 | 의미 |
| ------------ | ------------------------- |
| **처리** | 그 노드로 들어와 정상 분기로 나간 메시지 수 |
| **에러** | `FAILURE` 분기 또는 예외 발생 수 |
| **평균 처리 시간** | 노드 자체 처리 ms (외부 IO 포함) |
| **최근 처리 시간** | 가장 마지막 한 건 처리 ms |
### 시스템 단위 메트릭
플로우 엔진 전체의 메트릭은 [시스템 → 모니터링](/plantpulse-platform/user/system.md) 화면에서 확인하실 수 있습니다.
| 메트릭 | 의미 | 정상 범위 |
| -------------------------------------------------- | ----------------------- | ------------------ |
| **flow\.engine.queue\_depth** | 트리거 처리 큐의 깊이 | < 100 (평시) |
| **flow\.engine.in\_flight** | 현재 처리 중인 메시지 수 | < `workers` 풀 크기 |
| **flow\.engine.dispatch\_lag\_ms** | 트리거 발화 \~ 첫 노드 진입까지 지연 | < 100ms |
| **flow\.engine.exec\_p50\_ms / p95\_ms / p99\_ms** | 처리 시간 백분위수 | p95 < 1초 (기준 워크로드) |
| **flow\.engine.failed\_per\_min** | 1분당 실패 메시지 수 | < 10 (배경 노이즈) |
| **flow\.engine.script\_timeout\_per\_min** | 스크립트 노드 타임아웃 (500ms 초과) | 0 이 정상 |
### 메트릭 기반 알람 등록 패턴
플로우 자체 메트릭을 [EQL 알람](/plantpulse-platform/user/alarm.md#config) 또는 [CEP → 트리거](/plantpulse-platform/user/cep.md#trigger) 로 감시하시는 패턴 4 종.
#### 패턴 A — 에러율 임계 초과
```sql
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 — 특정 플로우의 발화 끊김
```sql
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 타임아웃 폭증
```sql
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 |
| 자동 진단 (참고) | 진단 로그만 |
> ⚠️ 알람용 플로우 자체에는 절대 알람을 거는 메트릭에 의존하지 마세요 — 알람 플로우가 죽으면 알람 자체가 안 옵니다. 알람용 플로우는 [시스템 → 모니터링](/plantpulse-platform/user/system.md) 의 외부 헬스 체크로 감시하세요.
***
## 메시지 처리 의미론과 백압
플로우 엔진이 메시지를 처리할 때의 보장 수준과 백압(backpressure) 동작.
### 전달 보장 — At-Least-Once
플로우 엔진은 **at-least-once 전달**을 보장합니다.
| 케이스 | 동작 |
| ----------- | --------------------------------------- |
| 정상 처리 | 한 번 발화 → 한 번 그래프 통과 → 한 번 완료 |
| 처리 중 엔진 재시작 | 트리거 큐에 남은 메시지는 다음 부팅 후 다시 처리됨 |
| 노드 단위 예외 | `FAILURE` 분기로만 전이. 메시지가 사라지지 않음 |
| 외부 IO 타임아웃 | `FAILURE` 분기 + `flow_retry` 로 자동 재시도 가능 |
> **중복 가능성** — 정확히 한 번(exactly-once) 이 아닙니다. `flow_create_*` 가 중간에 끊겼다 재시도되면 같은 ID 가 두 번 들어올 수 있으니, `on_duplicate=skip` 또는 외부 시스템의 멱등성 키를 사용해 주세요.
### 멱등성 키 패턴
플로우에서 외부 시스템을 호출할 때 멱등성을 보장하는 패턴.
```javascript
// 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 의 이벤트는 서로 다른 워커에서 병렬 처리될 수 있습니다.
### 백압(Backpressure)
처리 속도가 발화 속도를 따라가지 못할 때의 동작.
| 상황 | 동작 |
| ------------------ | --------------------------------- |
| 큐 깊이 < 80% | 정상 — 신규 트리거 즉시 enqueue |
| 큐 깊이 80%\~100% | 진단 WARN 로그 발생 — 큐는 계속 수용 |
| 큐 깊이 = 100% (가득 참) | 신규 트리거를 **drop** + 진단 ERROR 로그 발생 |
#### 큐가 가득 찰 때 운영자가 할 일
1. [시스템 → 모니터링](/plantpulse-platform/user/system.md) 에서 `flow.engine.queue_depth` 확인
2. [목록 화면](#list) 에서 실행 건수가 폭증한 플로우 식별
3. 그 플로우의 트리거 사전 필터(`*_pattern`)를 좁혀 부하 감소
4. 또는 그 플로우의 배포를 일시 해제하여 비상 처리
### 서킷 브레이커 (외부 IO)
외부 IO 노드(HTTP/Kafka/MQTT/이메일)는 다음 조건이 충족되면 **30초 동안 일괄 차단**됩니다.
| 조건 | 임계 |
| ------------- | ----- |
| 최근 1분 내 연속 실패 | ≥ 10건 |
| 평균 응답 시간 | ≥ 10초 |
차단 동안 그 노드로 들어오는 모든 메시지는 즉시 `FAILURE` 로 분기됩니다. 외부 시스템이 응답을 회복하면 자동으로 차단 해제. 이 동작은 외부 장애가 플로우 엔진 자체를 마비시키지 않도록 격리하는 역할을 합니다.
> 서킷이 열린 시점은 진단 로그에 `code = CIRCUIT_OPENED` 로 기록됩니다. 외부 시스템이 빨리 회복했다면 [실행 이력](#log) 의 `metadata.retry_count` 를 보고 누락된 메시지를 수동 재실행할 수 있습니다.
### 사이클(순환) 방지
플로우 안에 다음 같은 사이클이 생기면 무한 루프가 발생할 수 있습니다.
```
A → B → C → A (잘못된 결선)
```
플로우 엔진은 두 가지 방어선으로 사이클을 차단합니다.
| 방어선 | 동작 |
| ------------------- | ------------------------------------ |
| **`max_depth=100`** | 노드 100 개를 거치면 강제 종료 |
| **`max_revisit=3`** | 같은 노드를 4번째 방문하는 순간 메시지 폐기 + 진단 ERROR |
> `flow_retry` 의 SUCCESS 분기 → 원래 노드 루프는 의도적 사이클이며, `metadata.retry_count` 가 함께 증가하므로 max\_revisit 이 발동하기 전에 정상 종료됩니다.
***
## 에러 코드 카탈로그
플로우 엔진에서 발생하는 진단 메시지의 `code` 필드값과 의미·대응.
### 트리거 단계
| 코드 | 의미 | 1차 대응 |
| ----------------------- | ---------------------------- | ------------------- |
| `TRIGGER_NO_MATCH` | 메시지가 트리거 패턴에 매칭되지 않아 SKIPPED | 정상 동작 — 카운트만 모니터링 |
| `TRIGGER_QUEUE_FULL` | 처리 큐 가득 참 → 신규 메시지 drop | 부하 평탄화 / 트리거 패턴 좁히기 |
| `TRIGGER_DISPATCH_FAIL` | 트리거 → 첫 노드 dispatch 중 예외 | 트리거 노드 설정 확인 + 재시작 |
### 노드 실행 단계
| 코드 | 의미 | 1차 대응 |
| --------------------- | ------------------------- | ------------------------------- |
| `NODE_TIMEOUT` | 노드 처리 시간 한도 초과 | `timeout_ms` 늘리기 / 외부 시스템 응답 점검 |
| `NODE_SCRIPT_ERROR` | 스크립트 노드 예외 | 라이브 디버그에서 스크립트 본문 확인 |
| `NODE_SCRIPT_TIMEOUT` | 스크립트 500ms 초과 | 스크립트 단순화 / 외부 호출은 별도 노드로 분리 |
| `NODE_FIELD_MISSING` | `*_field` 옵션의 경로가 메시지에 없음 | 트리거 페이로드 확인 / 정적 값 폴백 추가 |
| `NODE_CIRCUIT_OPENED` | 서킷 브레이커 열림 | 외부 시스템 회복 대기 / 30초 후 자동 닫힘 |
| `NODE_HTTP_4XX` | 외부 HTTP 4xx 응답 | 호출 본문/헤더/인증 확인 |
| `NODE_HTTP_5XX` | 외부 HTTP 5xx 응답 | 외부 시스템 점검 / `flow_retry` 추가 |
| `NODE_SSL_ERROR` | TLS 인증서 검증 실패 | `verify_ssl` 점검 / 인증서 갱신 |
### 그래프 단계
| 코드 | 의미 | 1차 대응 |
| ------------------- | ------------------ | -------------------------- |
| `FLOW_MAX_DEPTH` | 노드 100개 초과 방문 | 그래프 단순화 / 서브플로우 분할 |
| `FLOW_MAX_REVISIT` | 같은 노드 4번째 방문 | 사이클 결선 점검 |
| `FLOW_TIMEOUT` | 한 메시지 처리 30초 초과 | 외부 IO 노드의 타임아웃 점검 / 비동기 분리 |
| `FLOW_NOT_DEPLOYED` | 트리거는 발화했으나 플로우 미배포 | 배포 토글 ON |
### 외부 인입 단계
| 코드 | 의미 | 1차 대응 |
| ------------------------ | --------------------- | ------------------------------------------ |
| `WEBHOOK_AUTH_FAIL` | `X-API-Key` 헤더 불일치/누락 | 외부 시스템의 인증 헤더 점검 |
| `WEBHOOK_BODY_TOO_LARGE` | 본문 256KB 초과 | 외부 시스템에 페이로드 축소 요청 / `max_body_kb` 상향 |
| `MQTT_BROKER_DOWN` | 외부 MQTT 브로커 연결 실패 | 브로커 점검 / 자동 재연결 대기 (지수 백오프) |
| `JDBC_POLL_ERROR` | 외부 DB 폴링 SQL 실행 실패 | SQL 확인 / DB 자체 점검 / `on_error_continue` 확인 |
### 도메인 CRUD 단계
| 코드 | 의미 | 1차 대응 |
| ------------------------- | ---------------------------- | ----------------------------------------- |
| `DOMAIN_DUP_KEY` | 같은 ID 중복 등록 | \`on\_duplicate=skip |
| `DOMAIN_FK_VIOLATION` | 참조 무결성 위반 (없는 사이트·자산 등) | 부모 도메인 먼저 등록 |
| `DOMAIN_NOT_NULL` | NOT NULL 컬럼 누락 + 자동 보강 실패 | 입력값 점검 / 필드 매핑 추가 |
| `DOMAIN_STATE_TRANSITION` | 잘못된 상태 전이 (예: WAIT → END 직접) | 정상 전이 경로 확인 ([워크오더 상태 전이](#actions-crud)) |
> 모든 에러 코드는 진단 로그(`Log` 스트림)의 `data.code` 필드로 검색할 수 있습니다. 운영 화면에서 임계 초과를 자동 감지하려면 위 [메트릭 기반 알람 패턴 D](#metrics) 를 활용하세요.
***
## 클러스터·고가용성(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 보장](#semantics) 으로 외부 시스템에 멱등성 키 필요 |
### 클러스터 배포 시 운영자 체크리스트
1. 모든 노드의 시간 동기화 (NTP) — 트리거 발화 시각이 노드 간 일치
2. 외부 시스템(MQTT/Kafka 브로커)을 모든 노드가 도달할 수 있는 네트워크 위치에 배치
3. `flow_send_email` 의 SMTP 설정은 시스템 설정에 한 번만 등록 — 모든 노드가 공유
4. 워커 풀 크기(`flow.engine.workers`)는 노드별 CPU 코어 수에 맞게 조정
### 단일 노드 모드 (개발·소규모)
* 리더·워커·스케줄러 모두 한 프로세스에서 동작
* 페일오버 없음 — 노드가 다운되면 플로우 자체가 중단
* 트리거 큐는 인-메모리 + 디스크로 보장되어 재시작 후 복구
***
## 종단간 트레이스
한 메시지가 그래프 전체를 어떻게 통과했는지 사후 추적하는 방법.
### 자동 상관 ID 부여
플로우 엔진은 모든 트리거 발화 메시지에 **상관 ID(correlation ID)** 를 자동으로 부여합니다.
```json
{
"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 로 검색
[실행 이력 화면](#log) 에서 `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일 (실행 이력과 동일) |
| 외부 적재 (장기 보관) | [감사·이력 추적](#audit) 의 외부 시스템 미러링 사용 |
> 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 + 슬랙 + 워크오더 생성 |
> 같은 메시지의 복제본이 여러 노드로 분배됩니다. 각 분기는 독립적으로 처리되며, 한 분기 실패가 다른 분기에 영향 없습니다.
### 패턴 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 | 그래프 전체 시퀀스·분기 |
| **③ 시뮬레이션** | 배포 + 실제 트리거 대기 (스테이징 환경) | 실 메시지 흐름·외부 시스템과의 결합 |
### 테스트 실행 페이로드 라이브러리
[테스트 실행](#test-run) 다이얼로그에 붙여넣을 수 있는 트리거별 페이로드 예제는 [트리거별 페이로드 예제](#trigger-payloads) 섹션 참고. 운영 환경에서 자주 발생하는 경계 케이스도 미리 만들어 두세요.
#### 경계 케이스 예시
| 케이스 | 페이로드 |
| ----------- | -------------------------------------- |
| **null 필드** | `{"value": null}` — 스크립트가 null 안전한지 |
| **빈 문자열** | `{"value": ""}` — 빈 문자열을 0 으로 해석하는지 검증 |
| **음수** | `{"value": -1}` — 임계 검사가 ± 부호 의도대로 |
| **거대 숫자** | `{"value": 1e20}` — 오버플로/정밀도 |
| **유니코드** | `{"name": "한글-Émoji-🚀"}` — 외부 시스템 인코딩 |
### 변경 안전 절차
```
1. 기존 플로우를 [내보내기] (JSON 파일 저장)
2. 새 플로우를 사본으로 만듦 (이름 끝에 `_v2`)
3. 사본의 트리거 패턴을 좁혀 일부 자산만 매칭 (예: TEST-* 사이트만)
4. 신구 동시 배포 — 새 버전 데이터 확인
5. 1주일 안정성 검증 후 신 버전을 전체 트리거 패턴으로 변경
6. 구 버전 해제 + 보관
```
### 회귀 테스트 시나리오 보관
자주 사용되는 시나리오를 텍스트 파일로 보관하시고, 변경 후 반드시 재실행하시기를 권장합니다.
```
# 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 를 로** | 실수 발송 방지 |
### 카운터 초기화 후 부하 테스트
1. 신규 버전 플로우의 [카운트 초기화](#카운트-초기화) (전체)
2. 5분 동안 정상 트리거 발화 시키기
3. [목록 화면](#list) 에서 처리/에러 카운트, 평균 처리 시간 확인
4. p95 처리 시간이 기존 대비 20% 이상 늘면 원인 분석 후 롤백 검토
***
## 성능 한계와 튜닝
플로우 엔진의 처리 한계와 튜닝 팁입니다.
### 기본 한계
| 항목 | 기본값 | 비고 |
| ----------------------------- | ---------------------- | ------------------------------ |
| 한 메시지의 노드 방문 수 (`max_depth`) | 100 | 노드 100개를 거치는 동안 종료되지 않으면 강제 중단 |
| 같은 노드 재방문 횟수 (`max_revisit`) | 3 | 사이클 무한 루프 방지 |
| 플로우 처리 시간 (`flow_timeout_ms`) | 30,000ms | 한 메시지 처리에 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_alarm]
│
▼
[flow_kafka_publish: topic=audit.alarm.events]
```
***
## 신규 플로우 배포 체크리스트
운영 환경에 새 플로우를 배포하기 전 확인할 항목입니다.
**그래프 구조**
* [ ] 트리거 노드가 정확히 하나(또는 의도된 다수) 결선되었는가
* [ ] 모든 액션 노드의 `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 의 핵심은 **두 버전이 동시에 배포된 상태에서 즉시 전환** — 문제 발견 시 v1 을 즉시 다시 배포해 롤백.
#### v2 격리 결선 패턴
```
[trigger 모든 메시지]
↓
[flow_script_filter — metadata.test_mode === true]
↓ TRUE
[새 로직]
```
테스트용 메시지는 [/flow/{id}/run](#rest-api) API 로 `metadata.test_mode=true` 를 명시해 보냅니다.
### 전략 3 — A/B 테스트 (성능·결과 비교)
두 버전이 같은 메시지를 받아 서로 다른 액션 후 결과를 비교.
```
[trigger 메시지]
↓
[flow_split (메시지 복제)]
├ A 경로 → 기존 v1 액션 → [flow_save_attributes target=stats_v1]
└ B 경로 → 새 v2 액션 → [flow_save_attributes target=stats_v2]
```
이후 [일일 통계](/plantpulse-platform/user/statistics.md) 화면에서 `stats_v1` vs `stats_v2` 누적 결과 비교.
#### A/B 비교 자동 분석
```sql
-- 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** — 즉시 롤백 가능 |
| 두 알고리즘 중 어느 게 더 나은지 측정 | **A/B 테스트** |
| 단순 옵션값 조정 | 직접 변경 — `metadata.audit_diff` 메모 후 1주 모니터링 |
### 롤백 절차 (공통)
문제 발견 시 즉시 롤백:
1. [목록 화면](#list) 에서 새 버전 **해제** 토글
2. (Canary/A·B 의 경우) 트리거 패턴을 0건 매칭으로 변경
3. 기존 버전이 단독으로 동작하는지 5분 모니터링
4. 진단 로그에서 `code=FLOW_NOT_DEPLOYED` 가 없는지 확인
5. 원인 분석 — [실행 이력](#log) 에서 `trace_id` 로 실패 메시지 추적
### 출시 사전 체크리스트 (압축판)
[신규 플로우 배포 체크리스트](#checklist) 의 핵심만 한 화면:
```
[ ] 테스트 실행으로 정상·경계·실패 시나리오 모두 통과
[ ] 외부 IO 노드에 timeout_ms / 재시도 정책 설정됨
[ ] 자격 증명은 ${creds.*} 참조 (평문 미포함)
[ ] 트리거 패턴이 의도한 자산만 매칭
[ ] 영향받는 도메인 (자산/태그/주문) 식별됨
[ ] 운영 시간(특히 야간) 영향 검토됨
[ ] 롤백 시점·기준·담당자 결정됨
[ ] [감사 로그](#audit) 에 변경 의도 메모 작성됨
```
***
## 노드 빠른 설정 레퍼런스
운영 중 자주 쓰는 노드의 핵심 설정만 모아 놓은 치트 시트입니다.
### 트리거 빠른 설정
| 노드 | 핵심 옵션 |
| ------------------------ | --------------------------------------------------------- |
| `flow_schedule` | `cron` (예: `0 */5 * * * ?` = 5분마다), 또는 `interval_ms` |
| `flow_on_webhook` | 옵션 없음 — 외부에서 `POST /flow/webhook/{flow_id}` 로 발화 |
| `flow_on_mqtt_subscribe` | `broker_url`, `topic`, `client_id`, `username`/`password` |
| `flow_jdbc_poll` | `dsn`, `sql`, `interval_ms` |
| `flow_on_*` (도메인) | `*_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](#actions-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 인증 토큰](/plantpulse-platform/user/security.md#api-token) 에서 발급한 토큰을 헤더로 첨부합니다.
```
Authorization: Bearer {api_token}
```
### 엔드포인트 목록
#### 1) 플로우 목록 조회
```
GET /flow/list
```
```bash
curl -H "Authorization: Bearer ${TOKEN}" \
https://platform.example.com/flow/list
```
응답:
```json
{
"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` 필드를 변경하면 자동 배포·해제됩니다.
```bash
# 배포
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 — 그래프 백업
```bash
# 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 /flow/webhook/{flow_id}
Authorization: Bearer {token} 또는 X-API-Key: {token}
Content-Type: application/json
{
"order_no": "PO-001",
"customer": "ACME",
"quantity": 1000
}
```
응답:
```json
{ "status": "ACCEPTED", "trace_id": "tr-..." }
```
| 응답 코드 | 의미 |
| ------- | -------------------------------- |
| **200** | 메시지 큐에 enqueue 됨 (실제 처리 결과는 비동기) |
| **401** | 인증 토큰 잘못/누락 |
| **404** | 플로우 ID 없거나 미배포 |
| **413** | 본문 256KB 초과 |
| **422** | 트리거 노드가 `flow_on_webhook` 이 아님 |
| **429** | 분당 호출 한도 초과 |
### 외부 자동화 도구 연동 예시
#### GitHub Actions — PR 머지 시 플로우 배포
```yaml
- 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 — 플로우 배치 관리
```yaml
- 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
```groovy
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 산업 통합 패턴
산업 현장 특유의 통합 시나리오 모음. [엣지 노드 11종](#edge-nodes) 과 [자산 이벤트 발행 4종](#actions-asset) 의 조합 활용.
### 패턴 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 — 알람밴드 자동 조정
[예측 분석](/plantpulse-platform/user/forecast.md) 의 학습된 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 점유율 급증 위험 |
| 엣지 디바이스의 도커 컨테이너 자원 한도 | 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` 노드를 두 개 연결하여 첫 노드 응답 헤더의 토큰을 두 번째 노드의 헤더로 전파하세요.
### MES (외부 DB 직접 INSERT) — `flow_jdbc_query`
외부 MES 시스템의 작업지시 테이블에 직접 행 삽입.
```sql
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` 를 다음 노드에서 활용:
```javascript
// 응답에서 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` 단축 별칭을 사용합니다.
### 단위 변환
```javascript
// 섭씨 → 화씨
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
```
### 날짜·시각 변환
```javascript
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
```
### 문자열 정규화
```javascript
// 공백 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
```
### 중첩 객체 평탄화
```javascript
// 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
```
### 평탄 객체 → 중첩
```javascript
// 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 행 생성
```javascript
// 외부 시스템에 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
```
### 배열 집계
```javascript
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
```
### 조건부 필드 추가 (스키마 진화)
```javascript
// 알람 우선순위에 따라 색상·아이콘 자동 부여
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
```
### 페이로드 일부 마스킹
```javascript
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) 템플릿
```javascript
// 메시지 본문을 사용자 로케일에 따라 분기
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 안전 조회
```javascript
// 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` 후처리)
```javascript
// 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` 단축 별칭을 사용합니다.
### 변환 스크립트
#### 단위 변환·라벨링
```js
// 섭씨 → 화씨 + 등급 라벨
data.temp_f = data.temp * 1.8 + 32;
data.grade = data.temp > 80 ? 'HIGH' : data.temp < 0 ? 'LOW' : 'OK';
msg
```
#### 시각·시프트·요일 메타 보강
```js
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σ 알람밴드 계산
```js
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 → 작업지시 매핑
```js
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
```
#### 외부 페이로드 정규화 (다양한 필드명 통일)
```js
// 외부 시스템에 따라 키 이름이 다른 경우 — 한 줄로 정규화
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 패턴)
```js
// 자산 attribute 와 함께 사용 — 상태는 메시지 자체에 적재
data.consecutive_fail = (data.consecutive_fail ?? 0) + (data.passed ? 0 : 1);
data.alert = data.consecutive_fail >= 5;
msg
```
#### 메시지 통계 요약 (Merge 결과 가공)
```js
// 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
```
#### 시간대별 임계값 적용
```js
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
```
### 필터 스크립트
#### 우선순위 화이트리스트
```js
['ERROR', 'CRITICAL'].includes(data.priority)
```
#### 시간 윈도우 (주간만)
```js
const h = new Date(metadata.ts).getHours();
h >= 8 && h < 20
```
#### 자산·태그 ID 패턴
```js
/^MOTOR-.*$/.test(originator.id)
// 또는
metadata.tag_id && metadata.tag_id.startsWith('LINE-A.')
```
#### 임계값 + 안정성 (연속 N회)
```js
data.value > 100 && (data.consecutive_count ?? 0) >= 5
```
#### 영업시간·휴일 체크
```js
const d = new Date(metadata.ts);
const h = d.getHours();
const wd = d.getDay();
// 평일 09-18시만 통과
wd >= 1 && wd <= 5 && h >= 9 && h < 18
```
#### 특정 사이트만
```js
['SITE-01', 'SITE-02'].includes(metadata.site_id)
```
#### 필수 필드 모두 존재 검사
```js
data.value != null && data.tag_id && metadata.ts
```
#### 자기 발행 메시지 차단 (양방향 동기화)
```js
metadata.source !== 'flow'
```
### 스위치 케이스 스크립트
#### 우선순위 등급별
```js
// case "Critical"
['CRITICAL', 'EMERGENCY'].includes(data.priority)
// case "High"
data.priority === 'ERROR'
// case "Normal"
['WARN', 'INFO'].includes(data.priority)
```
#### 자산 상태별
```js
// case "Running"
data.status === 'RUN'
// case "Stopped"
['STOP', 'IDLE', 'PAUSED'].includes(data.status)
// case "Faulted"
data.status === 'FAULT' || data.error_count > 0
```
#### 작업지시 라이프사이클별
```js
// 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 시점을 기준으로 합니다.
***
## 관련 화면
* [CEP (복합 이벤트 처리)](/plantpulse-platform/user/cep.md): EQL 기반의 도메인 이벤트 룰 엔진
* [알람](/plantpulse-platform/user/alarm.md): 알람 발생·이력
* [데이터 포인트](/plantpulse-platform/user/data-point.md): 태그 데이터 조회·분석
* [개발자 가이드: 플로우 엔진](/plantpulse-platform/developer/flow.md): 엔진 아키텍처·노드 인터페이스·DB 스키마