plantpulse-cep (複合イベント処理)
役割
Complex Event Processing エンジンです。複数のセンサイベントの時間的・論理的パターンをリアルタイムで検知してアラーム・Flow トリガー・自動化を実行します。
| 項目 | 値 |
|---|---|
| モジュール名 | plantpulse-cep |
| インストール経路 | /opt/kopens/plantpulse-platform/plantpulse-cep/ |
| ポート | 7400 (HTTP) · 7401 (TLS) |
| エンジン | Esper CEP (内蔵 Tomcat 上の Web アプリ) |
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.<date>.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://<server-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 ラグ増加 | 処理スレッド不足 | cep.consumer.thread_count を増設 (テンプレート) |
| API 401 | API キー不一致 | passwd.sh PP_CEP_API_KEY |
| コンソールログイン拒否 | PP_CEP_WEB_PASSWORD が空 | Web 画面ログインアカウント |