企业官网建设流程全解析

Nightingale 集成 EMQX：基于 Dashboard API 的 MQTT 集群指标采集与告警实战

【免费下载链接】nightingaleNightingale is to monitoring and alerting what Grafana is to visualization.项目地址: https://gitcode.com/GitHub_Trending/ni/nightingale

EMQX 插件是 Nightingale 生态中用于监控 EMQX 分布式 MQTT 消息中间件的内置集成：它不依赖额外的 Agent 协议，而是直接调用 EMQX 自身的 Dashboard REST API 拉取节点状态、集群统计与消息指标，随后把指标送入 Nightingale 进行可视化与告警。读完本文，你将掌握input.emqx插件的完整配置方法、三大 API 与指标名之间的映射规律、仓库自带告警规则的用法，以及 EMQX 5.X 下更轻量的 Prometheus 原生采集方案。

一、插件原理与整体工作方式

从源码结构与目录组织看，Nightingale 的每个集成组件都遵循统一布局：integrations/<组件>/下包含collect/（categraf 采集配置）、alerts/（内置告警规则）、markdown/（说明文档）、i18n/（国际化词条）与icon/（图标）。EMQX 组件同样如此，其采集配置见 integrations/EMQX/collect/emqx/emqx.toml，告警规则见 integrations/EMQX/alerts/emqx_by_categraf.json。

插件的数据流可以概括为三步：

1. 配置实例：在 categraf 的input.emqx下配置 EMQX Dashboard 地址、版本与鉴权信息；
2. 轮询 API：采集器按interval周期依次调用/api/${version}/nodes、/api/${version}/stats、/api/${version}/nodes/${node}/metrics三个 REST 接口；
3. 转换上报：把接口返回的 JSON 字段名中的.（点）替换为_（下划线）并统一加上emqx_前缀，形成 Prometheus 风格的时序指标上报给 Nightingale 的数据源。

指标命名遵循一个非常直观的规律：API 字段名中的点号即指标名中的下划线。例如connections.count会变成emqx_connections_count，packets.publish.received会变成emqx_packets_publish_received。掌握这条规律后，面对 EMQX 5.X 中更丰富的字段也能自行推断指标名。

二、插件配置详解

插件通过 EMQX 的 Dashboard API 获取指标，核心配置如下（来自 integrations/EMQX/markdown/README.md 与 integrations/EMQX/collect/emqx/emqx.toml）：

# # 采集周期, 这里指定的 interval 会覆盖 config.toml 中的 interval # interval = 60 [[instances]] ## emqx 的 dashboard 地址，ip:port, 单个集群监控使用逗号分割多个 IP, 会随机挑一个可用 ip 进行连接请求 ## 多个集群，拆分到多个 instance 配置 address="http://192.168.11.11:18083,http://192.168.11.12:18083" ## 指定 emqx 的版本，支持 v3/v4/v5, 分别对应 emqx 的 3.X/4.X/5.X 版本 version="v3" ## api 需要鉴权，输入用户名和密码 username="admin" password="public"

配置参数说明：

两个重要的拓扑实践：

• 单个集群多节点：把集群内所有 Dashboard 地址用逗号拼进同一个address，插件会自动随机挑一个可用 IP 请求。这样即使某个节点的 Dashboard 不可用，采集也不会中断。
• 多个集群：一个[[instances]]对应一个集群，监控多个 EMQX 集群时应拆分成多个[[instances]]配置块，每个实例单独配置地址、版本与凭据。

三、三大 API 与指标映射全解

3.1/api/${version}/nodes：节点级基础指标

插件先调用/api/${version}/nodes获取每个节点的版本、运行状态、连接数、负载、内存、进程数与运行时长等信息。接口返回data数组，每个元素对应一个节点：

{ "code": 0, "data": [ { "connections": 0, "load1": "0.33", "load15": "0.37", "load5": "0.36", "max_fds": 1048576, "memory_total": "191.77M", "memory_used": "129.38M", "name": "emqx@node2.emqx.io", "node": "emqx@node2.emqx.io", "node_status": "Running", "otp_release": "R21/10.3.5.6", "process_available": 2097152, "process_used": 509, "uptime": "3 hours, 12 minutes, 25 seconds", "version": "3.2.6" } ] }

对应生成的节点级指标如下（node标签取自返回中的节点名）：

emqx_node_connections{node="xxx"} 10 emqx_node_status{node="xxxx"} 1 # 1表示running emqx_node_info{node="xxxx", name="xxxx", version="3.2.6", otp_release="R21/10.3.5.6"} 1 emqx_node_load1{node="xxxx"} 0.10 emqx_node_load5{node="xxxx"} 0.37 emqx_node_load15{node="xxxx"} 0.32 emqx_node_max_fds{node="xxxx"} 1048576 emqx_node_memory_total_bytes{node="xxxx"} 194938 emqx_node_memory_used_bytes{node="xxxx"} 132956 emqx_node_process_available{node="xxxx"} 2097152 emqx_node_process_used{node="xxxx"} 508 emqx_node_uptime_seconds{node="xxxx"} 1234567 #单位秒

几个值得注意的点：

• emqx_node_status是 0/1 布尔型指标，1 表示节点 running，这是节点存活告警的直接依据；
• emqx_node_memory_total_bytes/emqx_node_memory_used_bytes把接口返回的191.77M这类可读字符串统一换算为字节，便于计算内存使用率；
• emqx_node_info是一个携带version、otp_release等版本信息的常量指标（值恒为 1），可用于按版本维度统计与过滤；
• emqx_node_uptime_seconds已换算为秒，可用于检测节点是否发生过近期重启。

此外，插件会根据所有节点的状态额外生成两个集群级指标，cluster标签的值就是配置中的address，分别表示当前集群中有多少个节点正常运行、多少个节点已停止：

emqx_cluster_node_running{cluster="xxx"} 2 emqx_cluster_node_stopped{cluster="xxx"} 0

3.2/api/${version}/stats：集群统计指标

插件调用/api/${version}/stats获取每个节点的统计信息，覆盖连接、订阅、主题、会话、保留消息、规则与路由等维度。接口返回data数组，每个元素以node字段开头，其后是xxx.count/xxx.max成对出现的统计字段：

{ "code": 0, "data": [ { "node": "emqx@node2.emqx.io", "subscriptions.shared.max": 0, "subscriptions.max": 0, "subscribers.max": 0, "resources.max": 0, "topics.count": 0, "subscriptions.count": 0, "suboptions.max": 0, "topics.max": 0, "sessions.persistent.max": 0, "connections.max": 0, "sessions.persistent.count": 0, "actions.count": 5, "retained.count": 5, "rules.count": 0, "routes.count": 0, "subscriptions.shared.count": 0, "suboptions.count": 0, "sessions.count": 0, "actions.max": 5, "retained.max": 5, "sessions.max": 0, "rules.max": 0, "routes.max": 0, "resources.count": 0, "subscribers.count": 0, "connections.count": 0 } ] }

根据返回的数据生成如下指标（count表示当前值，max表示历史峰值/上限）：

emqx_subscriptions_shared_max{node="xxx"} 123 emqx_subscriptions_max{node="xxx"} 123 emqx_subscribers_max{node="xxx"} 123 emqx_resources_max{node="xxx"} 123 emqx_topics_count{node="xxx"} 0 emqx_subscriptions_count{node="xxx"} 0 emqx_suboptions_max{node="xxx"} 0 emqx_topics_max{node="xxx"} 0 emqx_sessions_persistent_max{node="xxx"} 0 emqx_connections_max{node="xxx"} 0 emqx_sessions_persistent_count{node="xxx"} 0 emqx_actions_count{node="xxx"} 5 emqx_retained_count{node="xxx"} 5 emqx_rules_count{node="xxx"} 0 emqx_routes_count{node="xxx"} 0 emqx_subscriptions_shared_count{node="xxx"} 0 emqx_suboptions_count{node="xxx"} 0 emqx_sessions_count{node="xxx"} 0 emqx_actions_max{node="xxx"} 5 emqx_retained_max{node="xxx"} 5 emqx_sessions_max{node="xxx"} 0 emqx_rules_max{node="xxx"} 0 emqx_routes_max{node="xxx"} 0 emqx_resources_count{node="xxx"} 0 emqx_subscribers_count{node="xxx"} 0 emqx_connections_count{node="xxx"} 0

这套指标的价值在于容量评估：connections、sessions、topics、subscriptions的count与max配对后，可以判断当前集群负载离历史峰值还有多大余量，是规划扩容的客观依据。

3.3/api/${version}/nodes/${node}/metrics：消息与报文级指标

插件进一步调用/api/${version}/nodes/${node}/metrics获取每个节点细粒度的消息与 MQTT 报文指标，包括消息收发、各类型报文（CONNECT、PUBLISH、SUBSCRIBE、PINGREQ 等）的收发与错误计数、认证情况、规则引擎动作成功率等：

{ "code": 0, "data": { "rules.matched": 0, "messages.sent": 0, "packets.disconnect.sent": 0, "bytes.sent": 0, "packets.disconnect.received": 0, "packets.pingresp.sent": 0, "packets.pingreq.received": 0, "packets.unsubscribe.received": 0, "packets.pubcomp.missed": 0, "packets.puback.missed": 0, "packets.pubcomp.sent": 0, "packets.pubcomp.received": 0, "packets.pubrec.missed": 0, "auth.mqtt.anonymous": 0, "packets.connack.auth_error": 0, "actions.failure": 0, "packets.suback.sent": 0, "packets.puback.sent": 0, "messages.retained": 5, "messages.received": 0, "packets.connect.received": 0, "messages.forward": 0, "packets.pubrel.missed": 0, "packets.publish.received": 0, "packets.connack.sent": 0, "packets.subscribe.received": 0, "packets.pubrel.received": 0, "packets.pubrec.received": 0, "packets.puback.received": 0, "packets.sent": 0, "packets.received": 0, "bytes.received": 0, "messages.expired": 0, "messages.dropped": 0, "messages.qos2.dropped": 0, "messages.qos2.expired": 0, "packets.pubrel.sent": 0, "packets.pubrec.sent": 0, "packets.publish.sent": 0, "actions.success": 0, "packets.publish.error": 0, "packets.unsubscribe.error": 0, "messages.qos2.received": 0, "messages.qos1.received": 0, "messages.qos0.received": 0, "packets.auth.sent": 0, "messages.qos2.sent": 0, "messages.qos1.sent": 0, "messages.qos0.sent": 0, "packets.auth.received": 0, "packets.unsuback.sent": 0, "packets.connack.error": 0, "packets.publish.auth_error": 0, "packets.subscribe.error": 0, "packets.subscribe.auth_error": 0 } }

对应生成的指标如下：

emqx_rules_matched{node="xxx"} 0 emqx_messages_sent{node="xxx"} 0 emqx_packets_disconnect_sent{node="xxx"} 0 emqx_bytes_sent{node="xxx"} 0 emqx_packets_disconnect_received{node="xxx"} 0 emqx_packets_pingresp_sent{node="xxx"} 0 emqx_packets_pingreq_received{node="xxx"} 0 emqx_packets_unsubscribe_received{node="xxx"} 0 emqx_packets_pubcomp_missed{node="xxx"} 0 emqx_packets_puback_missed{node="xxx"} 0 emqx_packets_pubcomp_sent{node="xxx"} 0 emqx_packets_pubcomp_received{node="xxx"} 0 emqx_packets_pubrec_missed{node="xxx"} 0 emqx_auth_mqtt_anonymous{node="xxx"} 0 emqx_packets_connack_auth_error{node="xxx"} 0 emqx_actions_failure{node="xxx"} 0 emqx_packets_suback_sent{node="xxx"} 0 emqx_packets_puback_sent{node="xxx"} 0 emqx_messages_retained{node="xxx"} 5 emqx_messages_received{node="xxx"} 0 emqx_packets_connect_received{node="xxx"} 0 emqx_messages_forward{node="xxx"} 0 emqx_packets_pubrel_missed{node="xxx"} 0 emqx_packets_publish_received{node="xxx"} 0 emqx_packets_connack_sent{node="xxx"} 0 emqx_packets_subscribe_received{node="xxx"} 0 emqx_packets_pubrel_received{node="xxx"} 0 emqx_packets_pubrec_received{node="xxx"} 0 emqx_packets_puback_received{node="xxx"} 0 emqx_packets_sent{node="xxx"} 0 emqx_packets_received{node="xxx"} 0 emqx_bytes_received{node="xxx"} 0 emqx_messages_expired{node="xxx"} 0 emqx_messages_dropped{node="xxx"} 0 emqx_messages_qos2_dropped{node="xxx"} 0 emqx_messages_qos2_expired{node="xxx"} 0 emqx_packets_pubrel_sent{node="xxx"} 0 emqx_packets_pubrec_sent{node="xxx"} 0 emqx_packets_publish_sent{node="xxx"} 0 emqx_actions_success{node="xxx"} 0 emqx_packets_publish_error{node="xxx"} 0 emqx_packets_unsubscribe_error{node="xxx"} 0 emqx_messages_qos2_received{node="xxx"} 0 emqx_messages_qos1_received{node="xxx"} 0 emqx_messages_qos0_received{node="xxx"} 0 emqx_packets_auth_sent{node="xxx"} 0 emqx_messages_qos2_sent{node="xxx"} 0 emqx_messages_qos1_sent{node="xxx"} 0 emqx_messages_qos0_sent{node="xxx"} 0 emqx_packets_auth_received{node="xxx"} 0 emqx_packets_unsuback_sent{node="xxx"} 0 emqx_packets_connack_error{node="xxx"} 0 emqx_packets_publish_auth_error{node="xxx"} 0 emqx_packets_subscribe_error{node="xxx"} 0 emqx_packets_subscribe_auth_error{node="xxx"} 0

这一组指标刻画了消息管道与协议层的健康状况。特别值得关注的是emqx_messages_dropped（消息丢弃）、emqx_packets_connack_auth_error（连接认证失败）、emqx_actions_failure/emqx_actions_success（规则引擎动作成败）——它们也是仓库内置告警规则的直接依赖项（见下一节）。由于这些是计数器指标，在告警中应使用rate()计算速率而非直接比较原始值。

四、仓库内置告警规则实战

Nightingale 为 EMQX 组件预置了 9 条开箱即用的告警规则，全部定义在 integrations/EMQX/alerts/emqx_by_categraf.json 中，类型为prometheus、生产类别metric，可直接导入 Prometheus 类数据源使用。规则清单如下：

这些规则的默认配置（prom_eval_interval15 秒、notify_repeat_step60 秒、notify_recovered开启、enable_stime/enable_etime全天 00:00–23:59、disabled为 1）在导入后可根据业务调整；每条规则都带有append_tags形式的告警标识（如alertname=EmqxNodeNotRunning、alertname=EmqxClusterNodeStopped），便于在 Nightingale 中做告警聚合与路由。

更难得的是，每条规则都内置了完整的排查 Runbook（annotations.action字段），例如：

• 节点不在运行状态：登录节点执行emqx ctl status与systemctl status emqx确认进程；进程存活但状态异常的查看log/emqx.log.1尾部（常见为 Erlang 虚拟机内存耗尽或磁盘写满触发节点自保）；进程不在则先df -h、free -g排除资源问题再启动；恢复后执行emqx ctl cluster status确认节点重新加入集群并观察连接数回流。
• 连接数逼近 fd 上限：执行emqx ctl listeners查看各监听器连接配置，用cat /proc/$(pgrep -f beam.smp)/limits | grep 'open files'确认实际 fd 上限，调大 systemd 的LimitNOFILE与 emqx.conf 的node.max_ports并重启（注意评估内存，每连接约占几十 KB），同时在网关侧限制异常客户端疯狂重连。
• Erlang 进程数逼近上限：emqx ctl vm查看 process/used 与 process/limit 实时值，区分正常业务增长与会话泄漏（大量clean_session=false的持久会话累积），必要时调大node.process_limit。
• 内存使用率过高：emqx ctl vm memory查看 processes/binary/ets 分布——binary 占比高多为消息积压于会话队列，应急可调低mqueue_max_len限制单会话队列长度，避免个别慢消费者拖垮节点。
• 消息丢弃激增：emqx ctl metrics | grep dropped区分丢弃原因（queue_full / no_subscribers / expired）后分别处理。
• 认证失败激增：查看log/emqx.log中失败的 clientid 与来源 IP 分布，区分认证后端故障与暴力破解（后者在网关侧限流并开黑名单）。
• 规则引擎动作失败率过高：在 Dashboard 规则引擎页面定位失败规则，区分下游不可达、超时还是数据格式不匹配。
• 节点近期重启：先排除计划内发布/扩缩容，再查journalctl -u emqx --since '-30min'与dmesg -T | grep -i 'killed process'排除 OOM，重启后对设备重连做限速避免重连风暴。

这些规则的中英文文案由 integrations/EMQX/i18n/en_US.json 提供，英文环境下展示的告警名称与 Runbook 均为英文译文。

五、EMQX 5.X 的 Prometheus 原生采集方案

原文档特别指出：5.X 的 EMQX 提供了原生的 Prometheus 接口，可以完全不使用上述 Dashboard API 插件方案，而是直接用 categraf 的input.prometheus插件采集：

http://ip:18083/api/v5/prometheus/stats

这条路径相比 Dashboard API 方案有两个明显优势：一是无需额外配置用户名密码认证，部署更简单；二是原生暴露的即为 Prometheus 文本格式，指标语义与_count/_max等命名更统一，配合 Prometheus 生态的 relabel 与 recording rule 也更顺手。

需要注意的适用前提：

• 该方案仅适用于EMQX 5.X；3.X / 4.X 没有此接口，必须使用本文前几章的 Dashboard API 插件（version分别设为v3/v4）；
• 若集群存在多个节点，需要为每个节点各配置一个input.prometheus采集目标（Dashboard API 方案则只需在address中逗号分隔多 IP 即可自动容错），或用服务发现机制自动发现节点；
• input.prometheus与input.emqx两种方式产出的指标命名不同，告警规则需对应调整，不能混用。

六、集成资源在 Nightingale 中的落地方式

最后从源码角度说明这些集成资源是如何进入 Nightingale 运行时的：

1. 内置集成初始化：center/integration/init.go 会在启动时扫描integrations/目录。对每个组件，依次装载markdown/下的 README（含README.en_US.md等多语言副本）、alerts/*.json（解析为告警规则并写入builtin_payloads）、dashboards/、metrics/等资源，并把icon/下的图标注册为组件 Logo，供集成中心页面展示与一键导入。若设置了disable_integration_init配置项，则跳过该初始化过程。
2. AI 文档检索索引：aiagent/tools/integrations_loader.go 会把每个组件的markdown/README.md转成[integration-doc]条目、collect/*/*.toml转成[integration-config]条目并入文档索引，使 AI 助手通过search_n9e_docs检索时能直接命中真实的[[instances]]写法——这意味着本文介绍的配置样例不仅是人读的文档，也是 AI 配置助手的事实依据。
3. 采集端落地：采集配置最终由 categraf 加载执行，指标经 Nightingale 的 Prometheus 数据源入库后，即可在仪表盘中绘图，并配合上一节的告警规则实现闭环监控。

至此，从「categraf 拉取 EMQX Dashboard API」到「指标入库」再到「内置告警 + 排障 Runbook」的完整链路已经打通：EMQX 3.X/4.X 用input.emqx插件按本文配置，5.X 可按需切换到input.prometheus原生方案，两层手段共同覆盖了 MQTT 集群的全生命周期监控需求。

【免费下载链接】nightingaleNightingale is to monitoring and alerting what Grafana is to visualization.项目地址: https://gitcode.com/GitHub_Trending/ni/nightingale