白皮书
车联网设计与实现:搭建可靠、高效、符合行业需求的车联网平台 →

使用 Prometheus 监控 eKuiper 规则运行状态

Jiyong Huang
2022-9-7
使用 Prometheus 监控 eKuiper 规则运行状态

Prometheus 是一个托管于 CNCF 的开源系统监控和警报工具包,许多公司和组织都采用了 Prometheus 作为监控告警工具。

eKuiper 的规则是一个持续运行的流式计算任务。规则用于处理无界的数据流,正常情况下,规则启动后会一直运行,不断产生运行状态数据。直到规则被手动停止或出现不可恢复的错误后停止。eKuiper 中的规则提供了状态 API,可获取规则的运行指标。同时,eKuiper 整合了 Prometheus,可方便地通过后者监控各种状态指标。

本教程面向已经初步了解 eKuiper 的用户,将介绍规则状态指标以及如何通过 Prometheus 监控特定的指标。

规则状态指标

使用 eKuiper 创建规则并运行成功后,用户可以通过 CLI、REST API 或者管理控制台查看规则的运行状态指标。例如,已有规则 rule1,可通过 curl -X GET "<http://127.0.0.1:9081/rules/rule1/status"> 获取 JSON 格式的规则运行指标,如下所示:

{
  "status": "running",
  "source_demo_0_records_in_total": 265,
  "source_demo_0_records_out_total": 265,
  "source_demo_0_process_latency_us": 0,
  "source_demo_0_buffer_length": 0,
  "source_demo_0_last_invocation": "2022-08-22T17:19:10.979128",
  "source_demo_0_exceptions_total": 0,
  "source_demo_0_last_exception": "",
  "source_demo_0_last_exception_time": 0,
  "op_2_project_0_records_in_total": 265,
  "op_2_project_0_records_out_total": 265,
  "op_2_project_0_process_latency_us": 0,
  "op_2_project_0_buffer_length": 0,
  "op_2_project_0_last_invocation": "2022-08-22T17:19:10.979128",
  "op_2_project_0_exceptions_total": 0,
  "op_2_project_0_last_exception": "",
  "op_2_project_0_last_exception_time": 0,
  "sink_mqtt_0_0_records_in_total": 265,
  "sink_mqtt_0_0_records_out_total": 265,
  "sink_mqtt_0_0_process_latency_us": 0,
  "sink_mqtt_0_0_buffer_length": 0,
  "sink_mqtt_0_0_last_invocation": "2022-08-22T17:19:10.979128",
  "sink_mqtt_0_0_exceptions_total": 0,
  "sink_mqtt_0_0_last_exception": "",
  "sink_mqtt_0_0_last_exception_time": 0
}

运行指标主要包括两个部分,一部分是 status,用于标示规则是否正常运行,其值可能为 running, stopped manually 等。另一部分为规则每个算子的运行指标。规则的算子根据规则的 SQL 生成,每个规则可能会有所不同。在此例中,规则 SQL 为最简单的 SELECT * FROM demo, action 为 MQTT,其生成的算子为 [source_demo, op_project, sink_mqtt] 3 个。每一种算子都有相同数目的运行指标,与算子名字合起来构成一条指标。例如,算子 source_demo_0 的输入数量 records_in_total 的指标为 source_demo_0_records_in_total。

运行指标

每个算子的运行指标是相同的,主要有以下几种:

  • records_in_total:读入的消息总量,表示规则启动后处理了多少条消息。
  • records_out_total:输出的消息总量,表示算子正确处理的消息数量。
  • process_latency_us:最近一次处理的延时,单位为微妙。该值为瞬时值,可了解算子的处理性能。整体规则的延时一般由延时最大的算子决定。
  • buffer_length:算子缓冲区长度。由于算子之间计算速度会有差异,各个算子之间都有缓冲队列。缓冲区长度较大的话说明算子处理较慢,赶不上上游处理速度。
  • last_invocation:算子的最后一次运行的时间。
  • exceptions_total:异常总量。算子运行中产生的非不可恢复的错误,例如连接中断,数据格式错误等均计入异常,而不会中断规则。

在 1.6.1 版本以后,我们又添加了两个异常相关指标,方便异常的调试处理。

  • last_exception:最近一次的异常的错误信息。
  • last_exception_time:最近一次异常的发生时间。

这些运行指标中的数值类型指标均可使用 Prometheus 进行监控。下一节我们将描述如何配置 eKuiper 中的 Prometheus 服务。

配置 eKuiper 的 Prometheus 服务

eKuiper 中自带 Prometheus 服务,但是默认为关闭状态。用户可修改 etc/kuiper.yaml 中的配置打开该服务。其中,prometheus 为布尔值,修改为 true 可打开服务;prometheusPort 配置服务的访问端口。

  prometheus: true
  prometheusPort: 20499

若使用 Docker 启动 eKuiper,也可通过配置环境变量启用服务。

docker run -p 9081:9081 -d --name ekuiper MQTT_SOURCE__DEFAULT__SERVER="$MQTT_BROKER_ADDRESS" KUIPER__BASIC__PROMETHEUS=true lfedge/ekuiper:$tag

在启动的日志中,可以看到服务启动的相关信息,例如:

time="2022-08-22 17:16:50" level=info msg="Serving prometheus metrics on port <http://localhost:20499/metrics"> file="server/prome_init.go:60"
Serving prometheus metrics on port <http://localhost:20499/metrics>

点击提示中的地址 http://localhost:20499/metrics ,可查看到 Prometheus 中搜集到的 eKuiper 的原始指标信息。eKuiper 有规则正常运行之后,可以在页面中搜索到类似 kuiper_sink_records_in_total 等的指标。用户可以配置 Prometheus 接入 eKuiper,进行更丰富的展示。

使用 Prometheus 查看状态

上文我们已经实现了将 eKuiper 状态输出为 Prometheus 指标的功能,接下来我们可以配置 Prometheus 接入这一部分指标,并完成初步的监控。

安装和配置

Prometheus 官方网站 下载所需要的系统版本然后解压。

修改配置文件,使其监控 eKuiper。打开 prometheus.yml,修改 scrape_configs 部分,如下所示:

global:
  scrape_interval:     15s
  evaluation_interval: 15s

rule_files:
  # - "first.rules"
  # - "second.rules"

scrape_configs:
  - job_name: ekuiper
    static_configs:
      - targets: ['localhost:20499']

此处定义了监控任务名为 eKuiper, targets 指向上一节启动的服务的地址。配置完成后,启动 Prometheus 。

./prometheus --config.file=prometheus.yml

启动成功后,打开 http://localhost:9090/ 可进入管理控制台。

简单监控

监控所有规则的 sink 接收到的消息数目变化。可以在如图的搜索框中输入需要监控的指标名称,点击 Execute 即可生成监控表。选择 Graph 可切换为折线图等展示方式。

点击 Add Panel,通过同样的配置方式,可监控更多的指标。

总结

本文介绍了 eKuiper 中的规则状态指标以及如何使用 Prometheus 简单地监控这些状态指标。用户可以基于此进一步探索 Prometheus 的更多高级功能,更好地实现 eKuiper 的运维。

推荐阅读

2020-5-21eKuiper Team
Kuiper 正式成为 EdgeX 规则引擎

在 EdgeX Geneva 版本中, EMQX Kuiper - 基于 SQL 的轻量级流式数据处理软件与 EdgeX 进行了集成。本教程将介绍如何使用 Kuiper 处理来自于 EdgeX 的数据。