본문으로 건너뛰기

plantpulse-cep (복합 이벤트 처리)

역할​

Complex Event Processing 엔진입니다. 여러 센서 이벤트의 시간적 · 논리적 패턴을 실시간으로 감지해 알람 · Flow 트리거 · 자동화를 수행합니다.

항목값
모듈명plantpulse-cep
설치 경로/opt/kopens/plantpulse-platform/plantpulse-cep/
포트7400 (HTTP) · 7401 (TLS)
엔진Esper CEP (내장 Tomcat 위의 웹앱)
pd 서비스cep — MASTER 만
API 인증X-API-Key: <PP_CEP_API_KEY>
콘솔 로그인PP_CEP_WEB_USER / PP_CEP_WEB_PASSWORD — 비어 있으면 콘솔만 닫힙니다

사용 예시​

"온도가 80°C 를 5초 이상 초과하고 동시에 압력이 평균보다 15% 이상 상승하면 위험 알람 발생"

이런 복합 조건을 EQL(Event Query Language) 로 정의하면 CEP 엔진이 실시간으로 매칭해 결과를 발행합니다.

구성​

디렉토리 구조​

plantpulse-cep/
├── config/plantpulse-cep.properties     # 생성물 ← /etc/kopens/conf/plantpulse-cep.properties.template
├── bin/start.sh stop.sh
├── logs/
└── server/                              # 내장 Tomcat — webapps/ROOT.war · logs/catalina.<날짜>.log

주요 설정​

템플릿 /etc/kopens/conf/plantpulse-cep.properties.template 의 핵심 항목입니다. 자리표시자는 렌더 때 채워집니다.

# 캐시 · HA 스냅샷 (Valkey)
cache.host=${PP_REDIS_HOST}
cache.port=${PP_REDIS_PORT}
cache.password=${PP_REDIS_PASSWORD}
cep.ha.enabled = true                 # 재시작 때 엔진 상태(윈도 · 테이블) 보존
cep.ha.snapshot.interval_ms = 1000

# 이벤트 저장 (Cassandra)
storage.host=${PP_CASSANDRA_HOST}
storage.port=${PP_CASSANDRA_PORT:9042}
storage.keyspace=${PP_KEYSPACE}
cep.storage.ttl_seconds=864000        # 10일

# 처리 스레드
cep.consumer.thread_count=16
cep.stream.shard_count=16
cep.async.max_pool_size=256

# 인증
cep.api.key=${PP_CEP_API_KEY}         # 관리 API 의 X-API-Key
cep.web.user=${PP_CEP_WEB_USER}       # 콘솔 로그인
cep.web.password=${PP_CEP_WEB_PASSWORD}

값을 바꾸는 절차는 설정 바꾸는 법 — 템플릿, API 키 회전은 비밀번호 · API 키 바꾸기입니다.

인증​

관리 API 는 X-API-Key 헤더로 인증합니다. 서버 콘솔의 PP_CEP_API_KEY 와 같은 값이어야 하고, 한쪽만 바꾸면 즉시 401 이 납니다 — passwd.sh PP_CEP_API_KEY 가 양쪽을 함께 바꿉니다.

curl -H "X-API-Key: ${PP_CEP_API_KEY}" http://127.0.0.1:7400/api/v1/status

EQL 예제​

서버 콘솔의 AUTOMATION > CEP 메뉴, 또는 CEP 콘솔 https://<서버IP>:7401/ 에서 EQL 을 등록 · 배포합니다.

-- 5초 윈도에서 평균 온도가 80도 초과
@Name('high-temp')
SELECT tagId, avg(value) AS avg_temp
FROM TagPoint(tagId = 'TEMP_REACTOR_01').win:time(5 sec)
GROUP BY tagId
HAVING avg(value) > 80
OUTPUT EVERY 1 sec;

-- 두 태그의 시간 정렬 패턴
@Name('temp-pressure-correlation')
SELECT a.tagId AS temp_id, b.tagId AS pressure_id
FROM pattern [
  every a = TagPoint(tagId = 'TEMP_01' AND value > 80)
  -> b = TagPoint(tagId = 'PRESS_01' AND value > 100) WHERE timer:within(10 sec)
];

문법은 EQL 도움말을 참고하세요.

운영 명령​

docker exec plantpulse-datalake pd status cep
docker exec plantpulse-datalake pd restart cep
docker exec plantpulse-datalake pd logs --lines 100 cep

# 표준 헬스체크 (익명, readiness — 엔진 + 컨슈머 + 복구 완료 · Valkey 연결 시에만 UP)
curl -fsS http://127.0.0.1:7400/api/health
# 200 {"status":"UP","service":"plantpulse-cep-server-web",…,"checks":{"engine":true,"redis":true}}
# 503 {"status":"STARTING"} 기동 중 / 503 {"status":"DEGRADED"} 의존성 다운

자주 발생하는 문제​

증상원인조치
룰이 동작하지 않음배포 안 됨콘솔에서 «배포»
처리 지연윈도 메모리 부족EQL 윈도 축소 또는 힙 상향
Kafka lag 증가처리 스레드 부족cep.consumer.thread_count 상향 (템플릿)
API 401API 키 불일치passwd.sh PP_CEP_API_KEY
콘솔 로그인 거부PP_CEP_WEB_PASSWORD 비어 있음웹 화면 로그인 계정

관련 문서​