찾는 정보	섹션
노드 ID 와 한 줄 설명	노드 카탈로그
노드 옵션 기본값	노드 옵션 상세 레퍼런스
메시지 JSON 예제	트리거별 페이로드 예제
그대로 쓸 스크립트	재사용 스크립트 모음 · 데이터 변환 Cookbook
API 호출 curl	플로우 REST API
실행 이력 해석	실행 이력에서 무엇을 보게 되나
문제 해결 가이드	단계별 디버깅 가이드 · 자주 겪는 문제
빠른 옵션 표	노드 빠른 설정 레퍼런스

모든 섹션 링크는 동일 문서 내 앵커입니다. Ctrl+F 로 키워드 검색도 효과적입니다.

핵심 개념

용어	설명
플로우 (Flow)	노드와 와이어(관계)의 집합으로 구성된 하나의 자동화 워크플로우입니다. 진입점이 있는 방향 그래프이며, 배포/해제 토글로 활성화 여부를 제어합니다
노드 (Node)	메시지를 받아 처리한 뒤 다음 노드로 전달하는 단위입니다. 7개 카테고리(트리거·필터·변환·액션·외부 연동·흐름 제어·엣지)로 구분됩니다
관계 (Relation)	노드 출력에서 다음 노드로 가는 와이어의 레이블입니다. 노드가 직접 `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` 등의 라벨을 지정해 분기합니다. 모든 라벨은 대문자로 표준화되어 있습니다
메시지 (Message)	플로우 내부를 흐르는 페이로드입니다. `type`(분류), `originator`(주체 엔티티), `data`(본문), `metadata`(컨텍스트)를 포함합니다
트리거 (Trigger)	플로우의 시작점이 되는 노드입니다. 도메인 이벤트(태그 포인트·알람·자산 이벤트 등), 외부 진입(웹훅·MQTT·외부 DB), 시간(스케줄) 세 종류가 있습니다

처음 시작하기

플로우는 다음 4단계로 가장 빨리 익숙해지실 수 있습니다.

목록 화면에서 새 플로우 버튼을 눌러 빈 플로우를 만듭니다 (이름과 설명만 입력).
편집 화면에서 좌측 팔레트의 트리거 노드(예: flow_on_tag_alarm) → 필터 → 액션(예: flow_send_email)을 차례로 드래그해 배치하고, 노드 사이를 와이어로 연결합니다.
각 노드를 클릭해 우측 인스펙터에서 옵션을 입력한 뒤 우상단 저장 → 테스트 실행으로 동작을 한 차례 확인합니다.
상단 배포 토글을 켜면 트리거 이벤트가 들어올 때마다 플로우가 자동 실행되며, 라이브 디버그 패널과 실행 이력 화면에서 결과를 확인하실 수 있습니다.

자동화 시나리오의 결선 패턴은 본 문서의 예시 플로우와 활용 예시를 참고해 주세요. 실제 플로우를 처음부터 끝까지 만드는 단계별 튜토리얼은 엔드 투 엔드 튜토리얼에서 따라하실 수 있습니다.

엔드 투 엔드 튜토리얼

운영 환경에서 실제로 사용 가능한 자동화 플로우 하나를 처음부터 끝까지 만들어 보는 단계별 튜토리얼입니다. 시나리오: 모터의 온도가 80°C 를 넘으면 자동으로 긴급 정비 작업지시를 발행하고 담당자에게 이메일로 알리기.

단계 1 — 새 플로우 만들기

왼쪽 메뉴에서 Automation > 플로우 클릭 → 목록 화면 열기
상단 우측 새 플로우 버튼 클릭
아래 정보를 입력하고 확인

항목	입력 값
이름	`모터 과열 자동 정비 발행`
설명	`[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com`

플로우가 생성되면 자동으로 빈 캔버스의 편집 화면이 열립니다.

단계 2 — 트리거 노드 배치

자산 텔레메트리 이벤트로부터 모터 온도 데이터를 받습니다.

좌측 팔레트의 트리거 (Trigger) 카테고리를 펼침
flow_on_tag_point 노드를 캔버스에 드래그
노드 클릭 → 우측 인스펙터에서 다음 옵션 입력

옵션	값
표시명	`태그 포인트 인입`
`tag_id_pattern`	`MOTOR-*.TEMP`

tag_id_pattern 으로 모터 온도 태그만 통과시키면 후속 처리량이 크게 줄어듭니다. 다른 태그는 SKIPPED 처리되어 카운터에 포함되지 않습니다.

단계 3 — 임계값 필터

80°C 초과 메시지만 통과시키도록 스크립트 필터를 추가합니다.

필터 (Filter) 카테고리에서 flow_script_filter 드래그
트리거 노드의 출력 포트에서 새 필터 노드의 입력 포트로 와이어 연결
인스펙터에 다음 입력

옵션	값
표시명	`임계값 필터 (80°C 초과)`
`language`	`JS`
`script`	`data.value > 80`

단계 4 — 도메인 변경 (태그 → 자산)

알람·작업지시는 자산 단위로 발행하는 게 자연스러우므로, 메시지의 originator 를 태그에서 상위 자산으로 변환합니다.

변환 (Transform) 카테고리에서 flow_change_originator 드래그
필터 노드의 TRUE 출력에서 와이어 연결
인스펙터:

옵션	값
표시명	`Tag → Asset 변경`
`entity_type`	`Asset`
`id_field`	`metadata.asset_id`

metadata.asset_id 는 태그 포인트 인입 시 자동 채워집니다. 만약 메시지에 없으면 태그 ID 의 prefix 부분(예: MOTOR-001.TEMP → MOTOR-001)을 스크립트로 추출하셔도 됩니다.

단계 5 — 작업지시 발행

긴급 정비 작업지시를 자동 생성합니다.

액션 (Action) — 도메인 CRUD 카테고리에서 flow_create_work_order 드래그
변환 노드의 SUCCESS 출력에서 와이어 연결
인스펙터:

옵션	값
표시명	`긴급 정비 작업지시 생성`
`asset_id_field`	`originator.id`
`title_field`	(정적 값) `긴급 점검 — 모터 과열`
`master_id_field`	(선택) `data.master_id` (없으면 자동 생성)
`description_field`	(정적 값) `자동 발행: 임계 온도 초과로 긴급 점검 필요`
`default_priority`	`HIGH`

단계 6 — 이메일 알림 (성공 분기)

작업지시 발행이 성공하면 담당자에게 알립니다.

외부 연동 (External) 카테고리에서 flow_send_email 드래그
작업지시 노드의 SUCCESS 출력에서 와이어 연결
인스펙터:

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

단계 7 — 실패 분기 처리

작업지시 발행 자체가 실패할 수도 있습니다(예: 자산이 사라졌거나 권한 문제). 운영자에게 즉시 푸시 알림을 보냅니다.

외부 연동 의 flow_send_push 드래그
작업지시 노드의 FAILURE 출력에서 와이어 연결
인스펙터:

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

단계 8 — 저장과 테스트 실행

우상단 저장 버튼 클릭. 자동 스냅샷이 적재되어 이후 롤백할 수 있습니다.
우상단 테스트 실행 버튼 클릭 → JSON 편집기에 다음 입력 → 발행

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

다이얼로그 하단에 ✓ JSON OK (type=POST_TELEMETRY) 표시 → 발행 클릭.

라이브 디버그 패널에서 다음을 확인:
- 트리거 노드 점등 녹색 → 메시지 통과
- 필터 노드: 92.5 > 80 이므로 TRUE 분기
- 변환 노드: originator 가 Asset/MOTOR-001 으로 변경
- 작업지시 노드: SUCCESS, data.work_order_id 자동 부여
- 이메일 노드: 발송 시도

단계 9 — 검증과 배포

작업지시 화면에서 새 작업지시가 등록되었는지 확인
이메일이 정상 도착했는지 확인 (테스트 환경의 메일함)
80°C 미만 메시지로 한 번 더 테스트 (필터에서 차단되어야 함):

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

필터 노드의 분기는 FALSE, 후속 노드 회색 — 정상.

모든 분기를 검증한 뒤 전체 카운트 초기화 로 통계 윈도우를 리셋
상단 배포 토글 ON

이제 실제 운영 환경에서 모터 온도가 80°C 를 넘는 즉시 자동으로 정비 작업지시가 발행되고 담당자에게 이메일이 갑니다.

단계 10 — 운영 모니터링

배포 직후 5~10분간 다음을 확인하시면 안전합니다.

위치	확인 항목
목록 화면	해당 플로우 행의 실행 건수가 정상 범위에서 증가하는지 (폭주 아닌지)
라이브 디버그 패널	노드 실패가 없는지
실행 이력 화면	실패한 메시지가 있다면 `NODE_ERROR` 의 원인 확인
받는 이메일	의도하지 않은 빈도로 알림이 가지 않는지

다음 단계

이 플로우를 발전시키시려면:

Retry 결선 추가 — 이메일/푸시 발송 실패 시 백오프 재시도 (예시 8 참고)
밴드 자동 조정 — 6시간 평균 ± 3σ 로 임계값 자동 보정 (예시 6)
다운타임 누적 — 과열 이력을 자산 attribute 에 누적 (예시 14)
품질 라인 격리 — 과열이 연속 발생하면 라인 자동 정지 (예시 15)

화면 구성

플로우는 다음 세 화면으로 구성됩니다.

화면	용도
목록	등록된 플로우 일람·검색·일괄 배포·가져오기
편집	시각 캔버스로 노드를 배치·연결·설정
실행 이력	노드 단위 실행 로그·타임라인 조회

목록 화면

상단 검색·생성 영역과 플로우 일람 테이블로 구성됩니다.

상단 도구

항목	설명
상태 필터	`전체` / `배포` / `해제`
이름·설명 검색	키워드로 플로우를 필터링합니다
새 플로우	빈 플로우를 생성합니다 (이름·설명 입력)
가져오기	내보내기로 받은 JSON을 업로드해 플로우를 복원합니다
모두 재배치	활성화된 모든 플로우를 한 번에 다시 적재합니다
새로고침	목록을 다시 불러옵니다

일람 테이블

목록 테이블은 25행 단위로 페이지네이션되며, 각 행에 처리 추세 스파크라인과 에러 비율 도넛 차트가 함께 표시됩니다. 페이지를 넘겨도 차트 상태가 유지됩니다.

컬럼	설명
선택	일괄 배포·해제용 체크박스
상태	`배포` / `해제` 뱃지
플로우 ID	`FLOW_NNNNN` 형식의 시퀀스 ID
이름 \| 설명	운영자가 지정한 메타 정보
노드	포함된 노드 개수
트리거	트리거 노드 개수
실행 건수	누적 메시지 처리 카운트
처리 시간	노드 평균/최근 처리 시간
에러	누적 에러 카운트
최종 수정	그래프 마지막 저장 시각
액션	`편집` / `삭제` 버튼

일괄 배포·해제

선택한 플로우들을 한꺼번에 배포(활성화) 또는 해제(비활성화)합니다. 배포된 플로우만 트리거 이벤트를 받습니다.

모두 재배치

상단 모두 재배치 버튼은 활성화된 모든 플로우를 다시 적재합니다. 다음 상황에서 사용하세요.

외부에서 그래프를 일괄 가져오기 한 직후
스케줄·외부 MQTT 구독·외부 DB 폴링 등 자체 스케줄러가 있는 트리거를 다시 등록하고 싶을 때
운영 중 캐시 일관성에 문제가 의심될 때

편집 화면 (시각 캔버스)

캔버스 좌측 팔레트에서 노드를 드래그해 배치한 후, 노드의 출력 포트를 클릭&드래그하여 다음 노드와 연결합니다.

상단 툴바

버튼	동작
이름·설명	플로우 메타 정보를 편집합니다
배포/해제	현재 플로우를 즉시 활성/비활성화합니다
내보내기	그래프 전체를 JSON 파일로 다운로드합니다
테스트 실행	임의의 JSON 메시지를 주입해 한 차례 실행하며 결과를 확인합니다 (아래 테스트 실행 사용법 참고)
저장	현재 그래프를 서버에 저장합니다. 저장 시 자동 스냅샷이 적재되어 이후 롤백할 수 있습니다

트리거가 없는 그래프를 저장하려고 하면 "수동 테스트 실행으로만 시작됩니다" 경고가 표시됩니다. 트리거를 의도적으로 빼고 수동 실행 전용 플로우로 사용하실 수 있습니다.

테스트 실행 사용법

테스트 실행 버튼을 누르면 JSON 편집 다이얼로그가 열립니다. 운영자가 직접 메시지를 작성해 한 차례 발행할 수 있어, 트리거 이벤트를 기다리지 않고 그래프 동작을 검증할 수 있습니다.

항목	설명
편집기	줄 번호·문법 강조가 있는 JSON 에디터. 메시지 본문을 자유롭게 작성합니다
검증 표시	`type` 필수 필드가 있으면 `✓ JSON OK (type=X)`, 누락이면 `⚠ type 필수` 표시
발행	`발행` 버튼으로 메시지를 디스패처에 주입. 결과는 라이브 디버그 패널과 실행 이력에서 확인

기본 템플릿 예시

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

저장하지 않은 변경이 있을 때 테스트 실행을 누르면 "서버는 저장된 버전을 실행합니다" 안내가 뜹니다. 변경을 검증하려면 먼저 저장하세요.

좌측 — 노드 팔레트

카테고리별로 접고 펼칠 수 있으며, 검색 입력으로 즉시 필터링이 가능합니다.

카테고리	색상	노드 수
트리거 (Trigger)	회색	22종
필터 (Filter)	파랑	6종
변환 (Transform)	녹색	8종
액션 (Action) — 통합·저장·자산 발행·명령 호출	주황	7종
액션 (Action) — 도메인 CRUD	주황	34종
외부 연동 (External)	보라	9종
흐름 제어 (Control)	회색	7종
엣지 (Edge)	청록	22종

중앙 — 캔버스

도구	단축키 / 조작	동작
확대/축소	`Ctrl/⌘ +` / `Ctrl/⌘ -` · 마우스 휠	캔버스 줌
100%	`Ctrl/⌘ 0`	줌 리셋
화면 맞춤	`Ctrl/⌘ 1`	모든 노드가 보이도록 자동 맞춤
선택 노드 삭제	`Del` / `Backspace`	선택한 노드/와이어 삭제
캔버스 이동 (panning)	빈 영역 좌클릭 후 드래그	노드를 클릭하지 않은 상태에서 빈 캔버스 배경을 잡고 끌면 캔버스 전체가 따라 움직입니다

캔버스 우하단에는 미니맵이 표시되며, 미니맵을 클릭하면 해당 위치로 즉시 이동할 수 있습니다.

💡 panning 사용 팁: 노드 위에서 드래그하면 노드가 이동합니다 — 캔버스를 이동시키려면 반드시 노드/와이어가 없는 빈 배경 영역을 잡으세요. 큰 플로우에서는 미니맵보다 panning 이 더 빠릅니다.

우측 — 노드 설정

캔버스에서 노드를 클릭하면 우측 인스펙터에 해당 노드의 설정 폼이 표시됩니다. 입력 필드는 노드 종류에 따라 자동 생성됩니다.

입력 방식	설명
정적 값	폼에 직접 입력한 값을 그대로 사용합니다
*`_field` 동적 값**	메시지 페이로드의 경로(예: `data.tag_id`, `metadata.site_id`)에서 값을 추출합니다. 값이 없으면 정적 값으로 폴백합니다

스크립트 노드(필터·변환·스위치)는 인스펙터 안에서 직접 코드 에디터로 편집할 수 있으며, 별도 다이얼로그로 확장하여 큰 화면에서 작성할 수도 있습니다.

코드 에디터 폰트는 가독성 강화된 모노스페이스 폰트 스택(Cascadia Code · JetBrains Mono · Consolas · Menlo 우선)으로 표시되며, 한글 코멘트도 안정적으로 정렬됩니다.

카운트 초기화

노드 설정 패널 우상단에는 두 개의 초기화 버튼이 아이콘으로 표시됩니다. 각 버튼에 마우스를 올리면 툴팁이 안내합니다.

아이콘 버튼	동작
🩹 (반창고) — 에러 카운트 초기화	이 플로우 내 모든 노드의 누적 에러 카운트만 0으로 리셋합니다
🔄 (회전 화살표) — 전체 카운트 초기화	처리·에러·처리 시간 모두 + 플로우 단위 통계를 모두 0으로 되돌립니다. 운영 검증을 끝내고 새로 통계를 시작할 때 사용하세요

두 버튼 모두 확인 다이얼로그 없이 즉시 적용됩니다 — 통계만 초기화될 뿐 노드 동작이나 메시지 처리에는 영향이 없습니다.

우측 — 라이브 디버그

라이브 디버그 패널이 인스펙터 아래에 표시됩니다.

항목	설명
갱신 주기	2초
레벨 컬러	좌측 보더에 INFO(파랑)/WARN(노랑)/ERROR(빨강) 표시
표시 정보	노드 표시명 · 처리 시간(ms) · 메시지 미리보기
일시정지	패널 우상단 토글로 갱신을 일시정지합니다
비우기	누적된 디버그 항목을 화면에서만 비웁니다

캔버스의 노드 우상단에는 작은 점등이 표시됩니다.

색상	의미
회색	대기 — 메시지를 받지 않은 상태
녹색	메시지가 통과 중
빨강	처리 중 에러 발생

노드 우하단에는 평균 X · 최근 Y 형태로 처리 시간이 표시됩니다.

메시지 구조

플로우 내부를 흐르는 메시지는 다음 4개 영역으로 구성됩니다.

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

영역	의미
`type`	메시지 분류. 필터 노드의 분기 기준
`originator`	메시지 주체 엔티티 (어떤 자산/태그/주문에 대한 것인가)
`data`	페이로드 본문
`metadata`	컨텍스트 (시각·사이트·시프트·태그 ID 등)

메시지 타입

타입	발생 진입점
`POST_TELEMETRY` / `TAG_POINT`	태그 포인트 인입
`POST_ATTRIBUTES`	태그/자산 메타 갱신
`TAG_ALARM`	태그 단위 알람
`ENTITY_CREATED` / `UPDATED` / `DELETED`	엔티티 라이프사이클 이벤트
`ASSET_DATA` / `ASSET_EVENT` / `ASSET_ALARM` / `ASSET_COMMAND` / `ASSET_AGGREGATION` / `ASSET_CONTEXT`	자산 도메인 이벤트
`ASSET_HEALTH_STATUS` / `ASSET_CONNECTION_STATUS`	자산 주기 평가 (헬스/연결 상태)
`OEE_EVENT` / `RAM_EVENT` / `EMS_EVENT`	ISO 분석 결과 이벤트
`OPC_STATUS` / `EDGE_STATUS`	OPC/엣지 디바이스 상태
`DIAGNOSTIC` / `DOMAIN_CHANGED`	진단·도메인 변경
`ALARM`	알람 발생
`WEBHOOK`	HTTP 웹훅 수신
`KAFKA_INBOUND` / `MQTT_INBOUND`	외부 토픽 수신
`TIMER`	스케줄 발화

트리거별 페이로드 예제

스크립트 노드 작성 시 어떤 필드에 접근할 수 있는지 정확히 알아야 합니다. 아래는 각 트리거가 만들어 내는 실제 메시지 JSON 예제입니다. 모든 트리거 메시지에는 공통으로 type, originator, data, metadata 가 포함됩니다.

`flow_on_tag_point` — 태그 포인트 인입

태그 한 개에 값이 들어올 때마다 발화합니다. 가장 흔한 트리거입니다.

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

필드	의미	스크립트 접근
`data.value`	수신한 값 (수치/문자열/불린)	`msg.data.value`
`data.quality`	OPC 품질 (GOOD/BAD/UNCERTAIN)	`msg.data.quality`
`data.ts`	수신 시각 (epoch ms)	`msg.data.ts`
`metadata.tag_id`	태그 ID	`msg.metadata.tag_id`

`flow_on_tag_alarm` — 태그 알람 발생

태그의 알람밴드(hi/lo/...)를 넘기는 순간 발화합니다.

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

`data.alarm_band` 값	의미
`NORMAL` / `HI` / `LO` / `HI_HI` / `LO_LO` / `TRIP_HI` / `TRIP_LO`	수치형 알람 단계
`BOOL_TRUE` / `BOOL_FALSE`	불린형 알람

`flow_on_asset_data` / `flow_on_asset_event` — 자산 이벤트

자산 단위로 집계된 이벤트(CEP 처리 결과·플러그인 평가 결과 등)가 발생할 때 발화합니다.

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

flow_on_asset_data 는 자산 단위 시계열 데이터(data.values 가 키-값 맵)를 전달하며, flow_on_asset_aggregation 은 분/시 단위 집계값을 전달합니다.

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 주기 평가

플랫폼이 1분 주기로 자산 단위 헬스/연결 상태를 평가합니다.

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

`data.status` 값	의미
`OK` / `WARN` / `ERROR` / `UNKNOWN`	헬스 단계
`CONNECTED` / `LATENT` / `ERROR` / `DISCONNECTED` / `UNKNOWN`	연결 단계 (connection_status)

prev_status 와 비교해 상태가 전이된 순간에만 후속 액션을 발화하도록 필터링하는 패턴이 많이 쓰입니다.

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — 플러그인 이벤트

워크오더 단위 OEE/RAM/EMS 평가 결과가 갱신될 때 발화합니다.

{
  "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/엣지 상태

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

`flow_on_diagnostic` — 시스템 진단 메시지

서버 모듈에서 발생한 진단 메시지가 들어오면 발화합니다.

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

`data.level`	의미
`INFO` / `WARN` / `ERROR`	진단 심각도

`flow_on_domain_changed` — 도메인 변경 이벤트

자산·태그·사이트·작업지시 등의 도메인 엔티티가 CRUD 될 때 발화합니다.

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

`flow_on_entity_event` — 엔티티 라이프사이클

ENTITY_CREATED / ENTITY_UPDATED / ENTITY_DELETED 세 타입으로 통합 발화. 단일 노드 하나로 세 가지 라이프사이클을 모두 받습니다.

{
  "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 로 인증.

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

외부 시스템이 보낸 JSON 본문 전체가 data 에 그대로 들어갑니다. HTTP 헤더는 metadata 에 일부만(remote_addr/method/request_id) 표시됩니다.

`flow_on_mqtt_subscribe` — MQTT 토픽 구독

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

`flow_jdbc_poll` — 외부 DB 주기 폴링

설정된 SELECT 쿼리를 주기적으로 실행해 각 행마다 메시지 한 건씩 발화합니다.

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

행이 100건이면 100개의 메시지가 순차로 발화됩니다. 같은 행을 반복 처리하지 않도록 SELECT 쿼리 안에 처리 플래그를 함께 갱신하거나 processed_at 컬럼 비교 조건을 넣어 주세요.

`flow_schedule` — 시간 기반 발화

Cron 표현식 또는 고정 주기로 발화합니다. 페이로드는 비어 있고 metadata.ts 만 채워집니다.

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

페이로드 변환 시 주의 사항

originator.id 는 도메인 ID — flow_change_originator 노드로 변경하면 그 후의 flow_save_attributes·flow_publish_asset_* 액션 노드가 새 originator 기준으로 동작합니다.
스크립트로 data/metadata 를 갈아끼울 때 얕은 복사 (Object.assign) 가 아닌 직접 할당을 권장합니다 — 원본 변경 시 같은 트리거를 구독하는 다른 플로우에 영향이 갈 수 있습니다.
모든 epoch 시각 필드는 밀리초 (ms) 입니다. 초 단위가 필요하면 Math.floor(msg.metadata.ts / 1000).

노드 카탈로그

자세한 노드 ID와 옵션은 편집 화면의 인스펙터에서 확인하실 수 있습니다.

트리거 (22종)

모든 플로우의 시작점은 트리거 노드입니다. 별도의 진입점/종단점 노드는 없습니다.

도메인 자동 수신 — 시스템 내부 이벤트가 자동으로 디스패치됩니다.

카테고리	트리거 노드
태그	`flow_on_tag_point` (텔레메트리), `flow_on_tag_alarm`
자산	`flow_on_asset_data`, `flow_on_asset_event`, `flow_on_asset_alarm`, `flow_on_asset_command`, `flow_on_asset_aggregation`, `flow_on_asset_context`, `flow_on_asset_health_status`, `flow_on_asset_connection_status`
플러그인	`flow_on_oee_event`, `flow_on_ram_event`, `flow_on_ems_event`
OPC/엣지	`flow_on_opc_status`, `flow_on_edge_status`
진단·도메인	`flow_on_diagnostic`, `flow_on_domain_changed`
엔티티	`flow_on_entity_event`

flow_on_alarm 이라는 노드는 없습니다

알람 트리거는 대상에 따라 둘로 나뉩니다 — 태그 알람은 flow_on_tag_alarm, 자산 알람은 flow_on_asset_alarm 입니다. 예전 문서의 예제가 flow_on_alarm 을 쓰고 있었는데 팔레트에 없는 이름이라 그대로 따라 하면 노드를 찾을 수 없습니다.

모든 트리거 노드는 *_pattern 옵션(글롭: *, ?)으로 메시지 단위 사전 필터링을 할 수 있습니다. 패턴 미매칭 메시지는 후속 노드로 전달되지 않으며 실행 카운트도 증가하지 않습니다(SKIPPED 처리). 운영 부하를 최소화하기 위해 트리거 단계에서 우선 걸러내시는 것을 권장합니다.

외부 진입

노드	동작
`flow_on_webhook`	외부 시스템이 HTTP로 푸시한 페이로드 수신
`flow_on_mqtt_subscribe`	외부 MQTT 브로커 토픽 구독
`flow_jdbc_poll`	외부 데이터베이스 SELECT 결과를 주기적으로 읽어 행마다 발화

시간 기반

노드	동작
`flow_schedule`	크론/주기 스케줄 — `TIMER` 메시지 발화

필터 (6종)

노드	설명
`flow_msg_type_filter`	`type`이 지정 목록에 포함되면 `TRUE`
`flow_originator_type_filter`	`originator.entity_type`이 지정 목록에 포함되면 `TRUE`
`flow_script_filter`	스크립트로 boolean 평가
`flow_check_existence_field`	`data`/`metadata` 특정 필드 존재 여부
`flow_switch`	다중 case 분기 (case마다 다른 relation)
`flow_check_relation`	이전 단계의 relation 기준 분기

변환 (8종)

노드	화면 이름	설명
`flow_script_transform`	스크립트 변환	스크립트로 `data`/`metadata` 변환
`flow_change_originator`	발신자 변경	`originator`를 다른 엔티티로 변경
`flow_rename_keys`	키 이름 변경	`data` 필드명 일괄 변경
`flow_template`	템플릿	`${path}` 치환 텍스트 생성
`flow_split`	분할	`data`가 배열이면 각 원소별 메시지로 분할
`flow_merge`	병합	여러 메시지를 시간 윈도우로 병합 — `flow_split` 의 반대
`flow_flatten`	평탄화	중첩 객체의 자식 키를 최상위로 끌어올림 (`data.data_map.x` → `data.x`)
`flow_to_email`	이메일 변환	메시지를 이메일 포맷으로 변환

flow_flatten 은 태그 포인트처럼 data_map 안에 값이 한 겹 더 들어 있는 메시지를 뒤 노드에서 ${data.x} 로 바로 참조하고 싶을 때 씁니다.

액션 — 통합·저장 (2종)

노드	설명
`flow_save_tag_point`	태그 포인트 적재 (정상 인제스트 경로와 동일)
`flow_save_attributes`	태그/자산 메타 부분 갱신

flow_dds_publish(내부 메시지 채널에 자유 발행) 를 찾고 계시다면 여기가 아니라 외부 연동 팔레트 그룹에 있습니다.

액션 — 자산 이벤트 발행 (4종)

CEP(EQL)와 동일한 처리 경로로 자산 이벤트를 발행합니다(영구 저장 + 캐시 + 타임라인 + 플러그인 + 메시지 채널 일관 처리).

노드	채널
`flow_publish_asset_event`	자산 이벤트
`flow_publish_asset_context`	자산 컨텍스트
`flow_publish_asset_aggregation`	자산 집계
`flow_publish_asset_command`	자산 명령

`flow_asset_command_invoke` — 발행이 아니라 실행

노드	화면 이름	하는 일
`flow_asset_command_invoke`	에셋 명령 호출	자산에 정의된 명령을 동기 실행하고 결과를 기다립니다

flow_publish_asset_command 와 헷갈리기 쉽습니다

flow_publish_asset_command — 자산 명령 이벤트를 채널에 발행합니다. 발행하고 끝입니다.
flow_asset_command_invoke — 자산에 정의된 명령을 실제로 실행하고 결과를 받을 때까지 기다립니다.

명령이 실행되기를 원했는데 발행 노드를 쓰면 아무 일도 일어나지 않은 것처럼 보입니다.

액션 — 도메인 CRUD (34종)

도메인 작업은 모두 동일 도메인 서비스에 위임되어 감사·정합성이 유지됩니다. *_field 동적 옵션으로 메시지 페이로드에서 값을 추출할 수 있습니다.

도메인	Create	Update	Delete
자산 (Asset)	`flow_create_asset`	`flow_update_asset`	`flow_delete_asset`
태그 (Tag)	`flow_create_tag`	`flow_update_tag`	`flow_delete_tag`
사이트/영역/라인	`flow_create_site`	`flow_update_site`	`flow_delete_site`
작업지시 (WorkOrder)	`flow_create_work_order`	`flow_update_work_order`	`flow_delete_work_order`
알람 설정 (EQL)	`flow_create_alarm_config`	`flow_update_alarm_config`	`flow_delete_alarm_config`
고객 (Customer)	`flow_create_customer`	`flow_update_customer`	`flow_delete_customer`
제품 (Product)	`flow_create_product`	`flow_update_product`	`flow_delete_product`
작업자 (Employee)	`flow_create_employee`	`flow_update_employee`	`flow_delete_employee`
캘린더 (시프트)	`flow_create_calendar`	`flow_update_calendar`	`flow_delete_calendar`

태그 알람밴드 부분 갱신 (2종)

노드	설명
`flow_update_tag_alarm_band_numeric`	수치형 알람밴드(`hi`/`lo`/`hi_hi`/`lo_lo`/`trip_hi`/`trip_lo`/`band_message`/`use_alarm`)를 입력한 필드만 부분 갱신
`flow_update_tag_alarm_band_boolean`	불린형 알람밴드(`bool_true`/`bool_false`/우선순위/메시지/`use_alarm`)를 입력한 필드만 부분 갱신

작업지시(WorkOrder) 상태 전이 (5종)

수치 컬럼을 직접 갱신하지 않고 도메인 서비스의 상태 전이 메서드를 호출하므로 OEE/RAM/EMS 가시성이 유지됩니다.

노드	전이
`flow_start_work_order`	`WAIT` → `START`
`flow_pause_work_order`	`START` → `PAUSED`
`flow_resume_work_order`	`PAUSED` → `START`
`flow_end_work_order`	`START` 또는 `PAUSED` → `END`
`flow_abort_work_order`	`START` 또는 `PAUSED` → `ABORTED` (`abort_code`, `notes` 옵션)

NOT NULL 자동 보강 — Create 노드는 NOT NULL 컬럼에 default 값을 자동으로 채웁니다. 예: 작업지시 status="WAIT" / master_id는 MES 마스터 ID(없으면 자동 백필), 고객 매니저 정보 "admin"/"admin@example.com", 작업자 org_id는 site_id로 폴백, 모든 행의 insert_user_id="flow". FK 컬럼(예: customer_id/product_id)에서 빈 문자열은 NULL로 변환됩니다.

Update 노드 — 부분 갱신 — Customer/Product/Employee/Calendar 와 알람밴드 Update 노드는 기존 레코드를 먼저 조회한 뒤 입력한 필드만 병합하여 저장합니다. 빈 문자열·null 값은 무시되어 기존 값이 유지됩니다. 전체 덮어쓰기가 필요하면 Delete + Create 조합을 사용해 주세요.

알람 직접 트리거 노드는 의도적으로 제외되었습니다. 알람은 flow_create_alarm_config 경로로만 발생해야 알람 이력의 정합성이 유지됩니다.

엣지 (Edge, 22종)

엣지 디바이스(OPC Agent)의 REST API를 호출하여 OPC 서버 등록·태그 CRUD·태그 값 읽기/쓰기· 조회·도커 앱 제어를 자동화합니다. 모든 노드가 의미상 기본 HTTP 메서드(GET/POST/PUT/DELETE)만 다른 동일 동작을 공유합니다.

그룹	노드
OPC 서버 관리	`flow_edge_opc_create`, `flow_edge_opc_update`, `flow_edge_opc_delete`, `flow_edge_opc_start`, `flow_edge_opc_stop`, `flow_edge_opc_list`
태그 관리	`flow_edge_tag_create`, `flow_edge_tag_update`, `flow_edge_tag_delete`, `flow_edge_tag_read`, `flow_edge_tag_write`, `flow_edge_tag_list`
조회	`flow_edge_monitoring`, `flow_edge_info`, `flow_edge_transfer`
도커 앱 제어	`flow_edge_app_list`, `flow_edge_app_inspect`, `flow_edge_app_start`, `flow_edge_app_stop`, `flow_edge_app_restart`, `flow_edge_app_logs`, `flow_edge_app_stats`

조회 3종

노드	화면 이름	하는 일
`flow_edge_monitoring`	모니터링	엣지 모니터링 지표 조회
`flow_edge_info`	엣지 정보	엣지 ID · 버전 · uptime 조회
`flow_edge_transfer`	전송 헬스	MQTT/Sparkplug 전송 상태

노드	화면 이름	하는 일
`flow_edge_opc_list`	OPC 목록	엣지의 OPC 서버 목록 조회
`flow_edge_tag_list`	태그 목록	OPC 의 태그 목록 조회

도커 앱 제어 7종

엣지 상세 화면의 도커 패널에서 손으로 하던 일을 플로우로 자동화합니다.

노드	화면 이름	하는 일
`flow_edge_app_list`	앱 목록	엣지 도커 컨테이너 목록
`flow_edge_app_inspect`	앱 상세	컨테이너 상세(inspect)
`flow_edge_app_start`	앱 시작	컨테이너 시작
`flow_edge_app_stop`	앱 중지	컨테이너 중지
`flow_edge_app_restart`	앱 재시작	컨테이너 재시작
`flow_edge_app_logs`	앱 로그	컨테이너 로그 (`line=N`)
`flow_edge_app_stats`	앱 통계	컨테이너 CPU / 메모리 / I/O

앱 제어 노드는 현장 장비를 실제로 조작합니다

app_stop · app_restart 는 현장 엣지에서 도는 컨테이너를 정말로 멈추고 다시 띄웁니다. 트리거에 붙여 자동 실행되게 만들 때는 조건을 좁게 잡으세요 — 플래핑하는 알람에 app_restart 를 물리면 컨테이너가 반복 재시작합니다.

공통 설정

옵션	설명
`url`	엣지 REST 엔드포인트. `${data.x}`/`${metadata.y}` 템플릿 치환 지원
`method`	HTTP 메서드 (미설정 시 노드별 기본값 — 예: create=POST, update=PUT, delete=DELETE, read/monitoring=GET)
`headers`	JSON 헤더 (예: `{"Authorization":"Bearer ${TOKEN}"}`)
`body_template`	요청 본문 (미설정 시 `data` 그대로 전송, GET/DELETE 는 본문 미전송)
`timeout_ms`	타임아웃 (기본 5000)

응답·분기

data.response_status — HTTP 상태 코드
data.response — 응답 본문(문자열)
data.error — 오류 메시지(실패 시)
SUCCESS (200~399) / FAILURE (그 외 또는 예외)

외부 연동 (9종)

모든 외부 노드는 *_field 동적 옵션을 지원합니다.

노드	동적 옵션
`flow_dds_publish`	내부 메시지 채널에 자유 발행 (외부 호출은 아니지만 이 그룹에 있습니다)
`flow_http_request`	`url_field` / `method_field` / `body_field`
`flow_kafka_publish`	`topic_field` / `key_field`
`flow_mqtt_publish`	`topic_field`
`flow_webhook_callback`	`url_field`
`flow_send_email`	`to_field` / `cc_field` / `subject_field` / `body_field`
`flow_send_sms`	`to_field` / `text_field`
`flow_send_push`	`title_field` / `body_field`
`flow_jdbc_query`	SQL 정적 (SELECT/INSERT/UPDATE/DELETE)

흐름 제어 (7종)

노드	설명
`flow_log`	디버그 로그 (level / prefix)
`flow_noop`	통과
`flow_delay`	`delay_ms` 후 다음 노드로 전달
`flow_throttle`	`max_msgs` / `window_ms` 제한 (초과 시 `THROTTLED` relation)
`flow_debounce`	`window_ms` 안정화 후 마지막 메시지만 발화
`flow_merge`	`window_ms` 동안 입력을 누적해 `data.merged` 배열로 한 번 emit
`flow_subflow`	`target_flow_id` — 다른 플로우 호출
`flow_retry`	`max_attempts`(기본 3) / `backoff_ms`(기본 1000) / `backoff_multiplier`(기본 2.0). 백오프 대기 후 `SUCCESS` 분기로 메시지 전달, 최대 시도 도달 시 `EXHAUSTED` 분기. `metadata.retry_count` / `metadata.retry_exhausted` 자동 갱신

Retry 권장 결선 패턴

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

노드 옵션 상세 레퍼런스

복잡한 노드의 옵션을 한 줄 표가 아닌 옵션별 기본값·예시·실패 처리까지 상세히 설명합니다.

`flow_script_filter` — 스크립트 필터

옵션	타입	기본값	설명
`script`	text	(필수)	평가식 — boolean 반환. `true` → `TRUE` 분기 / `false` → `FALSE` 분기
`script_type`	enum	`javascript`	`javascript` / `eql`
`on_error`	enum	`FALSE`	스크립트 예외 시 — `TRUE` / `FALSE` / `FAILURE` 분기로 보낼지

스크립트 컨텍스트

변수	의미
`msg.type`	메시지 타입 (POST_TELEMETRY 등)
`msg.data`	페이로드 (수정 가능하지만 필터에서는 의미 없음)
`msg.metadata`	컨텍스트
`msg.originator`	originator 객체

예시

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

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

`flow_script_transform` — 스크립트 변환

옵션	타입	기본값	설명
`script`	text	(필수)	변환식 — `msg` 객체를 수정하거나 새 객체 반환
`script_type`	enum	`javascript`	`javascript` / `eql`
`mode`	enum	`mutate`	`mutate`(in-place) / `return`(return 값 사용)

예시

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

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

mutate 모드에서 return 문이 있어도 무시됩니다. 새 객체로 교체하려면 mode=return 으로 설정해야 합니다.

`flow_switch` — 다중 분기

옵션	타입	기본값	설명
`cases`	array	(필수)	`[{expression, relation}]` 배열 — 위에서 아래로 평가, 첫 매치 채택
`default_relation`	string	`DEFAULT`	모든 case 미매치 시
`script_type`	enum	`javascript`	—

예시

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

후속 노드에서 4개의 분기 라벨(CRITICAL/WARN/INFO/NORMAL)로 각기 다른 처리가 가능합니다.

`flow_retry` — 자동 재시도

외부 IO 노드의 일시적 실패를 자동 복구합니다.

옵션	타입	기본값	설명
`max_attempts`	int	3	최대 시도 횟수 (이 값을 초과하면 `EXHAUSTED`)
`backoff_ms`	long	1000	첫 대기 시간 (ms)
`backoff_multiplier`	double	2.0	지수 백오프 배수 — 1차 1초 → 2차 2초 → 3차 4초
`max_backoff_ms`	long	30000	단일 대기 상한
`jitter_pct`	int	0	백오프에 ±N% 무작위 흔들기 (thundering herd 회피)

`metadata` 자동 보강

필드	의미
`metadata.retry_count`	현재까지 시도 횟수
`metadata.retry_exhausted`	true 이면 EXHAUSTED 분기로 진입
`metadata.retry_last_error`	마지막 실패 사유

백오프 합이 60초를 넘으면 트리거 처리 큐가 막힐 수 있습니다. 외부 시스템 응답이 일관되게 느리면 flow_throttle 로 진입 속도부터 제한하세요.

`flow_on_webhook` — HTTP 트리거

옵션	타입	기본값	설명
`auth_required`	boolean	`true`	`X-API-Key` 헤더 필수 여부 — 끄면 누구나 호출 가능
`allowed_origins`	csv	`*`	CORS Origin 화이트리스트
`max_body_kb`	int	256	본문 크기 상한 (이를 초과하면 413 응답)
`payload_pattern`	glob	`*`	메시지 사전 필터링 글롭

호출 방법

curl -X POST \
  https://platform.example.com/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 또는 API 인증 토큰 화면에서 발급한 토큰
응답: 200 OK (메시지 큐에 enqueue 됨) / 401(인증 실패) / 404(플로우 없음 또는 미배포) / 413(본문 초과)

`flow_http_request` — 외부 HTTP 호출

옵션	타입	기본값	설명
`url`	string	(필수)	호출할 URL. `${data.x}` 템플릿 치환
`url_field`	string	—	URL 을 페이로드에서 동적으로 가져올 때 — `data.endpoint` 등
`method`	enum	`GET`	`GET` / `POST` / `PUT` / `DELETE` / `PATCH`
`method_field`	string	—	메서드를 페이로드에서 동적으로
`headers`	json	`{}`	`{"Authorization": "Bearer ${TOKEN}"}` 형태
`body`	text	—	정적 본문 (템플릿 치환 지원)
`body_field`	string	—	본문을 페이로드에서 가져올 때 — 보통 `data`
`timeout_ms`	int	5000	응답 대기 상한
`follow_redirect`	boolean	`true`	3xx 리다이렉트 자동 추종
`verify_ssl`	boolean	`true`	TLS 인증서 검증 (테스트용으로만 끄세요)

응답 페이로드 보강

필드	의미
`data.response_status`	HTTP 상태 코드 (200 / 404 / 500 ...)
`data.response_body`	응답 본문 (JSON 이면 자동 파싱)
`data.response_headers`	응답 헤더 객체

분기

SUCCESS — 2xx/3xx
FAILURE — 4xx/5xx 또는 예외/타임아웃

`flow_send_email` — 이메일 발송

옵션	타입	기본값	설명
`to`	string	—	정적 수신자 (쉼표 구분)
`to_field`	string	—	페이로드에서 수신자 추출 — `data.recipient` 등
`cc` / `cc_field`	string	—	참조
`bcc` / `bcc_field`	string	—	숨은 참조
`subject` / `subject_field`	string	(필수 1개)	제목 — 템플릿 치환
`body` / `body_field`	text	(필수 1개)	본문 (HTML 허용)
`is_html`	boolean	`true`	텍스트 메일이면 끄세요
`attachments`	json	`[]`	`[{"url":"...","filename":"..."}]`

SMTP 설정은 시스템 → 설정 의 메일 설정에서 운영자가 미리 등록. 등록 전에는 모든 이메일 노드가 FAILURE 분기로 떨어집니다.

`flow_jdbc_poll` — 외부 DB 폴링

옵션	타입	기본값	설명
`datasource_id`	string	(필수)	시스템 → 설정 에 등록된 외부 DB 식별자
`query`	sql	(필수)	SELECT 쿼리 — 한 번에 최대 1,000 행 반환
`poll_interval_ms`	int	60000	폴링 주기 (기본 1분)
`marker_column`	string	—	"마지막 처리 시각" 컬럼 — 마커 이후 행만 SELECT
`marker_initial`	string	`1970-01-01 00:00:00`	첫 폴링 시 마커 시작값
`row_limit`	int	1000	한 폴링당 최대 행 (이를 초과해도 안전)
`on_error_continue`	boolean	`true`	DB 오류 시 진단만 기록하고 다음 폴링 진행

`marker_column` 사용 예

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

→ :marker 자리에 마지막 폴링에서 받은 최대 created_at 값이 자동으로 들어갑니다.

`flow_kafka_publish` / `flow_mqtt_publish` — 외부 발행

옵션	타입	기본값	설명
`broker`	string	(필수)	`kafka:9092` 또는 `tcp://mqtt:1883`
`topic`	string	—	정적 토픽 — `${data.x}` 치환 가능
`topic_field`	string	—	페이로드에서 토픽 추출 (`data.target_topic` 등)
`key` / `key_field`	string	—	(Kafka 만) 메시지 키
`body` / `body_field`	json/text	(필수 1개)	발행 본문 — 미설정 시 `data` 그대로
`qos`	int	1	(MQTT 만) 0/1/2
`retain`	boolean	false	(MQTT 만) Retained 플래그

`flow_publish_asset_*` — 자산 이벤트 발행

자산 이벤트(이벤트/컨텍스트/집계/명령)는 CEP/플러그인/타임라인이 동시에 인식하는 통합 채널로 발행됩니다. 4개 노드의 공통 옵션:

옵션	타입	기본값	설명
`asset_id`	string	—	정적 자산 ID
`asset_id_field`	string	—	페이로드에서 자산 ID 추출 (보통 `metadata.asset_id`)
`event_type`	string	—	자산 이벤트 분류 (예: `STARTUP`, `SHUTDOWN`, `MAINTENANCE`)
`event_type_field`	string	—	페이로드에서 분류 추출
`payload`	json	`${data}`	발행할 본문 — 미설정 시 `data` 그대로

flow_publish_asset_command 의 경우 자산의 명령 수신 토픽(asset_id/cmd/{event_type})으로 즉시 전달되어 엣지에 도달합니다.

`flow_create_` / `flow_update_` — 도메인 CRUD

모든 Create/Update 노드는 다음 옵션 패턴을 공유합니다.

옵션	타입	설명
`{컬럼명}`	string	정적 값 (입력하지 않으면 NULL/default)
`{컬럼명}_field`	string	페이로드에서 값 추출 (`data.foo` / `metadata.bar`)
`id_strategy`	enum	`auto`(시스템 발급) / `field`(`{도메인}_id_field` 에서 추출)
`on_duplicate`	enum	`error`(기본) / `skip` / `update` — Create 노드만

NOT NULL 자동 보강

Create 노드는 NOT NULL 컬럼에 default 값을 자동으로 채웁니다.

도메인	자동 채워지는 컬럼
작업지시	`status="WAIT"` / `master_id`(없으면 자동 백필) / `insert_user_id="flow"`
고객	매니저 정보 `"admin"` / `"admin@example.com"` (등록되지 않은 경우)
작업자	`org_id=site_id`(폴백)
공통	`insert_date=now()` / `insert_user_id="flow"`

FK 빈 문자열 처리

FK 컬럼(customer_id/product_id 등)에 빈 문자열 "" 이 들어오면 자동으로 NULL 로 변환됩니다. JS 에서 delete msg.data.customer_id 보다 msg.data.customer_id = '' 가 더 안전합니다.

`flow_edge_*` — 엣지 REST 호출

엣지 디바이스의 REST 엔드포인트를 호출하는 22종 노드는 공통 옵션을 공유합니다.

옵션	타입	기본값	설명
`edge_id`	string	—	정적 엣지 ID
`edge_id_field`	string	—	페이로드에서 엣지 ID 추출
`path`	string	(노드별 기본)	엣지 REST 경로 (예: `/api/v1/opc`, `/api/v1/app/grafana/start`)
`body_template`	text	—	요청 본문 — 미설정 시 `data` 그대로
`timeout_ms`	int	5000	—

자동 인증

엣지 노드는 호출 시 mm_edge 마스터 테이블에서 그 엣지의 api_key 를 자동 lookup 하여 X-API-Key 헤더로 첨부합니다. 운영자가 별도 설정할 필요가 없습니다.

응답 페이로드

필드	의미
`data.edge_response_status`	엣지 응답 코드
`data.edge_response`	응답 본문
`data.edge_id`	호출 대상 엣지 ID (확인용)

분기는 flow_http_request 와 동일 (SUCCESS / FAILURE).

스크립트 노드 작성

flow_script_filter / flow_script_transform / flow_switch 노드는 두 가지 표현식을 지원합니다.

JavaScript (기본 권장)

표준 ECMAScript 문법. 다중 라인·var/let/const·함수·객체 리터럴을 모두 지원합니다.

바인딩

변수	설명
`msg`	메시지 전체. `msg.data.x`, `msg.metadata.topic`, `msg.type`, `msg.originator.id` 모두 직접 접근
`data`	`msg.data` 단축 별칭
`metadata`	`msg.metadata` 단축 별칭

필터 예시

data.temp > 80

변환 예시

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

스위치 예시 (case별 boolean)

data.t > 100   // case "Critical"

자주 쓰는 변환 패턴

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

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

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

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

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

자주 쓰는 필터 패턴

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

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

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

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

사용자 스크립트는 안전한 샌드박스에서 실행되며, 파일·네트워크·스레드·임의 클래스 접근은 모두 차단됩니다. 외부 시스템 호출이 필요하면 flow_http_request 같은 외부 연동 노드를 별도로 결선하세요.

EQL 표현식

기존 EQL 사용자 호환용. 단일식만 지원(다중 라인·세미콜론 분리 미지원)되며 부수효과 패턴만 사용 가능합니다.

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

다중 동작이 필요한 경우 JavaScript를 사용해 주세요.

JavaScript 실행 환경 사양

스크립트는 격리된 샌드박스에서 실행됩니다. 어떤 기능이 가능한지/불가능한지 정확히 알아두셔야 안정적인 스크립트를 작성할 수 있습니다.

사용 가능 (✅)

기능	비고
표준 ECMAScript	`var`/`let`/`const`·함수·클래스·구조분해·spread·`?.`·`??` 등
객체 리터럴	`{ key: value, ... }`
배열 메서드	`map` / `filter` / `reduce` / `forEach` / `find` / `some` / `every` / `flat` / `slice`
문자열 메서드	`split` / `replace` / `includes` / `match` / `padStart` / `repeat`
수학 함수	`Math.*` 전체
JSON	`JSON.parse` / `JSON.stringify` (단, `data` 가 이미 객체면 다시 stringify 불필요)
Date	`new Date()` / `Date.now()` / `getHours()` / `toISOString()` 등
정규식	`/pattern/` 리터럴 + `RegExp` 생성자
에러 throw	`throw new Error('...')` — `FAILURE` 분기로 자동 분기
try/catch/finally	예외 처리

사용 불가 (❌)

기능	이유 / 대안
네트워크 호출 (`fetch` / `XMLHttpRequest`)	샌드박스 차단 — `flow_http_request` 노드를 별도 결선
파일 시스템 (`require('fs')`)	샌드박스 차단
스레드 (`setTimeout` / `setInterval` / `Worker`)	동기 실행만 허용 — 지연이 필요하면 `flow_delay` 노드
`require` / `import`	외부 모듈 로드 불가 — 필요한 함수는 같은 스크립트 안에 정의
`eval` / `new Function(string)`	보안상 차단
임의 Java 클래스	EQL 호환 모드에서도 사용자 코드에 노출 안됨
`process` / `global` / `window`	정의되지 않음
WebSocket / EventSource	차단 — 메시지 수신은 트리거 노드

실행 제한 — 없습니다

스크립트에는 시간·메모리 제한이 걸려 있지 않습니다

예전 문서는 실행 시간 500ms · 메모리 16MB · 스택 1024 · 출력 2MB 라는 표를 싣고 있었지만 넷 다 적용되지 않습니다.

노드 설정에 timeout_ms 를 쓸 수 있는 것처럼 보이지만, 스크립트 노드는 그 값을 읽지 않습니다. (ScriptTransformNode · ScriptFilterNode 는 script 와 language 만 읽습니다.)
JS 실행 컨텍스트에도 시간·구문 수·메모리 제한이 설정되어 있지 않습니다.

그래서 while (true) {} 같은 스크립트는 끊기지 않고 워커 스레드를 점유합니다. 플로우 전체 한도인 30초(flow_timeout_ms)는 실행 루프가 다음 작업을 꺼낼 때 확인하므로, 스크립트 안에서 도는 무한 루프는 그 검사에 도달하지 못합니다.

스스로 끝나는 스크립트만 쓰세요. 무한 루프·거대 반복·큰 문자열 누적을 피하고, 루프에는 반드시 상한을 두세요.

실제로 막혀 있는 것 — 보안 샌드박스

시간·메모리 대신, 바깥을 건드리는 행위는 확실히 차단되어 있습니다.

항목	상태
자바 클래스 접근 (`Java.type` 등)	차단
파일·네트워크 IO	차단
스레드 생성	차단
네이티브 접근	차단
실험적 옵션	차단
호스트 객체 접근	허용 목록에 있는 것만

외부 호출이 필요하면 스크립트 안에서 하지 마시고 flow_http_request 같은 전용 노드로 분리하세요 — 스크립트에서는 어차피 되지 않고, 전용 노드에는 timeout_ms 가 실제로 걸립니다.

무거운 처리는 스크립트 1개로 몰지 마시고 여러 노드로 나누세요.

`msg` 반환 규약

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

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

표준 시각·로케일

항목	동작
서버 시스템 타임존	UTC (epoch ms 직접 사용 권장)
한국 시각 표시	`toLocaleString('ko-KR', { timeZone: 'Asia/Seoul' })`
한국 시각 시간 계산	`new Date(ts + 9*3600000).getUTCHours()` 또는 위 로케일
0초 단위 timestamp	`Math.floor(Date.now() / 1000) * 1000`

메모리 안전 작성 패턴

스크립트 노드는 메모리 16MB 한도가 있지만, 자주 호출되는 노드에서 작은 메모리 누수가 누적되면 엔진 GC 부담이 커집니다. 다음 패턴을 피하세요.

안티패턴 1 — 클로저가 큰 데이터 유지

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

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

안티패턴 2 — 대용량 배열 그대로 전달

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

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

안티패턴 3 — 깊은 객체 복사

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

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

안티패턴 4 — 정규식 폭발 (Catastrophic Backtracking)

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

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

안티패턴 5 — `let` 누적 변수를 함수 밖에서

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

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

안티패턴 6 — 중첩 try/catch 무한 시도

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

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

스크립트 안에서 throw 한 예외는 FAILURE 분기로 가며 메시지는 보존됩니다. 외부 IO 노드는 따로 자체 분기를 가지므로 try/catch 로 감싸지 마세요.

메모리 압박 진단

NODE_SCRIPT_OOM 같은 코드는 없습니다

예전 문서가 NODE_SCRIPT_OOM · FLOW_MSG_TRIMMED · ENGINE_GC_PRESSURE 를 메모리 압박 신호로 안내했지만, 제품이 그런 코드를 만들지 않습니다. 스크립트 메모리 상한도 없고 페이로드 자동 잘림도 없습니다.

메모리 압박은 코드가 아니라 지표와 증상으로 판단하세요.

볼 것	압박 신호
`FLOW_EXECUTOR_DROPPED_COUNT`	0 에서 올라가기 시작 — 처리 풀이 포화되어 메시지를 버리는 중
`FLOW_EXECUTOR_QUEUE_SIZE`	줄지 않고 계속 증가
`FLOW_EXECUTION_TIME`	최대값이 평소의 몇 배로 튐
서버 로그	`work_queue full ... task dropped` 경고, `FlowExecutor timeout` 경고
JVM	GC 시간 증가 · 힙 사용률 상승

위 신호가 보이면 큰 페이로드를 다루는 플로우부터 점검하고, 스크립트에서 대용량 문자열· 배열을 누적하고 있지 않은지 확인하세요.

모바일 푸시 알림 채널

flow_send_push 노드로 운영자 모바일 앱에 푸시 알림을 발송합니다.

노드 옵션

옵션	타입	기본값	설명
`to` / `to_field`	string	(필수 1개)	수신자 사용자 ID (콤마 구분) 또는 페이로드 경로
`title` / `title_field`	string	(필수 1개)	알림 제목 (50자 이내 권장)
`body` / `body_field`	text	(필수 1개)	본문 (120자 이내 권장)
`priority`	enum	`NORMAL`	`NORMAL` / `HIGH` — HIGH 는 잠금화면에서도 표시
`sound`	enum	`default`	`default` / `silent` / 사용자 정의 사운드
`data`	json	`{}`	앱이 받아서 처리할 부가 데이터 (페이로드 4KB 한도)
`deep_link`	string	—	알림 탭 시 열릴 화면 (예: `pp://asset/MOTOR-001`)
`ttl_sec`	int	86400	미수신 시 보관 시간 (초) — 만료 시 자동 삭제

사용자별 토큰 자동 라우팅

운영자가 모바일 앱을 처음 로그인하면 디바이스 토큰이 보안 → API 인증 토큰 에 자동 등록됩니다. 플로우에서는 토큰을 직접 다루지 않고 사용자 ID 만 지정하시면 됩니다.

iOS / Android 양쪽 토큰이 등록되어 있으면 두 디바이스 모두 발송
토큰이 무효(앱 삭제 등) 면 자동으로 등록 해제

단순 발송 예시

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

다국어 푸시

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

이후 flow_send_push 의 title_field=data.push_title / body_field=data.push_body 로 사용.

그룹 알림 묶음 (Inbox-style)

여러 알람을 한 번에 묶어 한 알림으로 (5분 단위):

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

사용자 부재 시 자동 에스컬레이션

푸시 미확인 30분 후 SMS 또는 이메일로 자동 에스컬레이션.

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

발송 빈도 제한 권장

우선순위	권장 빈도
HIGH (잠금화면 표시)	사용자당 시간당 5건 이하 — 알람 피로 방지
NORMAL	사용자당 시간당 20건 이하

flow_throttle 노드를 발송 전에 결선하시거나, 같은 자산의 알람은 flow_debounce 로 안정화 후 발송하세요.

외부 인증 토큰 자동 갱신

OAuth2 같은 만료되는 토큰을 외부 시스템 호출 시 자동으로 갱신하는 패턴.

단순 갱신 — 시간 기반 (cron)

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

이후 다른 플로우의 flow_http_request 에서는:

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

적극적 갱신 — 401 받았을 때

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

Refresh Token 사용

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

토큰 만료 임계 자동 알람

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

자격 증명 보관 권장 위치

종류	권장 위치
정적 (변경 거의 없음)	시스템 → 설정 의 자격 증명 저장소
동적 (자동 갱신)	위 패턴으로 `flow_save_attributes` 사용해 자산 메타에 저장
사용자별 OAuth	보안 → API 인증 토큰 화면

자격 증명은 그래프 안에 평문으로 두지 마시고 반드시 외부 저장소 참조 (${creds.x}) 형태로 작성하세요. 내보내기/가져오기 시에도 평문 토큰이 JSON 에 포함되지 않습니다.

language 옵션으로 어떤 언어를 사용할지 선택합니다 (JS 또는 EQL).

실행 이력 화면

노드 단위 실행 이벤트를 시간순으로 조회할 수 있습니다.

검색 조건

항목	설명
시간	조회 기간을 지정합니다
레벨	`전체` / `INFO` / `WARN` / `ERROR`
플로우 ID	특정 플로우만 필터
메시지 ID	단일 메시지 추적용
개수	최근 200/500/1,000건

빠른 시간 범위

타임라인 패널 상단의 버튼으로 즉시 이동: 10분전 / 30분전 / 1시간전 / 6시간전 / 12시간전 / 전체기간.

결과 컬럼

컬럼	설명
레벨	INFO / WARN / ERROR
시간	이벤트 발생 시각
이벤트	`FLOW_START` / `NODE_IN` / `NODE_OUT` / `NODE_ERROR` / `FLOW_END`
플로우 ID	어떤 플로우인지
노드	노드 표시명
노드 타입	예: `flow_script_transform`
관계	`SUCCESS` / `FAILURE` / `TRUE` / `FALSE` / `MATCH` / `NO_MATCH` / `DEFAULT` / `THROTTLED` / `EXHAUSTED` (모두 대문자)
메시지	메시지 타입 · 주체 · 관계 · 처리 시간 · 데이터 미리보기 요약

CSV 다운로드 버튼으로 현재 조회 결과를 내보낼 수 있습니다.

실행 이력의 보존 기간은 7일입니다. 장기 보관이 필요하면 외부 로그 시스템으로 적재해 주세요.

일반 워크플로우

새 플로우 생성 — 목록 화면 새 플로우 버튼 → 이름·설명 입력
편집 화면 진입 — 자동으로 빈 캔버스가 열립니다
트리거 노드 배치 — 좌측 팔레트에서 트리거 노드를 드래그
처리 노드 추가 — 필터 → 변환 → 액션 순으로 배치하고 와이어로 연결
노드 설정 — 각 노드를 클릭해 우측 인스펙터에서 옵션 입력
저장 — 우상단 저장 버튼 (자동 스냅샷 적재)
테스트 실행 — 임의 메시지를 주입해 결과를 확인
배포 — 배포 토글로 활성화 → 트리거 이벤트가 들어오면 자동 실행
모니터링 — 라이브 디버그 패널과 실행 이력 화면으로 확인

예시 플로우

각 예시는 노드 결선 다이어그램 + 핵심 노드 설정 + 동작 설명으로 구성됩니다. JSON 형태의 노드 설정은 우측 인스펙터에서 입력하시는 값과 1:1 대응됩니다.

예시 1: MES 작업지시 자동 생성

MES에서 신규 PO를 매분 폴링해 작업지시로 변환합니다.

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

노드 설정

노드	핵심 설정
`flow_schedule`	`cron`: `0 * * * * ?` (매분 0초)
`flow_http_request`	`method`: `GET`, `url`: `https://mes.example.com/api/po/list?status=NEW`, `headers`: `{"Authorization":"Bearer ${MES_TOKEN}"}`
`flow_split`	`path`: `data` (배열로 분할)
`flow_script_transform`	`language`: `JS`, 아래 스크립트 참고
`flow_create_work_order`	`master_id_field`: `data.master_id`, `asset_id_field`: `data.asset_id`, `title_field`: `data.title`
`flow_send_email`	`to_field`: `metadata.alert_to`, `subject`: `[MES 동기화 실패] ${data.po}`

Script Transform 예시

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

예시 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 예시

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

예시 3: 알람 → 긴급 작업지시 자동 발행

ERROR 등급 이상의 알람을 받으면 상위 자산을 대상으로 긴급 정비 워크오더를 발행합니다.

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

Script Filter 예시

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

예시 4: 외부 DB 동기화 — 자산 일괄 등록

레거시 DB에서 신규 설비 행을 폴링해 자산으로 자동 등록합니다.

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

노드 설정

노드	핵심 설정
`flow_jdbc_poll`	`dsn`: 외부 DB 연결, `sql`: `SELECT * FROM legacy_assets WHERE sync_status='NEW' LIMIT 100`, `interval_ms`: `60000`
`flow_create_asset`	`asset_id_field`: `data.legacy_id`, `asset_name_field`: `data.name`, `site_id_field`: `data.plant_code`
`flow_jdbc_query`	`sql`: `UPDATE legacy_assets SET sync_status='OK' WHERE id=?`, `params_field`: `data.legacy_id`

예시 5: 작업지시 자동 상태 전이

자산에서 발생하는 가동/정지 이벤트로 작업지시 상태를 자동 전이시킵니다.

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

Switch 노드 케이스 예시

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

상태 전이가 실패한 경우(잘못된 현재 상태)는 자동으로 FAILURE로 라우팅되므로 별도의 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 예시

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

예시 7: 엣지 디바이스 자동 등록

새 OPC 서버 정보를 받으면 엣지 디바이스에 일괄 등록합니다.

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

호출 본문 예시

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

예시 8: 외부 API 신뢰성 보강 (Retry)

간헐적으로 실패하는 외부 API 호출에 백오프 재시도를 적용하고, 최대 시도를 넘어가면 운영자에게 알립니다.

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

Retry 노드 설정

max_attempts: 5
backoff_ms: 2000
backoff_multiplier: 2.0 → 2초, 4초, 8초, 16초, 32초 간격

예시 9: 플러그인 이벤트 라우팅 (OEE 저조 알림)

OEE 값이 임계 이하로 떨어지면 라인 관리자에게 푸시 알림을 보냅니다.

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

Template 예시

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

예시 10: 메시지 정제 파이프라인 (Throttle + Debounce)

고빈도 태그 변화를 1분에 1회로 제한하고, 추가로 5초간 변화가 없을 때만 후속 노드를 발화합니다.

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

예시 11: 다중 트리거 합산 (Merge)

OEE/RAM/EMS 3종 이벤트를 30초 단위로 묶어 한 번의 보고서 메시지로 발송합니다.

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

같은 플로우 안에 트리거 노드를 여러 개 두면 모두 진입점이 됩니다. flow_merge의 data.merged 배열에 윈도우 동안 들어온 모든 메시지가 누적됩니다.

예시 12: 서브플로우로 공통 처리 묶기

공통 메시지 정제 로직(중복 검사 + 단위 변환 + 적재)을 별도 플로우로 분리하고 여러 트리거에서 호출합니다.

메인 플로우 (각각 독립)

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

서브 플로우 FLOW_00099

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

서브플로우는 트리거가 없어도 저장되며, 외부 호출 전용으로 사용할 수 있습니다(저장 시 경고가 표시됩니다).

예시 13: 시프트 변경 시 일일 보고서 자동 발송

매일 야간 시프트 종료 시각(예: 06:00)에 어제~오늘에 걸친 생산·품질·다운타임 요약을 메일로 발송합니다.

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

Script Transform 예시

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

예시 14: 다운타임 자동 집계

자산 STOP 이벤트가 들어오면 시프트별 누적 다운타임을 갱신하고, 임계 초과 시 알림을 보냅니다.

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

예시 15: 품질 불량 라인 자동 격리

품질 점검에서 연속 5건 이상 불량이 보고되면 라인 자산에 정지 명령을 발행하고 작업지시를 중단합니다.

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

예시 16: 에너지 임계 초과 — 라인 일시 정지 권고

라인의 시간당 에너지 소비가 예산을 초과하면 운영자에게 SMS와 함께 권고 메시지를 보냅니다.

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

예시 17: 외부 시스템 양방향 동기화 — 작업지시 상태 미러링

외부 ERP에서 들어오는 상태와 내부 작업지시 상태를 양방향으로 동기화합니다.

아래로 (ERP → 내부)

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

위로 (내부 → ERP)

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

양방향 동기화 시 무한 루프를 피하려면 메시지 출처를 metadata.source 같은 키로 표기하고, 트리거 단계에서 자기 발행 메시지를 필터링하세요.

가져오기/내보내기

플로우 정의를 JSON으로 직렬화하여 다른 환경으로 이식하거나 백업할 수 있습니다.

동작	위치	설명
내보내기	편집 화면 상단 `내보내기` 버튼	그래프 + 노드 + 와이어 + 노드 설정 전체를 JSON으로 다운로드
가져오기	목록 화면 상단 `가져오기` 버튼	JSON 텍스트를 붙여넣기 또는 업로드
자동 스냅샷	저장 시 자동	그래프 저장 시 버전 단위로 적재 (롤백 대비)

가져오기 동작

항상 새 플로우 ID가 발급됩니다(기존 ID 덮어쓰기 방지)

노드 ID도 새로 부여되며 와이어 링크가 자동으로 재매핑됩니다

import 직후의 플로우는 해제 상태로 적재되며, 운영자가 검토 후 직접 배포해야 합니다

에러 핸들링

노드 단위

노드 처리 중 예외가 발생하면 자동으로 FAILURE relation으로 라우팅됩니다.
FAILURE 출력에 연결된 노드가 없으면 메시지는 drop되고 에러 로그만 남습니다.
트리거 노드의 *_pattern 미매칭 메시지는 SKIPPED 처리되어 후속 노드로 전달되지 않으며, 처리/에러 카운터에도 포함되지 않습니다.
모든 에러는 라이브 디버그 패널과 실행 이력 화면에 표시됩니다.

플로우 단위

항목	기본값	설명
max_depth	100	한 메시지 처리 중 누적 노드 방문 수 제한
max_revisit	3	같은 노드 재방문 횟수 제한 (사이클 무한 루프 방지)
flow_timeout_ms	30,000	처리 시간 초과 시 강제 종료

외부 IO 노드 재시도

HTTP·Kafka·외부 DB 등 외부 IO 노드는 retry_count / retry_delay_ms 옵션으로 노드 내부에서 즉시 재시도가 가능하며, 모든 재시도 실패 시 FAILURE relation으로 라우팅됩니다. 좀 더 정교한 백오프나 EXHAUSTED 분기 처리가 필요한 경우 flow_retry 노드를 별도로 사용하세요.

운영 진단

관리자는 시스템 메뉴에서 플로우 엔진의 디스패치 큐 상태(워커 가동 여부, 대기 큐 크기, 누적 처리/실패/드롭 건수, 마지막 오류)를 확인할 수 있습니다.

서버는 백그라운드에서 플로우 워커 풀의 부하를 주기적으로 점검하며, 부하가 일정 시간 이상 지속되면 운영 로그에 한 줄로 상태 변화(HEALTHY → DEGRADED → CRITICAL)를 기록합니다. 정상으로 회복되면 회복 로그가 한 줄 더 남습니다.

권한

플로우는 별도의 권한 모델을 도입하지 않고 기존 시스템 인증/권한을 그대로 사용합니다.

기능	필요 권한
목록 조회 / 실행 이력 조회	모든 인증된 사용자
플로우 생성·편집·배포·삭제	ADMIN
가져오기/내보내기 / 모두 재배치	ADMIN

운영 패턴 (Recipes)

자주 쓰이는 결선 형태를 모아 둔 라이브러리입니다. 각 패턴은 그대로 복제해 새 플로우의 출발점으로 활용하실 수 있습니다.

패턴 1 — 처리 + 알림 분기

처리 결과에 따라 SUCCESS는 적재, FAILURE는 알림으로 이중 분기.

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

패턴 2 — 안전한 재시도

외부 IO가 일시적 실패에 강건하도록 백오프 + EXHAUSTED 핸들러를 구성.

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

패턴 3 — 사전 필터로 부하 차단

트리거 단계의 *_pattern 으로 메시지를 미리 걸러 후속 처리량을 줄임. 패턴 미매칭은 SKIPPED 처리되어 카운터에 포함되지 않습니다.

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

패턴 4 — 다중 case 분기

상태/타입에 따라 여러 갈래로 분기.

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

패턴 5 — 윈도우 누적 + 한 번에 emit

고빈도 입력을 일정 시간 모아 한 메시지로 변환.

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

패턴 6 — 리트라이 + 재시도 한도 후 우회

재시도가 끝나면 백업 경로(다른 API, 알림, DB 적재)로 우회.

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

패턴 7 — 도메인 변경 후 후속 액션

태그 단위 알람을 상위 자산 단위로 변환해 자산별 처리에 위임.

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

패턴 8 — 스로틀 + 디바운스 결합

분당 1회 이하로 스로틀 + 5초간 변화가 없을 때만 처리.

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

패턴 9 — 서브플로우로 공통 로직 모듈화

여러 진입점이 같은 후속 처리(검증·정제·적재)를 공유해야 할 때 서브플로우로 분리.

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

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

패턴 10 — 직접 트리거 우회

다른 플로우의 결과를 다음 플로우의 입력으로 흘리려면 flow_dds_publish 로 도메인 채널에 발행하고, 다른 플로우는 동일 메시지 타입을 트리거로 받음.

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

활용 예시 (한눈 보기)

시나리오	구성
MES 연동	`flow_schedule` → `flow_http_request` → `flow_script_transform` → `flow_create_work_order`
이벤트 외부 중계	`flow_on_asset_event` → `flow_msg_type_filter` → `flow_mqtt_publish`
데이터 정제·적재	`flow_on_tag_point` → `flow_script_transform` → `flow_save_tag_point`
알람 자동화	`flow_on_tag_alarm` → `flow_script_filter` → `flow_send_email` + `flow_create_work_order`
외부 DB 동기화	`flow_jdbc_poll` → `flow_script_transform` → `flow_create_asset`
웹훅 수신	`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 참고)
`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)

운영 중인 플로우를 변경할 때 권장 순서입니다.

백업 — 편집 화면 내보내기 로 현재 그래프를 JSON 파일로 다운로드해 보관합니다 (자동 스냅샷도 적재되지만 외부 보관이 안전).
복제 또는 새 플로우로 작업 — 운영 중인 플로우를 즉시 수정하기보다 새 플로우(또는 가져오기로 만든 사본)에서 변경 후 검증하세요.
테스트 실행 — JSON 메시지를 직접 주입해 모든 분기(SUCCESS/FAILURE/EXHAUSTED 등)를 한 번씩 발화시켜 실행 이력에서 결과를 확인합니다.
운영 반영 — 검증된 플로우의 그래프 JSON을 내보내기 → 운영 환경에서 가져오기 → 운영자가 검토 후 배포 토글.
문제 발생 시 롤백 — 즉시 해제 토글로 비활성화. 자동 스냅샷이 적재되어 있으므로 개발자에게 의뢰하면 이전 버전으로 되돌릴 수 있습니다.

카운터·통계 활용

목록 화면의 처리 추세 스파크라인이 갑자기 평탄해진다면 트리거 미발화 또는 SKIPPED 처리 가능성을 확인하세요.
에러 비율 도넛 차트가 비정상이면 노드 설정의 *_field 경로를 의심해 보세요. 메시지 페이로드에 키가 없으면 자주 실패합니다.
운영 검증 후에는 전체 카운트 초기화 로 통계 윈도우를 리셋해 평상시 베이스라인을 다시 측정하세요.

긴급 대응 절차

운영 중 문제가 발생했을 때 신속하게 사용하실 단계별 절차입니다.

시나리오 1 — 특정 플로우가 폭주(메시지 폭증)

증상: 한 플로우의 실행 건수가 정상치 대비 수십~수백 배 급증, 에러도 동반 증가

조치

즉시 해당 플로우의 해제 토글로 비활성화 (목록 화면에서 1초)
라이브 디버그 패널·실행 이력에서 어떤 트리거가 폭주를 일으켰는지 확인
트리거 노드의 *_pattern 옵션을 좁히거나, 직후에 flow_throttle / flow_debounce 결선
필요 시 flow_check_existence_field 또는 flow_msg_type_filter 로 메시지 타입 한정
수정 후 테스트 실행 으로 재현 → 정상 확인 후 배포 재개

시나리오 2 — 외부 시스템 장애로 일괄 실패

증상: HTTP/Kafka/외부 DB 등 외부 IO 노드의 에러가 동시에 누적

조치

영향 범위가 넓다면 관련 플로우들을 일괄 해제 (목록 화면 일괄 선택 후 일괄 해제)
외부 시스템 복구 확인
flow_retry 결선이 없는 외부 IO 노드라면 결선 추가
외부 시스템 응답이 느려졌다면 timeout_ms 조정
운영 환경에 따라 모두 재배치 후 배포 재개

시나리오 3 — 사이클로 인한 무한 루프

증상: 단일 메시지가 같은 노드를 계속 통과, 처리 시간 누적

조치

max_revisit 보호로 최대 3회까지만 재방문되어 자동 차단되지만, 운영 안전을 위해 해제 후 점검
그래프에서 사이클을 시각적으로 추적 (편집 화면에서 와이어 따라가기)
의도된 루프(flow_retry)면 max_attempts 가 적정한지 확인
의도되지 않은 사이클이면 결선 제거 또는 flow_check_relation 으로 분기 추가
수정 후 전체 카운트 초기화 → 재배포

시나리오 4 — 데이터 손상 의심 (잘못된 자동 갱신)

증상: 자동화로 도메인 데이터가 의도와 다르게 갱신됨

조치

즉시 해당 플로우 해제
외부에 보관 중인 직전 그래프 JSON 백업 또는 자동 스냅샷에서 이전 버전 확인 (관리자 협조)
그래프 분석: 의도하지 않은 Update 노드 결선·잘못된 *_field 경로·스크립트 변환 오류 확인
영향 받은 도메인 데이터는 별도 절차로 백오피스 화면에서 정정
수정한 그래프를 테스트 실행 으로 검증 후 재배포

시나리오 5 — 시스템 점검·유지보수 중 중지

증상: 외부 시스템 점검 시간 동안 외부 IO 호출을 잠시 멈추고 싶음

조치

영향받는 플로우들을 일괄 해제 (목록에서 다중 선택)
점검 종료 후 배포 재개
자체 스케줄러 보유 트리거(스케줄·외부 MQTT 구독·외부 DB 폴링)는 모두 재배치 1회 실행으로 정상 등록 확인

시나리오 6 — 라이브 디버그 적재 큐 가득참

증상: 운영 진단 페이지의 적재 큐 크기가 임계 근처, 드롭 건수 증가

조치

디버그 모드가 켜진 플로우 수와 적재량을 줄이세요 (검증 끝난 플로우는 디버그 모드 OFF)
트리거 단계의 *_pattern 으로 후속 처리량 자체를 줄이세요
시스템이 자동 회복 모드로 들어가면 부하 워치독에 DEGRADED/CRITICAL 로그가 한 줄 남습니다 — 관리자에게 공유하세요

모든 시나리오에서 우선순위는 즉시 해제 → 원인 파악 → 수정 → 검증 → 재배포 순서입니다. 변경 절차(변경 안전 절차)도 함께 참고하세요.

보안·민감 정보 처리

플로우는 외부 API 호출·이메일/SMS 발송·웹훅 수신 등 민감한 입출력을 다루므로 다음 원칙을 지켜 주세요.

토큰·자격 증명

API 토큰·비밀번호를 노드 설정에 평문으로 직접 입력하지 마세요. ${ENV_VAR} 형태의 환경 변수 치환을 사용해 운영 환경의 비밀 저장소에서 주입받으세요.
운영 환경에서 토큰을 노출시키지 않도록 headers JSON 의 토큰 위치를 가능한 한 환경 변수로 분리합니다.

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

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

웹훅 진입점 보호

flow_on_webhook 은 인증된 사용자가 호출하는 내부 진입점이지만, 외부에 노출하실 때는 페이로드 검증 필터를 반드시 결선하세요.

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

민감 데이터 마스킹

라이브 디버그 패널과 실행 이력에는 메시지 미리보기가 표시됩니다. 개인정보(이름·연락처·계정)와 토큰은 적재 직전에 마스킹하세요.

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

외부 IO의 응답 본문 처리

flow_http_request 의 data.response 는 응답 본문 전체가 적재됩니다. 응답에 민감 데이터가 포함된 경우 즉시 후속 변환 노드로 필요한 키만 추출하고 원본을 제거하세요.

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

스크립트 노드 격리

스크립트 노드는 안전한 샌드박스에서 실행되며 파일·네트워크·임의 클래스 접근이 차단됩니다. 외부 호출이 필요하면 반드시 별도의 외부 연동 노드를 결선하세요.

플로우 메트릭과 알람

플로우가 운영 중일 때 자동으로 수집되는 지표와 그 지표를 보는 곳·알람을 거는 방법.

플로우 단위 KPI

목록 화면(/flow/index) 의 각 행에 다음 KPI 가 표시됩니다.

지표	의미	비정상 신호
전체 실행 건수	트리거가 발화하여 그래프를 한 번 통과한 건수 (성공/실패 합)	평소 대비 급감 → 트리거가 죽음 / 급증 → 폭주
에러 건수	그래프 중 어느 노드에서든 `FAILURE` 분기로 빠진 건수	전체 대비 5% 이상이면 점검
에러율	`에러 / 전체 × 100`	일정 임계 초과 시 알람 거세요
마지막 실행	가장 최근 발화 시각	"5분 이상 발화 없음" 이 정상 인 경우만 OK
평균 처리 시간	메시지 한 건이 그래프 전체를 통과하는 평균 ms	외부 IO 노드 추가/제거 시 변화 큼
최근 처리 시간 추이	5분 스파크라인 — 라이브 디버그 패널	spike 발생 시 외부 시스템 응답 점검

노드 단위 KPI

편집 화면에서 노드를 클릭하면 노드 우하단에 다음이 표시됩니다.

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

위치	표시
상단 점등	회색(대기) / 녹색(처리 중) / 빨강(에러)
하단 라벨	`처리 N · 에러 M · 평균 Xms · 최근 Yms`

노드 통계 4 종

카운터	의미
처리	그 노드로 들어와 정상 분기로 나간 메시지 수
에러	`FAILURE` 분기 또는 예외 발생 수
평균 처리 시간	노드 자체 처리 ms (외부 IO 포함)
최근 처리 시간	가장 마지막 한 건 처리 ms

시스템 단위 메트릭

플로우 엔진 전체의 메트릭은 시스템 → 모니터링 화면에서 확인하실 수 있습니다.

메트릭은 JMX 도메인 plantpulse.core.engine 아래 다음 다섯 개로 노출됩니다.

메트릭	종류	의미	보는 법
`FLOW_EXECUTOR_QUEUE_SIZE`	Gauge	플로우 실행 풀의 대기 큐 크기	계속 쌓이면 처리가 인입을 못 따라가는 것
`FLOW_EXECUTOR_ACTIVE_COUNT`	Gauge	실행 풀의 활성 스레드 수	풀 크기에 붙어 있으면 포화
`FLOW_EXECUTOR_DROPPED_COUNT`	Gauge (누적)	풀 포화로 버려진 task 누적 수	0 이 아니면 메시지가 유실된 것 — 가장 먼저 볼 값
`FLOW_DEBUG_QUEUE_SIZE`	Gauge	디버그 이벤트 적재 대기 큐 크기	디버그 모드를 켠 플로우가 많으면 커집니다
`FLOW_EXECUTION_TIME`	Timer	플로우당 처리 시간 분포	평균·최대 추적

flow.engine.* 이름의 메트릭은 없습니다

예전 문서가 flow.engine.queue_depth · in_flight · dispatch_lag_ms · exec_p95_ms · failed_per_min · script_timeout_per_min 을 표로 싣고 있었지만 그런 이름의 메트릭은 존재하지 않습니다. 위 다섯 개가 전부입니다.

(flow.engine.enabled 는 메트릭이 아니라 engine.properties 의 설정 키입니다 — 플로우 엔진 전체를 끄는 마스터 게이트입니다.)

메트릭 기반 알람 등록 패턴

플로우 자체 메트릭을 EQL 알람 또는 CEP → 트리거 로 감시하시는 패턴 4 종.

패턴 A — 에러율 임계 초과

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

패턴 B — 처리 큐 백압 발생

flow.engine.queue_depth > 500 이 1분 이상 지속될 때 알람 — 트리거 발화 속도가 처리 속도를 초과하는 상황.

패턴 C — 특정 플로우의 발화 끊김

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

정상적으로 분당 N건 발화하던 플로우가 15분 이상 침묵하면 알람.

패턴 D — 외부 IO 타임아웃 폭증

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

알람 출력 채널 권장

알람 종류	권장 채널
에러율·발화 끊김 (운영 직접)	이메일 + 푸시 알림
큐 백압·타임아웃 폭증 (시스템)	Slack/Teams webhook
자동 진단 (참고)	진단 로그만

⚠️ 알람용 플로우 자체에는 절대 알람을 거는 메트릭에 의존하지 마세요 — 알람 플로우가 죽으면 알람 자체가 안 옵니다. 알람용 플로우는 시스템 → 모니터링 의 외부 헬스 체크로 감시하세요.

메시지 처리 의미론과 백압

플로우 엔진이 메시지를 처리할 때의 보장 수준과 백압(backpressure) 동작.

전달 보장 — At-Least-Once

플로우 엔진은 at-least-once 전달을 보장합니다.

케이스	동작
정상 처리	한 번 발화 → 한 번 그래프 통과 → 한 번 완료
처리 중 엔진 재시작	트리거 큐에 남은 메시지는 다음 부팅 후 다시 처리됨
노드 단위 예외	`FAILURE` 분기로만 전이. 메시지가 사라지지 않음
외부 IO 타임아웃	`FAILURE` 분기 + `flow_retry` 로 자동 재시도 가능

중복 가능성 — 정확히 한 번(exactly-once) 이 아닙니다. flow_create_* 가 중간에 끊겼다 재시도되면 같은 ID 가 두 번 들어올 수 있으니, on_duplicate=skip 또는 외부 시스템의 멱등성 키를 사용해 주세요.

멱등성 키 패턴

플로우에서 외부 시스템을 호출할 때 멱등성을 보장하는 패턴.

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

이후 외부 호출 시 헤더로 첨부:

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

처리 순서 — 트리거 단위 FIFO

같은 originator 의 메시지	다른 originator
트리거 단위 FIFO (도착 순서대로)	병렬 처리 (순서 무관)

자산 A 의 이벤트 2건이 거의 동시 발생해도, A 의 이벤트는 발생 순서대로 처리됩니다. 자산 A 와 자산 B 의 이벤트는 서로 다른 워커에서 병렬 처리될 수 있습니다.

백압(Backpressure)

처리 속도가 발화 속도를 따라가지 못할 때의 동작.

상황	동작
큐 깊이 < 80%	정상 — 신규 트리거 즉시 enqueue
큐 깊이 80%~100%	진단 WARN 로그 발생 — 큐는 계속 수용
큐 깊이 = 100% (가득 참)	신규 트리거를 drop + 진단 ERROR 로그 발생

큐가 가득 찰 때 운영자가 할 일

시스템 → 모니터링 에서 flow.engine.queue_depth 확인
목록 화면 에서 실행 건수가 폭증한 플로우 식별
그 플로우의 트리거 사전 필터(*_pattern)를 좁혀 부하 감소
또는 그 플로우의 배포를 일시 해제하여 비상 처리

서킷 브레이커 (외부 IO)

외부 IO 노드(HTTP/Kafka/MQTT/이메일)는 다음 조건이 충족되면 30초 동안 일괄 차단됩니다.

조건	임계
최근 1분 내 연속 실패	≥ 10건
평균 응답 시간	≥ 10초

차단 동안 그 노드로 들어오는 모든 메시지는 즉시 FAILURE 로 분기됩니다. 외부 시스템이 응답을 회복하면 자동으로 차단 해제. 이 동작은 외부 장애가 플로우 엔진 자체를 마비시키지 않도록 격리하는 역할을 합니다.

서킷이 열린 시점은 진단 로그에 code = CIRCUIT_OPENED 로 기록됩니다. 외부 시스템이 빨리 회복했다면 실행 이력 의 metadata.retry_count 를 보고 누락된 메시지를 수동 재실행할 수 있습니다.

사이클(순환) 방지

플로우 안에 다음 같은 사이클이 생기면 무한 루프가 발생할 수 있습니다.

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

플로우 엔진은 두 가지 방어선으로 사이클을 차단합니다.

방어선	동작
`max_depth=100`	노드 100 개를 거치면 강제 종료
`max_revisit=3`	같은 노드를 4번째 방문하는 순간 메시지 폐기 + 진단 ERROR

flow_retry 의 SUCCESS 분기 → 원래 노드 루프는 의도적 사이클이며, metadata.retry_count 가 함께 증가하므로 max_revisit 이 발동하기 전에 정상 종료됩니다.

실행 이력에서 무엇을 보게 되나

예전 문서의 "에러 코드 카탈로그" 는 실재하지 않았습니다

NODE_SCRIPT_TIMEOUT · FLOW_TIMEOUT · FLOW_MAX_DEPTH · TRIGGER_QUEUE_FULL · WEBHOOK_AUTH_FAIL · DOMAIN_DUP_KEY 같은 코드가 표로 실려 있었지만, 제품 어디에서도 그 문자열을 만들지 않습니다. 로그에서 그 코드로 검색하면 영원히 0건입니다.

플로우 엔진이 실제로 남기는 것은 아래 이벤트 타입 5종입니다.

플로우 실행 이력(mm_flow_log, 좌측 메뉴 Automation > 플로우 실행 이력)에 적재되는 이벤트는 다섯 가지입니다.

이벤트 타입	언제	적재 조건
`FLOW_START`	플로우 진입 시	항상
`FLOW_END`	플로우 처리 종료 (성공·타임아웃 모두)	항상
`NODE_IN`	노드가 메시지를 받을 때	디버그 모드일 때만
`NODE_OUT`	노드가 메시지를 내보낼 때	디버그 모드일 때만
`NODE_ERROR`	노드 처리 중 예외	항상

각 행이 함께 갖는 필드입니다.

필드	내용
`level`	`INFO` / `WARN` / `ERROR`
`flow_id` · `flow_node_id` · `node_name` · `node_type`	어느 플로우의 어느 노드인지
`msg_id`	메시지 단위 추적용 — 한 메시지의 흐름을 이어 보려면 이 값으로 묶습니다
`message` · `error_message`	사람이 읽는 설명, 예외 메시지
`duration_ns`	노드 실행 시간 (나노초) — `NODE_OUT` · `NODE_ERROR` 에서만 의미가 있습니다

평소에는 노드 단위 로그가 남지 않습니다

NODE_IN · NODE_OUT 은 디버그 모드에서만 적재됩니다. 평소 실행 이력에 FLOW_START · FLOW_END 만 보이는 것은 정상입니다. 노드를 하나씩 따라가며 보려면 그 플로우의 디버그 모드를 켜세요 — 대신 로그량이 크게 늘어납니다.

증상으로 찾기

코드 대신 증상에서 출발합니다.

증상	먼저 볼 곳
트리거는 발화하는데 아무 노드도 안 도는 것 같다	플로우가 배포(토글 ON) 되어 있는지 · 트리거의 `*_pattern` 이 메시지를 걸러내고 있지 않은지
처리가 중간에 끊긴다	한 메시지 30초(`flow_timeout_ms`) 초과. 서버 로그의 `FlowExecutor timeout` 경고로 확인합니다
같은 메시지가 반복 처리된다	사이클. `max_revisit`(3회) 로 자동 차단되지만 결선을 점검하세요
그래프가 너무 길어 끝나지 않는다	`max_depth`(100) 초과. 서브플로우로 분할합니다
노드가 예외를 던진다	실행 이력에서 `NODE_ERROR` 행의 `error_message` 를 봅니다
외부 호출이 실패한다	해당 노드의 `data.response_status` · `data.error` 를 뒤 노드에서 확인합니다

한도값(100 · 3 · 30,000ms)은 ExecutionState 의 기본값입니다.

클러스터·고가용성(HA) 동작

플랫폼이 클러스터 환경으로 설치된 경우, 플로우 엔진은 다음 규칙으로 분산 동작합니다.

노드 역할 분리

역할	동작
리더(Leader)	플로우 그래프 변경(저장/배포/해제)을 직렬 처리하는 단일 노드
워커(Worker)	트리거 발화·메시지 처리를 병렬 실행하는 일반 노드 (모든 노드가 워커 역할 겸함)
스케줄러(Scheduler)	`flow_schedule` cron 평가 담당 — 리더와 동일 노드

리더 노드는 플랫폼 부팅 시 자동 선출되며, 리더가 다운되면 다른 노드 중 하나가 자동으로 리더로 승격합니다. 운영자가 직접 지정할 필요는 없습니다.

메시지 분배 — 자산 단위 일관 라우팅

분배 키	동작
`originator.id` (자산/태그/주문 ID)	같은 자산의 메시지는 항상 같은 워커로 라우팅 — 시퀀스 보장
외부 인입 (`flow_on_webhook` 등)	라운드 로빈 분배

이 규칙으로 자산 A 의 이벤트가 여러 워커에서 동시에 처리되어 순서가 꼬이는 일이 방지됩니다. 결과적으로 자산 단위로는 단일 워커 처리 보장, 자산 간에는 병렬 처리 가 동시에 성립합니다.

리더 장애 시 동작 — 페일오버

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

단계	영향
0~3 초	`flow_schedule` 발화 일시 정지 / 트리거 처리는 영향 없음
3~5 초	새 리더 확정, 스케줄 재개
5 초 이후	정상

cron 발화는 misfire 정책에 따라 한 번 누락된 발화를 즉시 재실행합니다. 운영 중에는 flow_schedule 의 평가 단위가 5초 미만이면 단기 누락이 보일 수 있습니다.

트리거 큐의 영속성

항목	동작
트리거 큐 위치	인-메모리 큐 + 영구 저장소 (트랜잭션 로그)
노드 재시작 시	큐에 남아있던 미처리 메시지는 다음 부팅 후 다시 디스패치
처리 중 노드 다운	그 메시지는 재처리되지만 at-least-once 보장 으로 외부 시스템에 멱등성 키 필요

클러스터 배포 시 운영자 체크리스트

모든 노드의 시간 동기화 (NTP) — 트리거 발화 시각이 노드 간 일치
외부 시스템(MQTT/Kafka 브로커)을 모든 노드가 도달할 수 있는 네트워크 위치에 배치
flow_send_email 의 SMTP 설정은 시스템 설정에 한 번만 등록 — 모든 노드가 공유
워커 풀 크기(flow.engine.workers)는 노드별 CPU 코어 수에 맞게 조정

단일 노드 모드 (개발·소규모)

리더·워커·스케줄러 모두 한 프로세스에서 동작
페일오버 없음 — 노드가 다운되면 플로우 자체가 중단
트리거 큐는 인-메모리 + 디스크로 보장되어 재시작 후 복구

종단간 트레이스

한 메시지가 그래프 전체를 어떻게 통과했는지 사후 추적하는 방법.

자동 상관 ID 부여

플로우 엔진은 모든 트리거 발화 메시지에 상관 ID(correlation ID) 를 자동으로 부여합니다.

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

필드	의미
`metadata.trace_id`	한 트리거 발화 전체에 고유한 ID — 그래프의 모든 노드 통과 메시지가 공유
`metadata.span_id`	노드별 고유 ID — 노드를 거칠 때마다 갱신
`metadata.parent_id`	직전 노드의 `span_id`

실행 이력 화면에서 trace_id 로 검색

실행 이력 화면 에서 trace_id 입력란에 ID 를 붙여넣으면 그 메시지가 거친 모든 노드의 시간 순 로그가 표시됩니다.

🔍 trace_id = tr-a1b2c3d4-...

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

서브플로우 호출 시 trace 전파

flow_subflow 노드로 다른 플로우를 호출하면 같은 trace_id 가 그대로 이어집니다. 즉 메인 플로우 + 호출된 서브플로우의 모든 노드 로그가 같은 trace 로 검색됩니다.

외부 시스템으로 전파

외부 HTTP 호출 시 자동으로 X-Trace-Id 헤더가 첨부됩니다.

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

외부 시스템이 이 헤더를 수신해 로그에 함께 기록하면, 양 시스템의 로그를 같은 ID 로 매칭할 수 있습니다.

알람·이메일에서 trace_id 노출

알람 본문 또는 이메일 템플릿에 ${metadata.trace_id} 를 포함하면 운영자가 알람을 받은 직후 이력 화면에서 즉시 그 사건을 추적할 수 있습니다.

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

트레이스 보존 기간

데이터	보존 기간
trace_id 와 노드별 span 로그	7일 (실행 이력과 동일)
외부 적재 (장기 보관)	감사·이력 추적 의 외부 시스템 미러링 사용

trace_id 가 길어 메시지 크기가 부담되면 flow_script_transform 으로 마지막 4자리(tr-...d4 같은)만 노출하는 짧은 형식을 만들어 알람·이메일에 쓰셔도 됩니다. 단, 검색 시에는 전체 ID 가 필요합니다.

그래프 패턴 카탈로그

플로우 그래프에서 자주 쓰이는 결선 패턴 10 종.

패턴 1 — Pipeline (단순 직렬)

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

| 특징 | 메시지가 노드들을 순서대로 통과 | | 사용 예 | 임계 초과 알람 → 이메일 발송 |

패턴 2 — Fan-out (1 to N)

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

| 특징 | 한 메시지를 여러 액션이 동시 처리 | | 사용 예 | 알람 발생 → 이메일 + SMS + 슬랙 + 워크오더 생성 |

같은 메시지의 복제본이 여러 노드로 분배됩니다. 각 분기는 독립적으로 처리되며, 한 분기 실패가 다른 분기에 영향 없습니다.

패턴 3 — Fan-in (N to 1) — Merge

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

| 특징 | 여러 트리거의 메시지를 시간 윈도우 안에 모아 한 번에 처리 | | 사용 예 | 5분 동안 발생한 모든 알람을 합쳐 일일 리포트 1통 |

패턴 4 — Scatter-Gather (분산 → 수집)

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

| 특징 | 배열을 원소별로 분할 처리 후 결과 재합 | | 사용 예 | 100 건의 외부 주문을 병렬 검증한 뒤 결과 일괄 보고 |

패턴 5 — Switch (조건 분기)

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

| 특징 | 한 메시지를 조건별로 다른 경로로 | | 사용 예 | 알람 우선순위별 처리 채널 분리 |

패턴 6 — Retry with Fallback

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

| 특징 | 일시 장애를 자동 재시도, 영구 장애는 대체 처리 | | 사용 예 | 외부 API 실패 시 3회 재시도, 그래도 실패면 사람에게 통지 |

패턴 7 — Circuit Breaker (서킷 차단 활용)

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

| 특징 | flow_throttle 로 호출 속도 제한 + 서킷 브레이커로 일괄 차단 | | 사용 예 | 외부 시스템 과부하 보호 |

패턴 8 — Dead Letter Queue (DLQ)

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

| 특징 | 영구 실패한 메시지를 별도 저장소로 | | 사용 예 | 외부 ERP 동기화 실패 메시지 모음 (수동 재처리 대상) |

패턴 9 — Sliding Window Aggregation

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

| 특징 | 최근 N건 윈도우 안의 상태로 의사결정 | | 사용 예 | 최근 5분 알람 30건 초과 시 사이트 비상 모드 |

패턴 10 — Saga (다단계 트랜잭션)

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

| 특징 | 여러 외부 시스템에 걸친 작업의 부분 실패를 보상으로 처리 | | 사용 예 | 워크오더 생성 → 자산 예약 → ERP 동기화 → 작업자 통보 (한 단계 실패 시 앞 단계 모두 취소) |

패턴 선택 가이드

요구사항	권장 패턴
단순 임계 알람	Pipeline (1)
한 사건 → 여러 채널	Fan-out (2)
여러 사건 → 한 요약	Fan-in / Merge (3)
배열 일괄 처리	Scatter-Gather (4)
분기 처리	Switch (5)
외부 API 신뢰성	Retry (6)
외부 시스템 보호	Circuit (7)
실패 메시지 보관	DLQ (8)
최근 N건 누적 의사결정	Sliding (9)
여러 외부 시스템 일관성	Saga (10)

플로우 테스트 모범 사례

플로우를 안전하게 변경·배포하기 위한 테스트 전략.

3 단계 테스트 — 단위 → 통합 → 시뮬레이션

단계	도구	검증 대상
① 단위	편집 화면 ▶ 테스트 실행	한 노드 단독 동작 (스크립트 식, 외부 호출 응답)
② 통합	같은 화면, 임시 페이로드 + 라이브 디버그 ON	그래프 전체 시퀀스·분기
③ 시뮬레이션	배포 + 실제 트리거 대기 (스테이징 환경)	실 메시지 흐름·외부 시스템과의 결합

테스트 실행 페이로드 라이브러리

테스트 실행 다이얼로그에 붙여넣을 수 있는 트리거별 페이로드 예제는 트리거별 페이로드 예제 섹션 참고. 운영 환경에서 자주 발생하는 경계 케이스도 미리 만들어 두세요.

경계 케이스 예시

케이스	페이로드
null 필드	`{"value": null}` — 스크립트가 null 안전한지
빈 문자열	`{"value": ""}` — 빈 문자열을 0 으로 해석하는지 검증
음수	`{"value": -1}` — 임계 검사가 ± 부호 의도대로
거대 숫자	`{"value": 1e20}` — 오버플로/정밀도
유니코드	`{"name": "한글-Émoji-🚀"}` — 외부 시스템 인코딩

변경 안전 절차

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

회귀 테스트 시나리오 보관

자주 사용되는 시나리오를 텍스트 파일로 보관하시고, 변경 후 반드시 재실행하시기를 권장합니다.

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

외부 시스템 모의 (Mock) 패턴

스테이징 환경에서 외부 시스템 응답을 모의하려면:

방법	설명
flow_http_request 의 url 을 모의 서버로	운영 URL 을 스테이징 URL 변수로 분리
flow_jdbc_poll 의 datasource 변경	운영 DB → 스테이징 DB 로만 변경
flow_send_email 의 to_field 를 fake@example.com 로	실수 발송 방지

카운터 초기화 후 부하 테스트

신규 버전 플로우의 카운트 초기화 (전체)
5분 동안 정상 트리거 발화 시키기
목록 화면 에서 처리/에러 카운트, 평균 처리 시간 확인
p95 처리 시간이 기존 대비 20% 이상 늘면 원인 분석 후 롤백 검토

성능 한계와 튜닝

플로우 엔진의 처리 한계와 튜닝 팁입니다.

기본 한계

항목	기본값	비고
한 메시지의 노드 방문 수 (`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_tag_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 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]

이후 일일 통계 화면에서 stats_v1 vs stats_v2 누적 결과 비교.

A/B 비교 자동 분석

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

전략 선택 가이드

상황	권장 전략
새 자동화 시나리오 첫 출시	Canary — 한 라인에서 검증 후 확대
기존 로직 큰 변경 (구조 개편)	Blue-Green — 즉시 롤백 가능
두 알고리즘 중 어느 게 더 나은지 측정	A/B 테스트
단순 옵션값 조정	직접 변경 — `metadata.audit_diff` 메모 후 1주 모니터링

롤백 절차 (공통)

문제 발견 시 즉시 롤백:

목록 화면 에서 새 버전 해제 토글
(Canary/A·B 의 경우) 트리거 패턴을 0건 매칭으로 변경
기존 버전이 단독으로 동작하는지 5분 모니터링
진단 로그에서 code=FLOW_NOT_DEPLOYED 가 없는지 확인
원인 분석 — 실행 이력 에서 trace_id 로 실패 메시지 추적

출시 사전 체크리스트 (압축판)

신규 플로우 배포 체크리스트 의 핵심만 한 화면:

[ ] 테스트 실행으로 정상·경계·실패 시나리오 모두 통과
[ ] 외부 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 표 참고)
`flow_update_*`	부분 갱신: 입력한 필드만 갱신, 빈 값은 무시. 전체 덮어쓰기는 Delete + Create 조합
`flow_delete_*`	`*_id_field`
`flow_start/end/pause/resume_work_order`	`order_id_field` (기본 `data.order_id`)
`flow_abort_work_order`	+ `abort_code_field`, `abort_notes_field`
`flow_update_tag_alarm_band_numeric`	`tag_id_field`, `hi_field` 등 (입력 필드만 갱신)

엣지 빠른 설정

엣지 노드는 모두 동일한 옵션 셋을 사용합니다.

옵션	설명
`url`	엣지 REST 엔드포인트 (예: `http://edge.local:60000/opc/server`)
`method`	HTTP 메서드 (미설정 시 노드별 기본값)
`headers`	JSON 헤더 (인증 토큰 등)
`body_template`	요청 본문 (미설정 시 `data` 그대로 전송)
`timeout_ms`	5000

플로우 REST API — 프로그램에서 플로우 조작

플로우를 외부 자동화 도구(Ansible/GitOps/CI 파이프라인)에서 코드로 관리하거나, 외부 시스템이 플로우를 즉시 실행시키고 싶을 때 사용하는 REST API.

인증

모든 API 호출은 보안 → API 인증 토큰 에서 발급한 토큰을 헤더로 첨부합니다.

Authorization: Bearer {api_token}

엔드포인트 목록

1) 플로우 목록 조회

GET /flow/list

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

응답:

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

2) 단일 플로우 조회

GET /flow/get/{flow_id}

응답 본문에 그래프 노드·관계·옵션이 모두 포함됩니다. 백업/버전 관리에 그대로 사용 가능합니다.

3) 플로우 생성

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

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

응답에서 flow_id 를 받아 후속 호출에 사용합니다.

4) 플로우 수정 (메타)

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

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

5) 그래프 저장 (노드·관계 일괄 교체)

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

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

그래프 저장은 트랜잭션 — 검증 실패 시 그래프 전체가 롤백됩니다.

6) 배포 / 해제

플로우 메타의 deployed 필드를 변경하면 자동 배포·해제됩니다.

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

7) 즉시 실행 (수동 트리거)

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

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

트리거 노드의 사전 필터를 거치지 않고 첫 노드부터 즉시 실행됩니다. 수동 검증·디버깅 시 유용.

8) Dispatch (트리거 우회 발화)

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

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

정상 트리거 처리 경로로 메시지를 인입시킵니다 — 해당 메시지에 매칭되는 모든 플로우가 동시에 발화합니다.

9) 통계 초기화

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

10) Export / Import — 그래프 백업

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

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

Import 시 충돌 ID 가 있으면 새 ID 로 자동 발급됩니다. 외부 의존성(자격 증명·자산 ID 등)은 import 후 따로 매핑해 주세요.

11) 모든 플로우 재배포

POST /flow/redeploy

대용량 변경 후 또는 노드 재시작 후 일괄 동기화에 사용. 운영 중 호출은 신중히 — 일시적 처리 지연이 발생합니다.

12) 자가 진단

POST /flow/selftest

플로우 엔진의 내부 컴포넌트(트리거 큐·디스패처·노드 레지스트리·스크립트 런타임)를 한 번 점검하고 상태를 반환. 결과는 진단 로그에도 기록됩니다.

13) 노드 카탈로그 조회

GET /flow/catalog

현재 등록된 모든 노드 타입·옵션 스키마를 반환. UI 가 인스펙터를 동적 생성하는 데 사용. 외부 도구가 그래프를 자동 생성할 때도 참고.

Webhook 트리거로 외부에서 발화

flow_on_webhook 트리거가 포함된 플로우는 다음 URL 로 외부 시스템이 직접 발화할 수 있습니다.

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

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

응답:

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

응답 코드	의미
200	메시지 큐에 enqueue 됨 (실제 처리 결과는 비동기)
401	인증 토큰 잘못/누락
404	플로우 ID 없거나 미배포
413	본문 256KB 초과
422	트리거 노드가 `flow_on_webhook` 이 아님
429	분당 호출 한도 초과

외부 자동화 도구 연동 예시

GitHub Actions — PR 머지 시 플로우 배포

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

Ansible — 플로우 배치 관리

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

Jenkins — 새 빌드마다 selftest

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

API Rate Limit

엔드포인트	분당 한도 (토큰당)
`GET /flow/list` · `get` · `catalog`	600
`POST /flow/create` · `update` · `delete`	60
`POST /flow/{id}/run` · `dispatch` · `webhook/{id}`	1,000
`POST /flow/redeploy` · `selftest`	10

한도 초과 시 429 Too Many Requests 응답.

OPC/PLC 산업 통합 패턴

산업 현장 특유의 통합 시나리오 모음. 엣지 노드 22종 과 자산 이벤트 발행 4종 의 조합 활용.

패턴 A — OPC 서버 자동 등록

새 라인이 가동될 때 ERP 에서 라인 정보를 받아 엣지에 OPC 서버를 자동 등록.

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

엣지 노드는 자동으로 mm_edge 마스터 테이블에서 api_key 를 lookup 합니다. 운영자가 별도 인증 설정을 하지 않아도 됩니다.

패턴 B — PLC 명령 자동 발행

자산 알람이 발생하면 PLC 에 정지 명령을 자동으로 발행.

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

flow_publish_asset_command 는 자산의 명령 토픽(asset/{id}/cmd/STOP)으로 즉시 전달되어 엣지가 PLC 에 OPC Write 합니다.

패턴 C — OPC 태그 일괄 등록 (CSV/Excel 인입)

새 설비 셋업 시 CSV 의 태그 100~1000건을 한 번에 등록.

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

1,000 건 등록 시 flow_throttle 로 분당 100 건 이하로 평탄화하시는 것을 권장합니다. 엣지가 일시적으로 응답이 느려질 수 있습니다.

패턴 D — OPC 끊김 자동 복구

OPC 연결이 끊겼다가 회복되면 자동으로 재시작.

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

패턴 E — 시프트 시작 시 라인 상태 점검

매 시프트 시작 시각에 사이트의 모든 라인을 점검.

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

패턴 F — 시프트 변경 시 작업자 자동 매핑

근무표 변경에 따라 자동으로 워크오더에 작업자 매핑.

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

패턴 G — 다중 PLC 동시 호출 (Scatter-Gather)

여러 PLC 의 데이터를 동시 호출 후 결과 통합.

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

패턴 H — 알람밴드 자동 조정

예측 분석 의 학습된 LIMIT_MIN/MAX 값을 알람밴드에 자동 적용.

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

패턴 I — 다운타임 자동 분류

자산 정지 이벤트를 사유별로 분류해 OEE 가용성에 정확히 반영.

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

패턴 J — 외부 ERP 와 양방향 작업지시 동기화

플랜트펄스 ↔ 외부 ERP 양방향 미러링.

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

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

무한 루프 방지를 위해 외부에서 인입된 메시지에는 metadata.source='ERP' 를 표시하고, 반대 방향 플로우에서 그 메시지는 필터링하세요.

산업 통합 시 주의 사항

사항	권장
OPC Write 명령은 안전 인터록 후 발행	운전원 확인 또는 자동 안전 룰 통과 후에만
PLC 폴링 주기를 너무 짧게 설정하지 마세요	1초 이하는 PLC CPU 점유율 급증 위험
엣지 디바이스의 도커 컨테이너 자원 한도	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 시스템의 작업지시 테이블에 직접 행 삽입.

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

노드 옵션	값
`datasource_id`	mes_db (시스템 설정에 사전 등록)
`query`	위 SQL
`binds`	`{"order_no":"${data.order_no}","asset_id":"${metadata.asset_id}",...}`

flow_jdbc_query 는 SELECT 외에 INSERT/UPDATE/DELETE 도 지원합니다. 트랜잭션은 노드 단위로 자동 commit/rollback 됩니다.

Grafana — 대시보드 자동 갱신 알림

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

알람 발생 시 그래프 위에 마커가 자동 표시됩니다 — 사후 분석 시 매우 유용.

Telegram — 봇 메시지

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

REST 인증 패턴 빠른 표

인증 방식	헤더
API Key (헤더)	`X-API-Key: {token}`
API Key (쿼리)	URL 끝에 `?api_key={token}`
Bearer Token	`Authorization: Bearer {token}`
Basic Auth	`Authorization: Basic {base64(user:pass)}`
OAuth2 (Client Credentials)	토큰 발급 → Bearer 헤더 (별도 노드로 갱신)
HMAC 서명	본문/시각 기반 서명을 스크립트로 계산해 `X-Signature` 헤더

응답 파싱 패턴

flow_http_request 응답 후 data.response_body 를 다음 노드에서 활용:

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

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

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

외부 시스템 호출 시 보안 권장

사항	권장
토큰을 그래프 안에 평문으로 두지 마세요	시스템 설정 → 자격 증명 저장소에 등록 후 `${creds.slack_webhook}` 처럼 참조
응답 본문에 민감 정보가 있으면 마스킹	`flow_script_transform` 으로 PII/토큰 제거 후 다음 노드로
외부 시스템마다 별도 retry 정책	호출 빈도가 높은 곳은 별도 `flow_retry` + `flow_throttle` 결선
호출 결과를 감사 로그로 저장	`flow_save_attributes` 로 자산에 호출 이력 메타 추가

데이터 변환 Cookbook

flow_script_transform 노드에 그대로 옮겨 쓰실 수 있는 변환 패턴 모음. 모든 예제는 JavaScript 기준이며 data / metadata 단축 별칭을 사용합니다.

단위 변환

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

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

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

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

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

msg

날짜·시각 변환

const d = new Date(metadata.ts);

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

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

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

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

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

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

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

msg

문자열 정규화

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

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

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

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

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

msg

중첩 객체 평탄화

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

평탄 객체 → 중첩

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

CSV 행 생성

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

배열 집계

var arr = data.values || [];

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

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

msg

조건부 필드 추가 (스키마 진화)

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

페이로드 일부 마스킹

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

다국어(i18n) 템플릿

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

JSON Path 안전 조회

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

메시지 병합 (`flow_merge` 후처리)

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

재사용 스크립트 모음

flow_script_filter / flow_script_transform / flow_switch 노드에 그대로 옮겨 쓰실 수 있는 스크립트 라이브러리입니다. 모든 스크립트는 JavaScript 기준이며 data / metadata 단축 별칭을 사용합니다.

변환 스크립트

단위 변환·라벨링

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

시각·시프트·요일 메타 보강

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

평균 ± 3σ 알람밴드 계산

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

MES PO → 작업지시 매핑

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

외부 페이로드 정규화 (다양한 필드명 통일)

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

임계 위반 횟수 누적 (Stateful 패턴)

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

메시지 통계 요약 (Merge 결과 가공)

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

시간대별 임계값 적용

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

필터 스크립트

우선순위 화이트리스트

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

시간 윈도우 (주간만)

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

자산·태그 ID 패턴

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

임계값 + 안정성 (연속 N회)

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

영업시간·휴일 체크

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

특정 사이트만

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

필수 필드 모두 존재 검사

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

자기 발행 메시지 차단 (양방향 동기화)

metadata.source !== 'flow'

스위치 케이스 스크립트

우선순위 등급별

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

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

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

자산 상태별

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

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

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

작업지시 라이프사이클별

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

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

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

위 스크립트는 모두 운영 환경에서 자주 사용되는 패턴을 모은 것입니다. 그래프 자체에 직접 결선해도 동작하며, 도메인에 맞춰 임계값·필드명만 조정하시면 됩니다.

용어 사전

플로우 엔진 용어

용어	의미
entity_type	메시지의 주체 엔티티 종류 — `Asset`, `Tag`, `Site`, `Order`, `Customer`, `Product`, `Employee`, `Calendar` 등
originator	메시지가 가리키는 주체 엔티티 (`entity_type` + `id`). 예: `Asset/MOTOR-001`
type	메시지 분류 라벨. 트리거가 어떤 종류의 이벤트를 받았는지 식별합니다
data	메시지의 본문(페이로드) — 변환·액션 노드가 주로 읽고 씁니다
metadata	메시지의 컨텍스트(시각·사이트·시프트·태그 ID 등) — 변환되더라도 흐름 끝까지 유지됩니다
relation	노드 출력 와이어의 라벨. `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` (모두 대문자)
*`_field` 동적 옵션**	정적 값 대신 메시지 페이로드의 경로(예: `data.tag_id`)에서 값을 읽는 입력 방식
글롭 패턴	트리거 `_pattern` 옵션의 와일드카드 표현 — ``는 0자 이상, `?`는 정확히 1자에 매칭
스냅샷	그래프 저장 시 자동 적재되는 버전 단위 백업. 사고 시 이전 상태로 되돌릴 때 사용
부분 갱신(fetch+merge)	Update 노드가 기존 레코드 조회 후 입력한 필드만 병합하는 방식. 빈 값은 무시
SKIPPED	트리거 패턴 미매칭 메시지가 후속 노드로 흐르지 않고 카운터에서도 제외되는 처리
EXHAUSTED	`flow_retry` 노드가 최대 재시도 횟수를 초과했을 때 발화하는 분기
THROTTLED	`flow_throttle` 노드가 윈도우 내 한도를 초과한 메시지를 차단할 때 발화하는 분기

도메인·약어 사전

플랜트 운영 환경에서 자주 쓰이는 약어를 매뉴얼 안에서 빠르게 참조하실 수 있도록 정리했습니다.

약어	뜻	본 매뉴얼에서
MES	Manufacturing Execution System — 작업지시·생산 실적 관리	외부 연동 대상 (HTTP/MQTT/외부 DB)
ERP	Enterprise Resource Planning — 전사 자원·계획 시스템	외부 연동 대상
SCADA	Supervisory Control and Data Acquisition — 감시·제어	외부 연동 대상 / `flow_publish_asset_command` 의 출구
OPC	Open Platform Communications — 산업 통신 표준	엣지 카테고리(`flow_edge_opc_*`)의 대상
OEE	Overall Equipment Effectiveness — 가동률 × 성능 × 품질	`flow_on_oee_event` 트리거
RAM	Reliability·Availability·Maintainability	`flow_on_ram_event` 트리거
EMS	Energy Management System	`flow_on_ems_event` 트리거
EQL	도메인 이벤트 룰 표현식	`flow_script_filter`/`flow_script_transform` 의 `language: "EQL"` 옵션
CEP	복합 이벤트 처리 (Complex Event Processing)	EQL 기반 룰 엔진. 알람은 CEP 경로로만 발생해야 정합성 유지
PO	Purchase Order — 구매·생산 주문	MES 연동 시 작업지시로 변환되는 단위
WO	Work Order (작업지시)	`flow_*_work_order` 노드 군
CMMS	Computerized Maintenance Management System	외부 연동 대상
MTTF/MTTR	Mean Time To Failure / To Repair	RAM 이벤트의 핵심 지표
HMI	Human–Machine Interface	SCADA 등의 운영 화면

버전 노트

매뉴얼이 현재 시점에서 다루는 주요 기능군의 도입 시점입니다. 이전 버전을 운영 중이시라면 일부 기능이 다르게 동작할 수 있습니다.

V2026.05 — 도메인 자동화 확장

엣지(Edge) 카테고리 11종 신설 — OPC 서버/태그 CRUD + 모니터링 자동화
작업지시 상태 전이 5종 — flow_start/end/pause/resume/abort_work_order
태그 알람밴드 부분 갱신 2종 — 수치형/불린형 알람밴드를 입력 필드만 갱신
flow_retry 흐름 제어 신규 — 백오프 + 최대 시도 + EXHAUSTED 분기
새 트리거 5종 — flow_on_asset_health_status / _connection_status / _oee_event / _ram_event / _ems_event
Update 노드 부분 갱신(fetch+merge) — 기존 레코드 조회 후 입력 필드만 병합
Relations 대문자 표준화 — SUCCESS/FAILURE/... 모두 대문자, 기존 그래프는 자동 변환
SKIPPED 처리 — 트리거 패턴 미매칭 메시지를 카운터에서 제외
모두 재배치(redeploy) 운영 도구 — 활성 플로우 일괄 재로드
전체 카운트 초기화 — 노드 + 플로우 단위 통계 일괄 리셋

V2026.03 — 릴리즈 안정성

그래프 저장 시 자동 스냅샷 적재
라이브 디버그 패널 (인스펙터 하단 2초 갱신) + 노드 점등·duration 표시
import/export (그래프 JSON 파일)
부하 워치독 도입 — 부하 지속 시 진단 로그 한 줄 발행

그 이전

M1 — 엔진·UI 골격, 그래프 CRUD, 시각 캔버스
M2 — 도메인 액션 노드(자산·태그·작업지시·생산 도메인 등)
M3 — 트리거 다양화, 필터/변환/외부 연동 8종, 스크립트 엔진
M4 — 디버그 적재·실행 이력 화면, 통계, 배포/해제, 임포트/익스포트
M5 — 도메인 트리거 통합, 시나리오 일괄 검증

이 매뉴얼이 다루는 모든 기능과 동작은 위 V2026.05 시점을 기준으로 합니다.

목차​

개요​

학습 로드맵 — 역할별 추천 진입 순서​

👶 처음 사용자 (1시간 내 첫 플로우 만들기)​

🧑‍🏭 현장 운영자 (자동화 시나리오 작성)​

🛠 시스템 관리자 (운영·튜닝·장애 대응)​

🔌 개발자·통합 엔지니어 (외부 시스템 연동)​

🔐 보안 담당 (감사·인증)​

📚 빠른 참조 (이미 익숙한 사용자)​

핵심 개념​

처음 시작하기​

엔드 투 엔드 튜토리얼​

단계 1 — 새 플로우 만들기​

단계 2 — 트리거 노드 배치​

단계 3 — 임계값 필터​

단계 4 — 도메인 변경 (태그 → 자산)​

단계 5 — 작업지시 발행​

단계 6 — 이메일 알림 (성공 분기)​

단계 7 — 실패 분기 처리​

단계 8 — 저장과 테스트 실행​

단계 9 — 검증과 배포​

단계 10 — 운영 모니터링​

다음 단계​

화면 구성​

목록 화면​

상단 도구​

일람 테이블​

일괄 배포·해제​

모두 재배치​

편집 화면 (시각 캔버스)​

상단 툴바​

테스트 실행 사용법​

좌측 — 노드 팔레트​

중앙 — 캔버스​

우측 — 노드 설정​

카운트 초기화​

우측 — 라이브 디버그​

메시지 구조​

메시지 타입​

트리거별 페이로드 예제​

flow_on_tag_point — 태그 포인트 인입​

flow_on_tag_alarm — 태그 알람 발생​

flow_on_asset_data / flow_on_asset_event — 자산 이벤트​

flow_on_asset_health_status / flow_on_asset_connection_status — 주기 평가​

flow_on_oee_event / flow_on_ram_event / flow_on_ems_event — 플러그인 이벤트​

flow_on_opc_status / flow_on_edge_status — OPC/엣지 상태​

flow_on_diagnostic — 시스템 진단 메시지​

flow_on_domain_changed — 도메인 변경 이벤트​

flow_on_entity_event — 엔티티 라이프사이클​

flow_on_webhook — 외부 HTTP 푸시​

flow_on_mqtt_subscribe — MQTT 토픽 구독​

flow_jdbc_poll — 외부 DB 주기 폴링​

flow_schedule — 시간 기반 발화​

페이로드 변환 시 주의 사항​

노드 카탈로그​

트리거 (22종)​

필터 (6종)​

변환 (8종)​

액션 — 통합·저장 (2종)​

액션 — 자산 이벤트 발행 (4종)​

flow_asset_command_invoke — 발행이 아니라 실행​

액션 — 도메인 CRUD (34종)​

태그 알람밴드 부분 갱신 (2종)​

작업지시(WorkOrder) 상태 전이 (5종)​

엣지 (Edge, 22종)​

조회 3종​

도커 앱 제어 7종​

외부 연동 (9종)​

흐름 제어 (7종)​

Retry 권장 결선 패턴​

노드 옵션 상세 레퍼런스​

flow_script_filter — 스크립트 필터​

스크립트 컨텍스트​

예시​

flow_script_transform — 스크립트 변환​

예시​

flow_switch — 다중 분기​

예시​

flow_retry — 자동 재시도​

metadata 자동 보강​

목차

개요

학습 로드맵 — 역할별 추천 진입 순서

👶 처음 사용자 (1시간 내 첫 플로우 만들기)

🧑‍🏭 현장 운영자 (자동화 시나리오 작성)

🛠 시스템 관리자 (운영·튜닝·장애 대응)

🔌 개발자·통합 엔지니어 (외부 시스템 연동)

🔐 보안 담당 (감사·인증)

📚 빠른 참조 (이미 익숙한 사용자)

핵심 개념

처음 시작하기

엔드 투 엔드 튜토리얼

단계 1 — 새 플로우 만들기

단계 2 — 트리거 노드 배치

단계 3 — 임계값 필터

단계 4 — 도메인 변경 (태그 → 자산)

단계 5 — 작업지시 발행

단계 6 — 이메일 알림 (성공 분기)

단계 7 — 실패 분기 처리

단계 8 — 저장과 테스트 실행

단계 9 — 검증과 배포

단계 10 — 운영 모니터링

다음 단계

화면 구성

목록 화면

상단 도구

일람 테이블

일괄 배포·해제

모두 재배치

편집 화면 (시각 캔버스)

상단 툴바

테스트 실행 사용법

좌측 — 노드 팔레트

중앙 — 캔버스

우측 — 노드 설정

카운트 초기화

우측 — 라이브 디버그

메시지 구조

메시지 타입

트리거별 페이로드 예제

`flow_on_tag_point` — 태그 포인트 인입

`flow_on_tag_alarm` — 태그 알람 발생

`flow_on_asset_data` / `flow_on_asset_event` — 자산 이벤트

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 주기 평가

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — 플러그인 이벤트

`flow_on_opc_status` / `flow_on_edge_status` — OPC/엣지 상태

`flow_on_diagnostic` — 시스템 진단 메시지

`flow_on_domain_changed` — 도메인 변경 이벤트

`flow_on_entity_event` — 엔티티 라이프사이클

`flow_on_webhook` — 외부 HTTP 푸시

`flow_on_mqtt_subscribe` — MQTT 토픽 구독

`flow_jdbc_poll` — 외부 DB 주기 폴링

`flow_schedule` — 시간 기반 발화

페이로드 변환 시 주의 사항

노드 카탈로그

트리거 (22종)

필터 (6종)

변환 (8종)

액션 — 통합·저장 (2종)

액션 — 자산 이벤트 발행 (4종)

`flow_asset_command_invoke` — 발행이 아니라 실행

액션 — 도메인 CRUD (34종)

태그 알람밴드 부분 갱신 (2종)

작업지시(WorkOrder) 상태 전이 (5종)

엣지 (Edge, 22종)

조회 3종

도커 앱 제어 7종

외부 연동 (9종)

흐름 제어 (7종)

Retry 권장 결선 패턴

노드 옵션 상세 레퍼런스

`flow_script_filter` — 스크립트 필터

스크립트 컨텍스트

예시

`flow_script_transform` — 스크립트 변환

예시

`flow_switch` — 다중 분기

예시

`flow_retry` — 자동 재시도

`metadata` 자동 보강