plantpulse-cep (Complex Event Processing)
Role
Complex Event Processing engine. Detects temporal and logical patterns across multiple sensor events in real time and executes alarms, Flow triggers, and automation.
| Item | Value |
|---|---|
| Module name | plantpulse-cep |
| Installation path | /opt/kopens/plantpulse-platform/plantpulse-cep/ |
| Port | 7400 (HTTP) · 7401 (TLS) |
| Engine | Esper CEP (web app on embedded Tomcat) |
pd service | cep — MASTER only |
| API authentication | X-API-Key: <PP_CEP_API_KEY> |
| Console login | PP_CEP_WEB_USER / PP_CEP_WEB_PASSWORD — if empty, console only closes |
Use case
"If temperature exceeds 80°C for 5 seconds or more and at the same time pressure rises 15% or more above average, generate a critical alarm"
Define such a compound condition in EQL (Event Query Language), and the CEP engine matches it in real time and publishes the result.
Architecture
Directory structure
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.<date>.log
Key configuration
Core items from template /etc/kopens/conf/plantpulse-cep.properties.template. Placeholders are filled at render time.
# 캐시 · 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}
To change values, see How to change configuration — template. To rotate API keys, see Change passwords and API keys.
Authentication
The management API authenticates via X-API-Key header. It must match the value in the server console's PP_CEP_API_KEY, and changing only one will immediately return 401 — passwd.sh PP_CEP_API_KEY changes both together.
curl -H "X-API-Key: ${PP_CEP_API_KEY}" http://127.0.0.1:7400/api/v1/status
EQL examples
Register and deploy EQL in the server console under AUTOMATION > CEP, or in the CEP console https://<server-ip>:7401/.
-- 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)
];
For syntax, see EQL guide.
Operations commands
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"} 의존성 다운
Common issues
| Symptom | Cause | Action |
|---|---|---|
| Rule not triggering | Not deployed | Click Deploy in console |
| Processing lag | Window memory insufficient | Shrink EQL window or increase heap |
| Kafka lag rising | Processing threads insufficient | Increase cep.consumer.thread_count (template) |
| API 401 | API key mismatch | passwd.sh PP_CEP_API_KEY |
| Console login denied | PP_CEP_WEB_PASSWORD empty | Web console login account |