跳到主要内容

plantpulse-cep (复合事件处理)

角色​

Complex Event Processing 层。实时检测多个事件的时间和逻辑模式，以执行报警、Flow 触发和自动化。规则使用 EQL(PlantPulse CEP SQL) 编写。

项目值
模块名plantpulse-cep
安装路径/opt/kopens/plantpulse-platform/plantpulse-cep/
端口7400 (HTTP) · 7401 (TLS)
执行引擎启动时仅限一个 — Aster CEP(cep.backend=aster, 未设置时也相同)
pd 服务cep — 仅 MASTER
API 认证X-API-Key: <PP_CEP_API_KEY>
控制台登录PP_CEP_WEB_USER / PP_CEP_WEB_PASSWORD — 若为空则仅关闭控制台
语言 · 后端 · 运维 · 限制请参阅专用章节

本页涵盖 模块与文件。关于语句的编写和运行方法，请参阅 CEP 章节。

使用示例​

“若触发报警后 30 分钟内 未收到解除信号，则上报为未解除”

SELECT a.machineId AS machine
FROM pattern [every a=AlarmEvent -> not ClearEvent for 30 min]

构成​

API · 控制台 · 引擎位于同一进程。运行的引擎仅为启动时选择的 一个。

目录结构​

plantpulse-cep/
├── config/plantpulse-cep.properties     # 생성물 ← plantpulse-datalake-cli 의 템플릿
├── bin/start.sh
├── logs/
└── server/                              # 내장 Tomcat — webapps/ROOT.war · logs/catalina.<date>.log

主要配置​

# 백엔드 — 부팅 시 하나. 미설정이면 이전 것을 그대로 유지한다
cep.backend=aster
#cep.mode=native                       # native | portable

# 이벤트 입력 (Valkey 스트림)
cache.host=${PP_REDIS_HOST}
cache.port=${PP_REDIS_PORT}
cache.password=${PP_REDIS_PASSWORD}
cache.ssl.enabled=false
cep.stream.shard_count=16
cep.consumer.thread_count=16

# 룰 · 이벤트 타입 레지스트리 (PostgreSQL datalake DB — CEP 가 표를 스스로 만든다)
jdbc.datalake.url=jdbc:postgresql://${PP_POSTGRES_HOST}:${PP_POSTGRES_PORT}/datalake
jdbc.datalake.username=${PP_DATALAKE_DB_USER}
jdbc.datalake.password=${PP_DATALAKE_DB_PASSWORD}

# 운영 지표 (Cassandra)
storage.host=${PP_CASSANDRA_HOST}
storage.keyspace=${PP_KEYSPACE}
cep.storage.ttl_seconds=864000        # 10일

# 인증
cep.api.key=${PP_CEP_API_KEY}
cep.web.user=${PP_CEP_WEB_USER}
cep.web.password=${PP_CEP_WEB_PASSWORD}
cep.api.max_eql_text_length=100000
仅在以 Flink 启动时 cep.flink.* 才有意义

同一文件中保留有 cep.flink.reconciler.enabled · cep.flink.router.enabled。若以其他后端启动，这些配置将被忽略，且名称仅在日志中记录一次。

曾出现过在开启状态下以其他后端启动，导致 同一语句在两个引擎中运行并重复发布结果 的情况 → 后端

修改值的步骤请参阅 修改配置方法，API 密钥轮换请参阅 更改密码 · API 密钥。

主要 API​

路径功能
POST /api/v1/eql/deploy语句注册
POST /api/v1/eql/validate仅判定 — SUPPORTED / UNSUPPORTED / UNPARSEABLE
DELETE /api/v1/eql/undeploy/{id}注销注册
GET /api/v1/eql/registration注册结果表 (ACTIVE / REJECTED)
GET /api/v1/eql/portability可移植性
GET /api/v1/eql/metrics/snapshot · /metrics/engine每条语句的匹配率 · 输出率，引擎指标
GET·POST·PUT·DELETE /api/v1/event-types事件类型注册
GET·POST·DELETE /api/v1/keysAPI 密钥
GET /api/health匿名就绪状态 — 引擎 · 缓存 · 模式

运维命令​

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

# 어느 백엔드로 떴나 — 판정 지표는 로그 한 줄이다
docker exec plantpulse-datalake sh -c 'grep "CEP input path" ... | tail -1'

curl -fsS http://127.0.0.1:7400/api/health

常见问题​

现象原因措施
注册状态为 REJECTED启动的后端无法接收该配置在 拒绝令牌 中查找该行
注册被 400 拒绝未注册的事件类型事件类型注册
CEP 完全无法启动cep.backend 拼写错误或 WAR 中缺少该运行时日志会提示名称。无备选方案
无结果输出订阅者为 0查看日志中的 no subscriber
结果重复接收在 cep.flink.* 开关开启时以其他后端启动后端
健康状态 DEGRADED，原因为表名称缺失 CEP 专用表查看 DB 连接(jdbc.datalake.*)并重启 — 启动时会自动创建
API 401API 密钥不匹配passwd.sh PP_CEP_API_KEY
控制台登录被拒绝PP_CEP_WEB_PASSWORD 为空Web 界面登录账号

相关文档​