> For the complete documentation index, see [llms.txt](https://kopens.gitbook.io/plantpulse-platform/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://kopens.gitbook.io/plantpulse-platform/developer/flow.md). # 플로우 엔진 (Flow Engine) ## 개요 플로우 엔진은 룰 체인 모델을 PlantPulse 도메인에 맞춰 재설계한 시각적 워크플로우 런타임입니다. 운영자가 노드를 캔버스 위에 배치하여 데이터 처리 파이프라인을 정의하면, 엔진이 도메인 이벤트를 받아 그래프를 BFS로 순회하며 메시지를 라우팅합니다. ### 한눈에 보는 동작 흐름 ```mermaid flowchart LR EVT[도메인 이벤트
POST_TELEMETRY · ALARM · WEBHOOK ...] BR[FlowEventBridge
이벤트 → FlowMsg] TR[Trigger Node
FlowTriggerDefinition 매핑] EX[FlowExecutor
BFS 순회] subgraph FLOW["Flow 그래프"] N1[FILTER] N2[TRANSFORMATION] N3[ACTION] N4[EXTERNAL] end EVT --> BR --> TR --> EX EX --> N1 --> N2 --> N3 --> N4 EX -.-|tellSuccess
tellFailure
tellNext| EX ``` ### 핵심 컴포넌트 * **엔진 코어**: `plantpulse.core.engine.flow` (\~5,000 LOC) * **노드 패키지**: `plantpulse.core.engine.flow.node.impl.*` (\~4,000 LOC, 100여 종) * **트리거 정의**: `plantpulse.core.engine.flow.trigger.FlowTriggerDefinition` (메시지 타입 ↔ 트리거 노드 매핑 single source of truth) * **컨트롤러/서비스**: `plantpulse.server.mvc.flow.FlowController`, `plantpulse.core.service.FlowService` * **부하 진단**: `plantpulse.core.diagnostic.LoadHealthCheckTimer` * **UI**: `WebContent/WEB-INF/jsp/flow/*.jsp` + `WebContent/public/scripts/flow/*.js` (8개 모듈로 분할) 사용자 시각의 사용법은 [사용자 가이드 — 플로우](/plantpulse-platform/user/flow.md)를 참조하세요. ### 노드 카테고리 (7종) ```mermaid graph LR TRIG[Trigger
진입점] --> FILT[Filter
조건 분기] FILT --> TRANS[Transformation
데이터 변환] TRANS --> ACT[Action
실행 / 발행] TRANS --> EXT[External
외부 호출] ACT --> CTRL[Control
지연 / 락 / 분기] EXT --> EDGE[Edge
엔진 종점] CTRL --> EDGE ``` | 카테고리 | 역할 | 대표 노드 | | ------------------ | ------------------ | --------------------------------------------- | | **Trigger** | 외부 이벤트 진입점 | TagPointTrigger, AlarmTrigger, WebhookTrigger | | **Filter** | 조건 분기 (TRUE/FALSE) | TagValueFilter, SiteFilter, ScriptFilter | | **Transformation** | 페이로드 변환 | TagAggregator, JsonTransform, ScriptTransform | | **Action** | 부수 효과 실행 | CreateAlarm, PublishMQ, ExecuteCEP | | **External** | 외부 시스템 호출 | RestCall, KafkaPublish, DBQuery | | **Control** | 흐름 제어 | Delay, Lock, RoundRobin, Throttle | | **Edge** | 엔진 종점 | Acknowledge, Drop, Log | *** ## 핵심 모델 ### Flow * 노드(FlowNode)와 와이어(FlowLink)의 집합 = 하나의 워크플로우 * 진입점이 있는 방향 그래프 (트리거 노드가 시작점) * 활성/비활성 토글 가능 (운영 명칭은 **배포/해제**) * import/export (JSON) 지원 * ID 형식: `FLOW_NNNNN` 시퀀스 ### FlowNode * 메시지를 받아 처리한 뒤 다음 노드로 전달하는 단위 * 7개 카테고리: **Trigger / Filter / Transformation / Action / External / Control / Edge** * 각 노드는 1개 이상의 출력 relation을 가짐 (`SUCCESS`, `FAILURE`, `TRUE`, `FALSE`, `MATCH`, `NO_MATCH`, `DEFAULT`, `THROTTLED`, `EXHAUSTED`, 커스텀 — 모두 대문자 표준) ### Relation (와이어 레이블) * 노드 출력 → 다음 노드 연결의 **레이블 명시 라우팅** * 노드 내부에서 어느 relation으로 보낼지 결정 → 분기 노드를 따로 만들 필요 없음 ### FlowMsg (메시지) ```java class FlowMsg { FlowMsgType type; // POST_TELEMETRY / ALARM / WEBHOOK ... FlowOriginator originator; // entity_type + id JSONObject data; // 페이로드 본문 JSONObject metadata; // 컨텍스트 (ts, site_id, shift, tag_id ...) } ``` JSON 직렬화 형식과 메시지 타입 enum 전체 목록은 사용자 가이드의 [메시지 구조](/plantpulse-platform/user/flow.md#message)를 참고하세요. *** ## 패키지 구조 ``` plantpulse.core.engine.flow/ ├── FlowEngineService.java 진입점·디스패처 + load 락 + 캐시 + redeployAll() ├── FlowEngineBootstrap.java 노드 타입 등록 + eagerLoadEnabledFlows + shutdown ├── FlowEventBridge.java 도메인 이벤트 → FlowMsg 변환 ├── trigger/ │ └── FlowTriggerDefinition.java 메시지타입↔트리거노드 매핑 single source of truth ├── runtime/ │ ├── FlowExecutor.java 단일 플로우 실행 (BFS) │ ├── FlowNodeExecutor.java 단일 노드 실행 + 라우팅 (대소문자 강건 매칭) │ ├── FlowCache.java LoadedFlow 캐시 │ ├── LoadedFlow.java 정적 그래프 + 노드 인스턴스 │ ├── ExecutionState.java max_depth/max_revisit 추적 │ └── BuildOnlyContext.java init() 시 NodeContext stub ├── message/ │ ├── FlowMsg.java │ ├── FlowMsgType.java 타입 enum │ └── FlowOriginator.java ├── node/ │ ├── AbstractNode.java │ ├── NodeBehavior.java 노드 인터페이스 │ ├── NodeContext.java tellSuccess/tellFailure/tellNext/markSkipped/scheduleLater │ ├── NodeRegistry.java 타입 ID → Supplier │ ├── Relations.java SUCCESS/FAILURE/TRUE/FALSE/MATCH/NO_MATCH/DEFAULT/THROTTLED 상수 │ └── impl/ │ ├── LogNode / ScriptFilterNode / ScriptTransformNode / ScheduleNode │ ├── TriggerNode (단일 클래스 + FlowTriggerDefinition 데이터로 20 타입 등록) │ ├── MqttSubscribeTriggerNode / JdbcPollTriggerNode (자체 클라이언트 보유) │ ├── filter/FilterNodes 5 Filter (+ ScriptFilter) │ ├── transform/TransformNodes 5 Transform (+ ScriptTransform) │ ├── control/FlowControlNodes Delay/Throttle/Debounce/Noop/Merge/Subflow/Retry │ ├── external/ExternalNodes 8 External (Http/Email/Kafka/Mqtt/Webhook/Jdbc/Sms/Push) │ ├── edge/EdgeNodes 11 Edge (OPC×5 + Tag×5 + Monitoring) — AbstractEdgeRestNode 베이스 │ └── action/ 도메인 액션 패키지 │ ├── AbstractCrudNode │ ├── AssetCrudNodes (3) / SiteCrudNodes (3) │ ├── TagCrudNodes (5: Create/Update/Delete + UpdateTagAlarmBandNumeric/Boolean) │ ├── OrderCrudNodes (8: Create/UpdateStatus/Delete + Start/End/Pause/Resume/Abort) │ ├── AlarmConfigCrudNodes (3) │ ├── ProductionCrudNodes (12: Customer/Product/Employee/Calendar × 3) │ ├── IntegrationNodes (SaveTagPoint / DdsPublish / PublishAssetEvent/Context/Aggregation/Command) │ └── StorageActionNodes (SaveAttributes) ├── script/ │ ├── ScriptEngine.java │ ├── ScriptEngineFactory.java language=JS|EQL 디스패치 │ ├── JsScriptEngine.java JavaScript 샌드박스 — toJavaValue() 정확 변환 │ └── SpelScriptEngine.java EQL 표현식 엔진 ├── scheduler/ │ └── FlowScheduler.java ScheduleNode가 사용하는 cron 등록기 (NODE_LOCKS 직렬화) ├── stats/ │ └── NodeStats.java 노드별 in/out/error 카운터 + duration + 슬라이딩 윈도우 ├── util/ │ └── MsgPath.java "data.x.y" 경로 read/write └── debug/ └── DebugEventPublisher.java tm_flow_log 적재 (drainTo 200건 batch) ``` *** ## FlowNode 인터페이스 ```java public interface FlowNode { /** 노드 초기화 (플로우 로드 시 1회). */ void init(FlowNodeContext ctx, JSONObject configuration) throws Exception; /** 메시지 처리. 다음 노드로 보낼 relation을 ctx.tellNext()로 호출. */ void onMsg(FlowNodeContext ctx, FlowMsg msg) throws Exception; /** 노드 정리 (플로우 unload 시). */ void destroy(); } ``` `onMsg()`에서 던진 예외는 자동으로 `FAILURE` relation으로 라우팅되며, `FAILURE` 출력에 연결된 노드가 없으면 메시지는 drop됩니다. `NodeContext` 핵심 메서드: | 메서드 | 용도 | | ----------------------------------- | ---------------------------------------------------------- | | `tellSuccess(msg)` | `SUCCESS` relation으로 메시지 전달 | | `tellFailure(msg)` | `FAILURE` relation으로 메시지 전달 | | `tellNext(relation, msg)` | 임의 relation(대문자 표준)으로 분기 | | `markSkipped()` | 메시지를 카운터에 포함하지 않고 drop. 트리거 노드의 `*_pattern` 미매칭 메시지 처리에 사용 | | `scheduleLater(delay_ms, runnable)` | 비동기 풀에 지연 실행 등록. `flow_delay`/`flow_retry` 가 사용 | *** ## 메시지 디스패치 ``` FlowEngineService.dispatch(FlowMsg msg) ↓ msg.type → 트리거 노드 타입 매핑 (정적 표) ↓ FlowCache.getAllEnabled() — enabled flow 중 시작 노드 타입이 일치하는 것 선택 ↓ 각 매칭 flow에 대해 FlowExecutor.execute(loaded, msg) ↓ 플로우의 시작 노드부터 BFS 탐색 — FlowNodeExecutor.execute(node, msg) ↓ 각 노드 onMsg()에서 ctx.tellNext("SUCCESS", msg) 호출 ↓ 링크의 relation 매칭으로 다음 노드(들)에 메시지 전달 ↓ in-flight = 0이 되면 종료 (비동기 노드 포함) ``` ### FlowTriggerDefinition 메시지 타입 ↔ 트리거 노드 타입 매핑이 단일 enum(`plantpulse.core.engine.flow.trigger.FlowTriggerDefinition`)에 모여 있습니다. 이전에는 `FlowEngineService.TYPE_TO_TRIGGER` / `TriggerNode.TYPES` / `FlowMsgType` 세 곳에 분산되어 신규 도메인 이벤트 추가 시 누락이 잦았으나, 현재는 enum 한 곳만 갱신하면 dispatch 라우팅·NodeRegistry 등록·메시지 타입 매핑이 모두 derive 됩니다. | API | 용도 | | ---------------------------- | ------------------------------------------------------------------ | | `messageTypeToTriggerType()` | dispatch 핫패스에서 메시지 타입 → 트리거 노드 타입 매핑 | | `allTriggerTypes()` | 등록된 전체 트리거 노드 타입(중복 제거) | | `genericTriggerTypes()` | `TriggerNode` 단일 클래스로 등록할 타입(스케줄/MQTT/JDBC 같이 자체 클래스가 등록하는 트리거 제외) | ### FlowEventBridge 도메인 이벤트(AssetEvent / AssetCommand / Alarm / DDS / OEE / RAM / EMS 등)를 FlowMsg로 변환해 디스패처에 주입합니다. 트리거 노드의 `*_pattern` 옵션(글롭)으로 메시지 단위 필터링이 가능하며, 미매칭 메시지는 `ctx.markSkipped()` 처리되어 후속 노드로 흐르지 않고 처리/에러 카운터에도 포함되지 않습니다. ### 비동기/스레드 모델 * 공통 `AsyncWorkerPool`을 활용하여 메시지 단위로 task 제출 * 노드 간 메시지 전달은 in-memory queue * 플로우 단위 동시성 제한 옵션 — `mm_flow.configuration_json` 의 `max_concurrent` * `flow_delay` / `flow_retry` 의 백오프 대기는 `NodeContext.scheduleLater()` 로 비동기 실행 풀에 위임 ### 캐시·락 * `FlowEngineService.dispatch`: enabled 플로우 ID 5초 캐시 (`selectFlows()` 핫패스 회피) * `FlowCache`: LoadedFlow 단위 캐시 — graph 저장 시 invalidate * `FlowEngineService.redeployAll()`: 전 캐시 invalidate + 활성 플로우 재로드 (운영자 도구) * 동시 dispatch race를 막기 위한 LOAD\_LOCKS / NODE\_LOCKS 적용 * LOAD\_LOCKS: dispatch 시 lazy-load 경로 보호 + `reload()` 메서드 신설 * NODE\_LOCKS: `FlowScheduler.cancelInternal()` + `register()` 원자성 보장 — 잡 등록 충돌 / 영속화 실패 예외 0건 * ScheduleNode 의 미발화(misfire) 정책: `DoNothing` → **`FireAndProceed`** (지연 메시지 drop 방지) *** ## 데이터 모델 (관계형 메타 저장소) ```sql -- 플로우 (워크플로우 단위) CREATE TABLE mm_flow ( flow_id VARCHAR(64) PRIMARY KEY, name VARCHAR(200) NOT NULL, description TEXT, enabled BOOLEAN NOT NULL DEFAULT FALSE, is_root BOOLEAN NOT NULL DEFAULT FALSE, -- Root Flow (메시지 1차 진입점) configuration_json TEXT, -- 전역 설정 (max_concurrent 등) debug_mode BOOLEAN DEFAULT FALSE, insert_user VARCHAR(64), insert_date TIMESTAMP DEFAULT NOW(), update_user VARCHAR(64), update_date TIMESTAMP DEFAULT NOW() ); -- 노드 CREATE TABLE mm_flow_node ( flow_node_id VARCHAR(64) PRIMARY KEY, flow_id VARCHAR(64) NOT NULL REFERENCES mm_flow(flow_id) ON DELETE CASCADE, type VARCHAR(64) NOT NULL, -- 'flow_save_tag_point' 등 name VARCHAR(200), configuration_json TEXT, -- 노드별 설정 JSON position_x INT, position_y INT, debug_mode BOOLEAN DEFAULT FALSE, insert_date TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_mm_flow_node_flow ON mm_flow_node (flow_id); -- 노드 간 링크 (와이어 + relation 레이블) CREATE TABLE mm_flow_link ( from_node_id VARCHAR(64) NOT NULL REFERENCES mm_flow_node(flow_node_id) ON DELETE CASCADE, to_node_id VARCHAR(64) NOT NULL REFERENCES mm_flow_node(flow_node_id) ON DELETE CASCADE, relation VARCHAR(64) NOT NULL, PRIMARY KEY (from_node_id, to_node_id, relation) ); -- 플로우 버전 스냅샷 (M4) CREATE TABLE mm_flow_snapshot ( snapshot_id VARCHAR(64) PRIMARY KEY, flow_id VARCHAR(64) NOT NULL REFERENCES mm_flow(flow_id) ON DELETE CASCADE, version INT NOT NULL, snapshot_json TEXT NOT NULL, -- 플로우 + 노드 + 링크 일괄 직렬화 note TEXT, insert_user VARCHAR(64), insert_date TIMESTAMP DEFAULT NOW() ); -- 플로우 ID 시퀀스 CREATE TABLE mm_flow_seq (...); -- getNewFlowId() — FLOW_NNNNN 형식 ``` > `mm_flow_entry` 테이블은 archive/V11\_\_drop\_mm\_flow\_entry.sql 에서 드롭되었습니다. 트리거 노드(`flow_on_*`)로 진입점이 단일화되어 더 이상 사용되지 않습니다. `replaceFlowGraph` 는 변경 전 그래프를 in-memory 백업하여 중간 실패 시 보상 롤백을 수행합니다 (트랜잭션 차선책). ### DB 마이그레이션 기존 `V1~V11` 파일들은 `src/resources/db/migration/archive/` 로 이관되었고, 현재는 baseline 정합성을 위해 다음 두 파일만 active 입니다. | 파일 | 내용 | | --------------------------------- | --------------------------------------------------------------------------------------------- | | `V1__current_schema_baseline.sql` | 현재 스키마 전체의 idempotent baseline (`CREATE TABLE IF NOT EXISTS` / `ALTER ... ADD IF NOT EXISTS`) | | `V2__relations_uppercase.sql` | `mm_flow_link.relation` 일괄 대문자 변환 + `NoMatch`/`NOMATCH` → `NO_MATCH` 표준화 | `mm_order` 의 `master_id` 는 NOT NULL로 강제되었으며, 마이그레이션은 기존 NULL 행에 MES 마스터 ID(`WO-MES-MES-{001,002,003}`) 를 자동 백필한 뒤 NOT NULL 제약을 적용합니다. *** ## 데이터 모델 (시계열 로그 저장소) 실행 이벤트는 시계열 로그 테이블 `tm_flow_log` 에 적재됩니다. 별도 핸들러 없이 시스템 공용 핸들러(`Create/Insert/SelectSystemHandler`)에 통합되어 있습니다. `/flow/log` 화면 + 라이브 디버그 패널이 동일 테이블을 조회합니다. ```cql CREATE TABLE tm_flow_log ( bucket int, -- 파티션: yyyymmddhh ts timestamp, log_id timeuuid, flow_id text, flow_node_id text, msg_id text, event_type text, -- FLOW_START / FLOW_END / NODE_IN / NODE_OUT / NODE_ERROR relation text, -- SUCCESS / FAILURE / TRUE / FALSE / 커스텀 ... (모두 대문자) node_type text, -- 예: flow_script_transform node_name text, -- 노드 표시명 (디버그 가시성) level text, -- INFO / WARN / ERROR message text, -- "type=X | originator=Y | relation=Z | took=N ms | data=..." error_message text, duration_ns bigint, -- onMsg 실행 시간 (NODE_OUT/ERROR만) PRIMARY KEY ((bucket), ts, log_id) ) WITH CLUSTERING ORDER BY (ts DESC, log_id DESC) AND default_time_to_live = 604800; -- 7일 -- 보조 인덱스: flow_id, msg_id, level -- ALTER TABLE ... ADD IF NOT EXISTS 로 idempotent 컬럼 보강 (race 차단) ``` `DebugEventPublisher` 는 Throwable 처리 강화 + `drainTo` 200건 batch insert 로 적재합니다. *** ## 노드 Configuration 스키마 각 노드는 `mm_flow_node.configuration_json` 에 자기 설정을 직렬화합니다. UI는 노드 클래스가 노출하는 스키마를 기반으로 폼을 자동 렌더링합니다. ### `flow_script_filter` ```json { "language": "JS", "script": "return msg.data.temp > 80;", "timeout_ms": 500 } ``` * `SUCCESS` (true) / `FAILURE` (false 또는 예외) ### `flow_script_transform` ```json { "language": "JS", "script": "msg.data.temp_f = msg.data.temp * 1.8 + 32; return msg;", "timeout_ms": 500 } ``` ### `flow_save_tag_point` ```json { "tag_id_field": "metadata.tag_id", "value_field": "data.value", "timestamp_field": "metadata.ts" } ``` ### `flow_create_work_order` ```json { "asset_id_field": "originator.id", "title_field": "data.title", "description_field": "data.description", "due_date_field": "data.due_date", "default_priority": "NORMAL" } ``` ### `flow_publish_asset_command` ```json { "asset_id_field": "originator.id", "tag_id": "vtag.cmd.motor-001", "cmd_key_field": "data.cmd_key", "payload_field": "data.payload" } ``` ### `flow_http_request` ```json { "method": "POST", "url": "https://mes.example.com/api/po/{id}", "headers": { "Authorization": "Bearer ${MES_TOKEN}" }, "body_template": "{\"po\":\"${msg.data.po}\"}", "timeout_ms": 5000, "response_to_field":"data.response" } ``` * URL/body에 `${MES_TOKEN}`(환경변수) / `${msg.data.po}`(메시지 필드) 치환 지원 ### `flow_kafka_publish` ```json { "topic": "external-events", "key_field": "originator.id", "value_template": "${msg.data}", "headers": { "source": "plantpulse" } } ``` ### `flow_subflow` ```json { "target_flow_id": "FLOW_00042", "wait_for_output": true } ``` > 현재 `wait_for_output` 옵션은 파싱만 하고 동작은 fire-and-forget(즉시 `SUCCESS` 라우팅)입니다. 향후 동기 호출이 필요하면 별도 구현이 추가됩니다. ### `flow_delay` ```json { "delay_ms": 5000, "max_pending_messages": 1000 } ``` ### `flow_retry` ```json { "max_attempts": 3, "backoff_ms": 1000, "backoff_multiplier": "2.0" } ``` * 출력 relation: `SUCCESS` (백오프 후 다음 시도 — 보통 원래 실패 노드로 루프 결선) / `EXHAUSTED` (max\_attempts 도달) * `metadata.retry_count` (현재 시도 회수, 0부터 시작) / `metadata.retry_exhausted` (true) 자동 갱신 ### Edge 노드 공통 (`flow_edge_*`) ```json { "url": "http://edge.local:60000/opc/server/${data.opc_id}", "method": "PUT", "headers": "{\"Authorization\":\"Bearer ${TOKEN}\"}", "body_template": "${msg.data}", "timeout_ms": 5000 } ``` * 노드별 의미상 기본 메서드(`defaultMethod()`): create=POST, update=PUT, delete=DELETE, start/stop/write=POST, read/monitoring=GET — `method` 미설정 시 사용 * 응답: `data.response_status` (HTTP 코드) / `data.response` (본문) / `data.error` (실패 시 메시지) * 분기: 2xx\~3xx → `SUCCESS`, 그 외/예외 → `FAILURE` ### `flow_update_tag_alarm_band_numeric` ```json { "tag_id_field": "data.tag_id", "hi_field": "data.hi", "lo_field": "data.lo", "hi_hi_field": "data.hi_hi", "lo_lo_field": "data.lo_lo", "trip_hi_field": "data.trip_hi", "trip_lo_field": "data.trip_lo", "band_message_field": "data.band_message", "use_alarm_field": "data.use_alarm" } ``` * 입력한 필드만 갱신, 미지정/빈 값은 기존값 유지 (DAO `COALESCE(?, column)` 패턴) * 불린형은 `flow_update_tag_alarm_band_boolean` (bool\_true/bool\_false/우선순위/메시지) ### `flow_start_work_order` / 상태전이 노드들 ```json { "order_id_field": "data.order_id" } ``` * `flow_abort_work_order` 만 추가 옵션: `abort_code_field`(기본 `data.abort_code`), `abort_notes_field`(기본 `data.abort_notes`) * 동작: 도메인 Service 의 `startOrder/endOrder/pauseOrder/resumeOrder/abortOrder` 호출 → 상태머신 검증 + DB 갱신 + 후속 도메인 이벤트 * 전이 실패(잘못된 현재 상태) 시 `FAILURE` *** ## 스크립트 엔진 ### JavaScript (기본 권장) * `language: "JS"` (기본) * ECMAScript 표준, 다중 라인·`var`/`let`/`const`·함수·객체 리터럴 모두 지원 * 바인딩: `msg`, `data`(=`msg.data`), `metadata`(=`msg.metadata`) * 샌드박스: 호스트 접근 차단 모드 — Java 클래스 lookup·파일·네트워크·스레드 모두 차단 **성능 최적화** | 항목 | 동작 | | -------------------------- | ------------------------------- | | **엔진 단일 인스턴스 공유** | 스크립트 엔진 인스턴스 1개를 전체 노드가 공유 | | **컴파일 캐시 (LRU 256)** | 스크립트 본문 hash → 컴파일 재사용 | | **ThreadLocal Context 풀링** | 스레드별 재사용으로 생성 비용 제거 (close 미수행) | ### EQL 표현식 * `language: "EQL"` (기존 표기 호환 유지) * 단일식 only — 다중 라인 + 세미콜론 분리는 미지원 * 부수효과 패턴만 사용 가능 * 기존 EQL 사용자 호환용 ``` #msg.getData().put('temp_f', #msg.getData().getInt('temp') * 1.8 + 32) ``` > 다중 동작이 필요하면 JavaScript(`language: "JS"`)를 사용하세요. *** ## 에러 핸들링 정책 ### 노드 단위 * `onMsg()` 예외 → 자동으로 `FAILURE` relation으로 라우팅 * `FAILURE` 출력에 연결된 노드 없으면 메시지 drop + 에러 로그 * `tm_flow_log` 에 `NODE_ERROR` 이벤트 기록 * `markSkipped()` 호출 시: 메시지가 카운터에 포함되지 않고 drop (트리거 패턴 미매칭 케이스) * `FlowNodeExecutor` 의 relation 매칭은 대소문자 무시(toUpperCase) 비교 — 사용자 그래프의 잔존 소문자 값 호환 ### 플로우 단위 * `max_depth` (기본 100) — 누적 노드 방문 수 제한 * `max_revisit` (기본 3) — 같은 노드 재방문 제한 (사이클 무한 루프 방지) * `flow_timeout_ms` (기본 30,000) — 처리 시간 초과 시 강제 종료 ### 외부 IO 노드 * `retry_count` / `retry_delay_ms` 옵션 (노드 내부 즉시 재시도) * 모든 재시도 실패 시 `FAILURE` relation * Circuit breaker 적용 (M4) — 연속 실패 N회 시 일시 차단 * 백오프·EXHAUSTED 분기가 필요한 경우 별도의 `flow_retry` 노드를 외부에 결선해 사용 (`metadata.retry_count` / `metadata.retry_exhausted` 자동 갱신) ### 메시지 백프레셔 * 진입점(Webhook/Kafka)에서 처리 큐 크기 제한 * 큐 임계 초과 시 inbound 일시 정지 또는 drop policy 선택 * 메트릭: `flow.queue.size`, `flow.dropped` ### 진단 엔드포인트 ``` GET /flow/debug-status ``` `worker_alive` / `queue_size` / `persisted` / `failed` / `dropped` / `last_error` 를 반환합니다. *** ## 컨트롤러 / 엔드포인트 `FlowController` (`plantpulse.server.mvc.flow`) — 세션 인증 기반 UI 엔드포인트. | 메서드 | 경로 | 용도 | | ----------- | ------------------------------------ | ------------------------------------------------------- | | GET | `/flow/index` | 목록 페이지 | | GET | `/flow/edit/{flow_id}` | 편집 페이지 | | GET | `/flow/log` | 실행 이력 페이지 | | GET | `/flow/list` | 플로우 일람 (JSON, 페이지네이션) | | GET | `/flow/get/{flow_id}` | 단일 플로우 상세 | | POST | `/flow/create` / `/flow/update` | 플로우 메타(이름/설명/enabled 등) 생성·수정 | | POST/DELETE | `/flow/delete/{flow_id}` | 플로우 삭제 | | POST | `/flow/{flow_id}/graph` | 노드 + 링크 + 노드 설정 일괄 저장 (자동 스냅샷 적재) | | GET | `/flow/{flow_id}/snapshots` | 자동 스냅샷 목록 | | GET | `/flow/{flow_id}/export` | 그래프 JSON 내보내기 | | POST | `/flow/import` | 그래프 JSON 가져오기 | | POST | `/flow/webhook/{flow_id}` | 외부 웹훅 진입점 (`flow_on_webhook` 트리거) | | POST | `/flow/dispatch` | 임의 메시지 직접 디스패치 (관리자 도구) | | POST | `/flow/{flow_id}/run` | 단건 테스트 실행 | | GET | `/flow/log/list` | 실행 이력 조회 (시계열 로그 `tm_flow_log`) | | GET | `/flow/{flow_id}/stats` | 노드별 stats (노드 ID 5초 캐시 + graph save 시 invalidate) | | POST | `/flow/{flow_id}/stats/reset-errors` | 에러 카운터만 초기화 | | POST | `/flow/{flow_id}/stats/reset-all` | 처리/에러/duration + 플로우 단위 카운터 모두 초기화 | | GET | `/flow/catalog` | NodeRegistry 등록 노드 타입 목록 (UI 카탈로그 ↔ 백엔드 일관성 검증용) | | GET | `/flow/debug-status` | 디버그 워커 + Flow Executor 풀 상태 | | POST | `/flow/redeploy` | 활성 플로우 일괄 재배치 (캐시 invalidate + 재로드, 응답 `data.reloaded`) | | POST | `/flow/selftest` | 자가진단 (임시 샘플 플로우 발화) | > 외부 프로그램·토큰 인증용 엔드포인트는 별도 `V4_FlowAPI` 컨트롤러를 사용합니다. 보안 규칙(`security-rule.xml`)은 `/flow` 와 `/flow/*` 패턴이 분리되어 있어 위 모든 하위 경로가 정상 인증됩니다. *** ## UI 구조 ### 라이브러리 * 벡터 그래프 캔버스 라이브러리(상용 확장팩 미사용) — 노드/와이어/포트/미니맵 렌더링 * 페이지 공용 프론트 유틸 라이브러리(`window.$` / `_` / 모델-뷰 라이브러리 호환) 위에서 동작 * 코드 에디터 컴포넌트 — 스크립트 노드 인스펙터에 임베드 + 큰 화면 다이얼로그 확장 ### JS 모듈 분할 `flow-editor.js` 1,386 → 236 LOC 로 축소되며 8개 모듈로 분리되었습니다. | 파일 | 역할 | | -------------------------- | -------------------------- | | `flow-editor.js` | 메인 lifecycle | | `flow-editor-canvas.js` | 캔버스 셰이프·종이(paper)·노드 추가/렌더 | | `flow-editor-palette.js` | 카탈로그·검색 | | `flow-editor-inspector.js` | 노드 설정 폼 | | `flow-editor-toolbar.js` | 줌·단축키·paper 사이즈 | | `flow-editor-stats.js` | 노드 카운터 폴링 | | `flow-editor-debug.js` | 라이브 디버그 패널 | | `flow-editor-dialogs.js` | 메타 편집·Export | `window.flowEditor.state` 객체로 graph 상태를 모듈 간에 공유합니다. ### 노드 카탈로그 분할 `actions.js` 단일 420 LOC → 4개로 분리되었습니다. * `actions-integration.js` * `actions-asset-tag-site.js` * `actions-order-alarm.js` * `actions-production.js` ### 폴링·이벤트 cleanup * `pagehide` 이벤트로 `stopStatsPolling` / `stopDebugPolling` 호출 * `keydown.flowCanvas` namespace + `.off()` 선행 — 이벤트 누적 방지 *** ## 마일스톤 진행 상황 ### M1 — 엔진 + UI 골격 (완료) * 데이터 모델 (`mm_flow` / `mm_flow_node` / `mm_flow_link` / `mm_flow_seq` / `tm_flow_log`) * `FlowEngineService` (디스패처 + load 락) / `FlowExecutor` / `FlowNodeExecutor` * `NodeRegistry` 타입 ID 등록, `Relations` 상수 * 노드 5종 + 가시화: Input/Output/Log/ScriptFilter/ScriptTransform * 벡터 그래프 캔버스 + 플로우 CRUD UI * 인스펙터(노드 설정 폼), 자가진단, 노드 팔레트 카테고리별 색상 ### M2 — 도메인 Action 노드 (완료) * Asset/Tag CRUD 6종, Site CRUD 3종 * Order CRUD 3종 (생성·상태갱신·삭제) * Production CRUD 12종 (Customer/Product/Employee/Calendar) * AlarmConfig CRUD 3종 (EQL 통합) * Storage 액션: SaveTagPoint / SaveAttributes / SaveAssetTelemetry / 관계형 DB 적재 / 시계열 적재 / GenerateReport * Integration: SendAssetCommand / DdsPublish ### M3 — 진입점 다양화 + Filter/Transform/External (완료) * 트리거 노드 21종 (`flow_on_*`) — 도메인 이벤트 자동 dispatch * ScheduleNode (`flow_schedule`) — 내장 잡 스케줄러 cron * Filter 6종, Transform 6종 모두 구현 * External 8종 (HTTP/Kafka/MQTT/Email/SMS/Push/Webhook/JDBC) * Flow 제어 노드 9종 (Delay/Throttle/Debounce/Noop/Merge/Subflow + Input/Output/Log) * JavaScript 스크립트 엔진 채택, EQL 표현식과 병행 (`language` 키워드) ### M4 — 디버그·운영 (완료) * `tm_flow_log` (시계열 저장소) 적재 + `/flow/log` 타임라인 화면 * 플로우 enable/disable (배포/해제), eager-load * 노드 단위 stats (`NodeStats`) — 처리/에러 카운트 + duration(평균/최근) * 라이브 디버그 패널 (인스펙터 하단) — 2초 폴링 * 플로우 import/export (JSON) — 텍스트 가져오기 + Export 다운로드 * 플로우 버전 스냅샷 (`mm_flow_snapshot`) — graph 저장 시 자동 적재 * 진단 endpoint `/flow/debug-status` * 라이브 메시지 추적 WebSocket push — 미구현 (현재는 폴링형 로그) ### M5 — MES 통합·운영 검증 (완료) * Tag Point Root Flow 통합 — `FlowEventBridge` 가 `flow_on_tag_point` 등 노출 * AssetEvent / AssetCommand 진입점 통합 * 시나리오 130개 등록 + 일괄 검증 (`FLOW_00244 ~ FLOW_00374`) * R1: 트리거×처리×액션 기본 패턴 30종 * R2: Filter/Transform/Control/Multi-Trigger Merge/Subflow 50종 * R3: OEE/품질/생산성, 외부 시스템 연동, 도메인 라이프사이클, 데이터 정제, 진단/모니터링 50종 * 자체 클라이언트 트리거 운영 검증 — `flow_on_mqtt_subscribe` (외부 MQTT 구독), `flow_jdbc_poll` (외부 DB 폴링) ### M6 — 릴리즈 안정성 (완료, V2026.03\_RELEASE) **백엔드** * `FlowEngineService.dispatch` enabled 플로우 ID 5초 캐시 * `JsScriptEngine` Source LRU 256 + ThreadLocal Context 풀링 * `DebugEventPublisher` Throwable 처리 강화 + `drainTo` 200건 batch insert * `/flow/{id}/stats` 노드 ID 5초 캐시 + graph save 시 invalidate * `MqttSubscribe` / `JdbcPoll` / `Schedule.destroy()` try-finally + Throwable * `OrderCrudNodes.CreateOrder` FK 빈 문자열 → null 변환 * `TagCrudNodes.CreateTag` `opc_id_field` 옵션 + 명확 fail 메시지 * `ProductionCrudNodes.CreateCustomer/Employee` NOT NULL default 보강 * `ExternalNodes` Kafka/Mqtt/Webhook/Email/Sms/Push 모두 `*_field` 동적 옵션 (`pick()` 헬퍼) * Update\* 노드 빈 문자열 무시 **시계열 저장소 통합** * 별도 핸들러 삭제 → 시스템 공용 핸들러로 통합 * `tm_flow_log` 컬럼 `node_name text` / `duration_ns bigint` 추가 * `ALTER TABLE ... ADD IF NOT EXISTS` 로 idempotent 스키마 보강 **FlowService** * `replaceFlowGraph` 변경 전 그래프 in-memory 백업 → 중간 실패 시 보상 롤백 **UI** * 8개 모듈 분할 (위 [UI 구조](#ui) 참조) * 폴링 cleanup (`pagehide` + `stopStatsPolling/stopDebugPolling`) * 다이얼로그 컴포넌트 통일 (생성/Import/메타 편집) * 노드 뱃지 명시적 `rect+text` (paint-order 트릭 제거 — 모든 SVG 렌더러 호환) * 노드 우하단 duration 텍스트 (`평균 X · 최근 Y`) * 캔버스 가로 스크롤바 항상 표시 **v1 출시 후 추가 결함 수정 (2026-05-03\~05)** * Flow ID UUID → 시퀀스 (`FLOW_00001`) * "활성/비활성" → "**배포/해제**" 명칭 변경 * TriggerAlarm `alarm_config_id` NOT NULL 누락 fix * 동시 dispatch race fix (FlowEngineService load 락) * `AlarmMutationDAO.saveAlarm` 에 `tag_id` / `site_id` 컬럼 추가 * `/flow/log` PPTimeline·PPDateRange 통합 * JavaScript `msg.data` setter wrapJson 픽스 (JSONObject 래퍼 위임) * Statement view 에러 카운트 초기화 endpoint (`/statement/reset-error/{asset_id}/{statement_name}`) ### M7 — 도메인 자동화 확장·운영 안정화 (완료, V2026.05\_RELEASE) **노드 카탈로그 확장** * 트리거 5종 추가 — `flow_on_asset_health_status`, `flow_on_asset_connection_status`, `flow_on_oee_event`, `flow_on_ram_event`, `flow_on_ems_event` * 흐름 제어 신규 — `flow_retry` (max\_attempts/backoff\_ms/backoff\_multiplier, EXHAUSTED 분기, `metadata.retry_count` 추적) * 태그 알람밴드 부분 갱신 2종 — `flow_update_tag_alarm_band_numeric` / `flow_update_tag_alarm_band_boolean` * 작업지시 상태 전이 5종 — `flow_start_work_order` / `flow_end_work_order` / `flow_pause_work_order` / `flow_resume_work_order` / `flow_abort_work_order` * 엣지 디바이스 카테고리 11종 신설 — `flow_edge_opc_*` (5) + `flow_edge_tag_*` (5) + `flow_edge_monitoring` (1), `AbstractEdgeRestNode` 베이스로 HTTP 호출 일원화 **Update 노드 fetch+merge 패턴** * Customer/Product/Employee/Calendar Update 노드가 기존 레코드 조회 → 입력 필드만 병합 → 저장 * DAO 측은 `COALESCE(?, column)` SQL로 부분 갱신 안전 보장 * 빈 문자열·null 무시 → 기존 값 유지 **Relations 대문자 표준화** * `Relations.java` 상수 모두 대문자 (`SUCCESS`, `FAILURE`, `TRUE`, `FALSE`, `MATCH`, `NO_MATCH`, `DEFAULT`, `THROTTLED`) * 신규 `EXHAUSTED` (Retry 노드) * DB 마이그레이션 `V2__relations_uppercase.sql` 로 기존 그래프 자동 변환 * `FlowNodeExecutor` 라우팅이 대소문자 무시 매칭으로 잔존 소문자 호환 **SKIPPED 처리** * `NodeContext.markSkipped()` 추가 * 트리거 노드의 `*_pattern` 미매칭 메시지가 후속 노드로 흐르지 않고 카운터에서도 제외 **FlowTriggerDefinition 단일 정의** * 메시지 타입 ↔ 트리거 노드 타입 매핑이 `plantpulse.core.engine.flow.trigger.FlowTriggerDefinition` enum 한 곳으로 통합 * `FlowEngineService.TYPE_TO_TRIGGER` / `TriggerNode.TYPES` / `FlowMsgType` 의 중복 매핑 제거 **컨트롤러 / 운영 엔드포인트** * `POST /flow/redeploy` — 모든 활성 플로우 캐시 invalidate + 재로드. 자체 스케줄러 보유 트리거(Schedule/JdbcPoll/MqttSubscribe) 재등록용 * `POST /flow/{id}/stats/reset-all` — 처리/에러/duration + 플로우 단위 카운터(`started`/`finished`/평균/최근) 모두 초기화 * `GET /flow/catalog` — NodeRegistry 등록 타입 목록(프론트 카탈로그 일관성 검증) * 보안 규칙: `/flow*` → `/flow` + `/flow/*` 분리 (하위 경로 인증 정상화) **Race condition / Cron 안정성** * LOAD\_LOCKS: dispatch lazy-load 보호 + `FlowService.reload()` 메서드 신설 * NODE\_LOCKS: `FlowScheduler.cancelInternal()` + `register()` 원자성 — 잡 등록 충돌/영속화 실패 예외 0건 * 미발화 정책: `DoNothing` → `FireAndProceed` (지연 메시지 drop 방지) * `WorkerTask` idle 단락 (`hasActiveEntries()` 체크) — zero overhead **JsScriptEngine 정확 변환** * 사용자 스크립트의 `msg.data = {...}` 통째 할당 시 `[object Object]` 문자열로 직렬화되던 버그 수정 * `toJavaValue(Value)` 도입 — JavaScript Value → JSONObject/JSONArray/원시값 정확 변환 **부하 워치독 (LoadHealthCheckTimer)** * 위치: `plantpulse.core.diagnostic.LoadHealthCheckTimer` (`MonitoringTimer` 구현) * 주기 10초(`load.health.interval.ms`) / 시작 지연 60초 / 상태 전이 히스테리시스 30초(`load.health.hysteresis.ms`) * 상태머신: `HEALTHY` / `DEGRADED` / `CRITICAL` (Flow와 DDS 영역 각각 추적) * 신호: * Flow — pending 큐 크기, 노드 실패율 delta, finished 처리량 * DDS — 큐 크기, 큐 FULL delta, latency\_warn delta * 임계값 외부화 (`engine.properties`): `load.flow.pending.warn`/`crit`, `load.flow.fail_rate.warn`/`crit`, `load.dds.queue.warn`/`crit`, `load.dds.drop_delta.warn`/`crit`, `load.dds.latency_delta.warn`/`crit` * 상태 전이 시점에만 `DiagnosticLogger` 1줄 발행 (핫패스 0 오버헤드) **DB 마이그레이션 baseline 재정의** * 기존 `V1~V11` → `archive/` 이관, `mm_flow_entry` 드롭 * 신규 `V1__current_schema_baseline.sql` 전체 idempotent baseline + `V2__relations_uppercase.sql` * `mm_order.master_id` NOT NULL 강제 (MES 마스터 ID 자동 백필 후 적용) **MQTT 클라이언트 업그레이드** * 외부 MQTT 구독 클라이언트 라이브러리 마이너 업그레이드 — `MqttSubscribeTriggerNode` 재연결 안정성 개선 *** ## 리팩토링 계획 ### P0 — 즉시(이번 분기) 1. **노드 자동 등록** — 현재 Bootstrap 200+ LOC 의 수동 register 호출을 `@FlowNodeType("flow_xxx")` 어노테이션 + ServiceLoader 또는 컴포넌트 스캔으로 대체. 신규 노드 추가 시 register 단계 제거. 2. **거대 파일 분리** * `ProductionCrudNodes.java (475 LOC)` → `production/{Customer,Product,Employee,Calendar}CrudNodes.java` * `ExternalNodes.java (390 LOC)` → `external/{Http, Kafka, Mqtt, Email, Sms, Push, Webhook, Jdbc}Node.java` 3. **노드 설정 스키마 메타데이터화** — 백엔드의 `optString` 호출과 프론트 `flow-nodes.js` 의 schema가 이중 관리되어 동기화 누락이 빈번. 백엔드 노드 클래스에 `static SCHEMA = JSON` 노출 → `/flow/api/node-types` 엔드포인트가 자동 직렬화 → 프론트는 카탈로그를 API로 페치. (`flow-nodes.js` 1,000+ LOC 삭제 가능) ### P1 — 다음 분기 4. **ScriptEngine 캐시 통합** — 공통 `ScriptCache(language, hash)` 키 LRU + 만료. `NodeStats` 와 통합한 evict 메트릭. 5. **`TriggerNode` 21종 데이터 위주 재구성** — 빈 inner static class 21개를 단일 `TriggerNode` + `messageType` 파라미터로 통합. 6. **FlowExecutor BFS / DFS 통합 + 사이클 검사 추출** — `Traversal` 추상화 + 사이클 검사 Visitor. ### P2 — 장기 7. **노드 인스턴스 풀링 검토** — 현재는 LoadedFlow 빌드 시 1회 생성하여 모든 메시지가 공유 → Stateful 노드(Debounce/Throttle/Merge)에 동시성 보호 필수. 향후 메시지 단위 인스턴스 풀링 검토. 8. **FlowMsg immutable 변환** — `MsgPath.write()` in-place 변경 → 동시 fan-out 시 가시성 문제 잠재. copy-on-write / builder 패턴. 9. **NodeStats → NodeMetrics** — Counter 외에 latency p50/p99, error rate 윈도우 → 메트릭 익스포터 연결. ### Out of Scope * 노드 카탈로그 축소 (현재 100여 종 모두 유지) * 별도 SDK 모듈 분리 * 외부 워크플로우 엔진과의 호환 *** ## 운영 결정 사항 | 항목 | 결정 | | ------------------- | --------------------------------------------- | | UI 캔버스 | 벡터 그래프 라이브러리 (상용 확장팩 미사용) | | 명칭 | "플로우 (Flow)" — Rule Chain 용어 폐기 | | Flow ID 형식 | `FLOW_NNNNN` 시퀀스 | | 운영 액션 명칭 | "배포/해제" (활성/비활성 폐기) | | 권한 모델 | 기존 시스템 인증/권한 그대로 사용 (별도 도입 없음) | | 사용자 함수 언어 | JavaScript + EQL 표현식 병행 (default JS, 샌드박스 실행) | | Tag Point Root Flow | `FlowEventBridge` 로 통합 | *** ## 관련 화면·문서 * [사용자 가이드 — 플로우](/plantpulse-platform/user/flow.md) * [메시징 아키텍처](/plantpulse-platform/developer/messaging.md) * [스케줄러 아키텍처](/plantpulse-platform/developer/scheduler.md) * [데이터베이스 스키마](/plantpulse-platform/developer/database-schema.md) * [API 사용 매뉴얼](/plantpulse-platform/developer/api-manual.md) *** ## 관련 파일 ### 엔진 * `src/plantpulse/core/engine/flow/**` — 100여 종 노드, runtime, scheduler, stats, util, debug, message, script * `src/plantpulse/core/meta/dao/FlowDAO.java` — `mm_flow_*` JDBC DAO + `getNewFlowId()` 시퀀스 ### Controller / Service * `src/plantpulse/server/mvc/flow/FlowController.java` * `src/plantpulse/core/service/FlowService.java` ### UI * `WebContent/WEB-INF/jsp/flow/index.jsp` / `edit.jsp` / `log.jsp` * `WebContent/public/scripts/flow/flow-list.js` / `flow-editor*.js` (8 모듈) / `flow-log.js` / `flow-nodes.js` * `WebContent/public/scripts/flow/nodes/*.js` (카탈로그 4 분할) * `WebContent/public/css/pp-flow.css` * `WebContent/public/js//.{js,css}` — 벡터 그래프 캔버스 라이브러리 자산 ### DDL * `src/resources/db/migration/V1__current_schema_baseline.sql` — 전체 스키마 idempotent baseline * `src/resources/db/migration/V2__relations_uppercase.sql` — Relations 대문자 변환 * `src/resources/db/migration/archive/` — 기존 V1\~V11 보관 (`mm_flow_entry` 드롭 포함) ### 통합 지점 * `engine.properties` — `flow.tag_point_dispatch=true` 등 * `FlowEventBridge` — 도메인 이벤트(AssetEvent/AssetCommand/Alarm/DDS 등) → FlowMsg 변환