edge-processor — 엣지 데이터 처리기

한 줄 정의. 별도 수집기(OPC/PLC 게이트웨이 추정)가 Redis에 적재한 실시간 설비 태그를, 이 프로세서가 꺼내 가공·저장·감시하는 처리 계층이다. 이름에 "edge"가 있지만 MQTT/Kafka/시리얼/TCP 등 디바이스 직접 통신은 없다 — 유입은 전적으로 Redis를 통한 간접 방식.

주요 책임 4가지

1. 데이터 파이프라인 — Redis 태그 수집 → 가상태그 생성 → 알람 평가 → InfluxDB 저장 → Redis 정리
2. 알람 엔진 — 태그 조건식 DSL 평가 → 알람 발생/해제 및 외부 알림 REST 전송
3. 가상 태그 — 실측 태그를 산술식으로 조합한 파생 태그 계산·저장
4. CPK 계산 · 태그 동기화 — 공정능력지수 산출, 외부 DB(MSSQL/MariaDB) 태그 메타 동기화

최초 체크아웃 후 (필수)

# Spotless 커밋 훅 + 커밋 메시지 템플릿 설치
./gradlew setup

빌드 · 실행 · 테스트

./gradlew clean build          # 빌드 (jar 생성)
./gradlew bootRun              # 로컬 실행 → http://localhost:8082
./gradlew test jacocoTestReport # 테스트 + 커버리지(JaCoCo)
./gradlew spotlessApply        # 코드 포맷 자동 적용

패턴. 전통적 레이어드(api / service / domain / repository). 데이터 파이프라인만 Abstract Factory + Strategy로 legacy/json_v1 버전을 application.collect.version으로 전환한다.

실시간 파이프라인 5단계 DataPipelineService.processPipeline()

• 각 단계가 try/catch로 격리 — 한 단계 실패가 전체를 중단시키지 않음(저장 실패만 error, 나머지 warn).
• 수집 결과를 Map<String,Object>로 전달 → 타입 안전성 약함(§10 개선 대상).
• 폴링: json_v1은 fixedDelay=20ms, legacy는 cron 매초 + @Async.

버전별 컴포넌트 매핑

팩토리: JsonV1ComponentsFactory / LegacyComponentsFactory 가 @ConditionalOnProperty로 선택 활성.

4.1 데이터 파이프라인 pipeline/data

DataPipelineService가 @PostConstruct에서 팩토리로 5개 컴포넌트를 주입받아 순차 실행하는 오케스트레이터.

누락 데이터 복구 MissedDataRecoveryService: 10분마다 timeThreshold(2000ms)를 초과한 오래된 Redis 키를 배치(기본 60개)로 수집·저장. legacy 구현은 미지원(UnsupportedOperationException) — json_v1 전용. 복구 경로에선 알람 미실행.

4.2 알람 엔진 alarm, common/util/alarm

• TagAlarmService / MultiAlarmService — 인메모리 룰 관리 + 평가 + 로그 저장 + 이벤트 발행. @Scheduled(1s)로 룰 활성/비활성 상태 전이.
• TagRuleManager / MultiRuleManager — Collections.newSetFromMap(ConcurrentHashMap) 기반 스레드세이프 룰 셋.
• 표현식 파서 — Tokenizer → AlarmExpressionPreprocessor → RecursiveDescentParser → Condition 트리 (§8 문법 참조).
• 알림 발송 — AlarmFiredEvent → @Async("alarmEventExecutor") + @TransactionalEventListener(AFTER_COMMIT) → RestClient POST /api/v1/alerts/notify (타임아웃 미적용·재시도 없음 — 상수 3s는 코드상 주석처리, 심층분석 ② 참조).

4.3 가상 태그 virtualtag

• VirtualTag — tagId, expression({{TAG}} 참조), savingPolicy 애그리거트.
• VirtualTagExpressionEvaluator — Shunting-Yard 알고리즘으로 + - * / 산술식 평가(중위→후위, 0 나눗셈·괄호 불일치 예외 처리).
• SavingPolicy / SavingInterval — 기간·간격("5m","10s","2h") 저장 정책, shouldSaveAt(LocalDateTime).
• VirtualTagService — @Scheduled(fixedRate=1s)로 DB 변화 감지, 병렬 스트림 생성·저장.

4.4 CPK · 동기화 service

4.5 도메인 모델 domain

• Tag — id, value(Double 또는 String 다형성), process, line, influxTags(Map). isNumeric()/getNumericValue() 등 제공, id 기준 equals.
• TagInfoMap — StampedLock 기반 싱글톤 스레드세이프 레지스트리(낙관적 읽기 락).

InfluxDB 버킷: raw test / aggr aggr, measurement default_line. 저장 정밀도 WritePrecision.S(초).

거의 모든 기능이 토글로 제어된다. 아래는 src/main/resources/application.yml 실제 기본값 기준.

모든 스케줄은 단일 TaskScheduler(ThreadPoolTaskScheduler, poolSize 10)를 공유. 알림 이벤트만 별도 alarmEventExecutor(core 5/max 10/queue 100, CallerRunsPolicy).

common/util/alarm의 재귀하강 파서가 처리하는 조건식 문법.

expression ::= term (OR term)*
term       ::= factor (AND factor)*
factor     ::= '(' expression ')' | comparison
comparison ::= {{TAG_ID}} OP value        # OP: == != > >= < <=

• 우선순위: () > AND > OR
• 비교 연산자: == != > >= < <= — 등호/부등호는 숫자·문자 모두, 대소 비교(> < 등)는 숫자만
• 태그 참조: {{TAG_ID}} 형식으로 실측/가상 태그 값을 치환

# 예시: 온도 태그가 80 초과이고 압력이 100 이상일 때
{{TEMP_01}} > 80 AND {{PRESS_01}} >= 100

# 예시: 상태가 정지이거나 알람 플래그가 켜졌을 때
{{STATUS}} == STOP OR {{ALARM_FLG}} == 1

가상 태그 산술식은 별도로 VirtualTagExpressionEvaluator(Shunting-Yard)가 + - * /와 괄호를 평가한다.

읽는 순서

1. HistorianEdgeProcessorApplication.java — 진입점
2. config/ — DataSource·Redis·InfluxDB·Scheduler 설정으로 인프라 그림 파악
3. pipeline/data/service/DataPipelineService.java — 파이프라인 5단계 오케스트레이션(핵심)
4. pipeline/data/factory/ + component/impl/{json_v1,legacy} — 버전별 구현
5. domain/Tag.java, domain/TagInfoMap.java — 도메인 모델
6. alarm/service/* + common/util/alarm/RecursiveDescentParser — 알람 엔진·파서
7. virtualtag/VirtualTagExpressionEvaluator — 가상태그 산술 평가
8. src/main/resources/application.yml — 토글·연동 총람

범위. 운영 기본값 application.collect.version=json_v1(application.yml:48) 경로를 중심으로, legacy는 대조군으로 병기한다. 데이터가 Redis 해시 문자열 → Map<String,Object> → List<Tag> → InfluxDB Point로 변형되는 전 과정을 추적한다.

1) Redis 수집 소스 — 두 개의 해시

직렬화는 전부 StringRedisSerializer(RedisConfig.java:16-18) — Redis raw는 순수 문자열.

신선도(±2초) 판정 · 최신 키 선택 getLatestTimestampKey() (RedisService.java:28-54)

long currentTime = Instant.now().getEpochSecond() * 1000;         // 40
return keys.stream()
    .filter(key -> Math.abs(Long.parseLong(key) - currentTime) <= timeThreshold) // 47
    .max((k1,k2) -> Long.compare(Long.parseLong(k1), Long.parseLong(k2)))        // 52
    .orElse(null);

• timeThreshold = application.redis.time-threshold-ms:2000. 현재 시각 ±2초 이내 키만 통과, 그중 가장 최근(max) 1건만 소비.
• 누락 분류 getMissedTimestampKey()(:56-83): (currentTime - timestamp) > 2000 — 2초 넘게 과거인 키 전부를 오래된 순 List로. → 복구 대상.

2) Collector 파싱 — 데이터 변형

JsonV1Collector (JsonV1Collector.java:135-193)

입력 JSON 스키마와 매핑:

{ "data":[ { "field":{ "<tagId>": <value> }, "tag":{ "PROC":"P1", "<k>":"<v>" } } ] }
   field 엔트리 → Tag.id = key, Tag.value = 숫자면 Double / 문자면 String
   tag.PROC    → Tag.process          (PROC 키는 influxTags에서 제외)
   tag의 나머지 → Tag.influxTags(Map)

• 타입 판별 parseValueFromJson(:202-215): isNumber→Double, isTextual→String, isNull→스킵, 미지원 타입은 warn 후 스킵.
• 성공/실패 모두 dataKey를 반환 Map에 포함(정리 단계에서 사용). line 필드는 수집 경로에서 세팅되지 않음(확인된 사실).

LegacyCollector (LegacyCollector.java:60-68)

data.split("\^")          // 행 구분 '^'
    → row.split(",")       // 열 구분 ','
    → Tag.builder().id(t[0]).value(Double.parseDouble(t[1])).build();

Tag.equals/hashCode는 id만 기준(Tag.java:75-86) → 동일 id 중복 시 Set/merge 충돌 여지.

3) 파이프라인 5단계 — 격리와 로깅 레벨

DataPipelineService는 @PostConstruct에서 PipelineComponentsFactory로 5개 컴포넌트를 생성해 보관(:44-53). processPipeline()(:59-119) 단계별 격리:

• 알람은 tags + virtualTags(alarmTags), 저장은 실물 tags만 대상. 가상태그 저장은 generate() 내부에서 별도 수행(추정).

4) 스케줄러 · 누락 복구

누락 복구 MissedDataRecoveryService(@Scheduled(fixedDelay=10분), json_v1 전용): getMissedTimestampKey()로 오래된 키를 batchSize 60 단위 Lazy Stream으로 수집 → virtualTag→influx→clean 순 처리. 알람은 미발화(AlarmCaller 미주입, :52). legacy는 UnsupportedOperationException(빈 자체가 생성 안 되어 실경로 도달 불가).

5) 수집 흐름 시퀀스 (json_v1)

알람은 단일 태그와 다중 태그 두 계열이 거의 대칭 구조로 존재한다. 현재 application.alarm.use=false(yml)라 발화 경로 자체는 기본 비활성이나, 활성화 시 아래 생애주기로 동작한다.

1) 인메모리 룰 셋 — tagId가 곧 식별자

Set<TagAlarmRule> tagRules       = Collections.newSetFromMap(new ConcurrentHashMap<>());
Set<TagAlarmRule> firedTagRules  = Collections.newSetFromMap(new ConcurrentHashMap<>());  // TagRuleManager:20-26

• tagRules = 평가 대상 활성 룰 전체. firedTagRules = 이미 발화되어 억제 중인 룰의 clone() 사본.
• equals/hashCode가 tagId만 사용(TagAlarmRule.java:256-266) → tagId당 룰 1개. (Multi는 multiAlarmNo 기준)
• 초기화 @PostConstruct init(): 재시작 잔여 deactivating 정리 → activated 룰 로드 → initialized=true. 이 플래그가 상태전이 스케줄의 조기 실행을 가드.

2) 표현식 DSL — 파싱 파이프라인

AlarmExpressionParser.parse() → Preprocessor → Tokenizer → RecursiveDescentParser → Condition 트리.

expression      ::= term (OR term)*        // OR 최저 우선순위, 좌결합
term            ::= factor (AND factor)*   // AND > OR
factor          ::= singleCondition | '(' expression ')'   // 괄호 최고
singleCondition ::= TAG_ID  OP  (NUMBER | STRING)          // OP: == != > >= < <=

• 전처리(Preprocessor): 단독 =→==, 'ON'→1, 'OFF'→0(작은따옴표만).
• 토큰화: {{...}}→TAG_ID, "..."→STRING, AND/OR은 대소문자 무시+경계 필요, 음수·소수 지원, 단독 =는 예외(전처리 전제).
• {{TAG}}는 값 치환이 아니다 — 토큰에 tagId 문자열만 담고, 평가 시점에 tagValues.get(tagId)로 런타임 lookup(SingleCondition.java:44-49).

토큰:  TAG_ID(TEMP) GT(>) NUM(80) AND TAG_ID(PRESS) GE(>=) NUM(100) EOF
트리:  CompoundCondition(AND)
        ├ SingleCondition(TEMP,  >,  80.0)
        └ SingleCondition(PRESS, >=, 100.0)
평가:  {TEMP:85, PRESS:120} → 85>80(T) AND 120>=100(T) → true, winning=(PRESS>=100)
       {TEMP:85, PRESS:90}  → 첫 조건 T, 둘째 90>=100(F) → 단락평가로 즉시 false

3) 두 종류의 스케줄 — 혼동 금지

(A)는 "룰을 평가 대상에 넣고 빼는" 관리 작업일 뿐, fire/clear가 아니다.

4) fire / clear 전이 (값 평가)

핵심 evaluateAndFilterRulesWithAlarmValue(TagAlarmService.java:263-317) — rules.parallelStream():

• 미발생→발생 addFiredRuleIfNew(:375-383): firedTagRules.contains면 억제(중복 차단), 아니면 clone()을 셋에 추가하고 신규 발화만 하류로 → occurred 로그 저장.
• 발생→해제 resolveFiredTagRule(:385-395): 열린 로그를 resolved로 전환 후 셋에서 제거.
• 디지털 이벤트성 예외: onToOff/offToOn 등은 자동 해제하지 않음(엣지 이벤트).
• 억제/재활성: 매 실행 앞에서 removeInvalidFiredRules(now)가 canReactivate(suppression 경과, 기본 FOREVER) 룰을 firedTagRules에서 제거 → 재발화 허용.

발화 시 createAndSaveAlarmLogs가 occurred 로그 saveAll 후 publishEvent(AlarmFiredEvent)(:539). 저장 실패는 catch만 하고 이벤트 미발행(조용히 유실).

5) 비동기 발화 · 알림 전송

// CommonAlarmCaller.java:37-38  (useAlarm=application.alarm.use, yml=false)
CompletableFuture.runAsync(() -> tagAlarmService.executeTagAlarm(tags, ts));
CompletableFuture.runAsync(() -> multiAlarmService.executeMultiAlarm(tags, ts));

스레드풀: runAsync(Runnable)는 Executor 미지정 → 공용 ForkJoinPool.commonPool(). 내부 parallelStream도 commonPool → 평가 태스크와 스트림이 같은 풀 공유. fire-and-forget이라 예외는 삼켜짐.

알림 AlarmNotificationEventListener:

6) REST API · 생애주기 시퀀스

7) 동시성 주의점 (요약)