SkyWalking Pulsar 监控实战：从 OpenTelemetry Collector 到 MAL 指标规则的全链路解析

SkyWalking Pulsar 监控实战从 OpenTelemetry Collector 到 MAL 指标规则的全链路解析【免费下载链接】skywalkingAPM, Application Performance Monitoring System项目地址: https://gitcode.com/gh_mirrors/sk/skywalking本篇基于 SkyWalking 仓库中 Pulsar 监控的官方文档与配套源码讲解如何通过 OpenTelemetry Collector 采集 Pulsar 集群Broker BookKeeper的 Prometheus 指标、经 OTLP gRPC 推送到 OAP Server并由 MAL 规则加工后以Layer: PULSAR的 Service/Instance 维度落库。读完本文你可以完整搭建一套 Pulsar 监控系统理解otel-rules/pulsar/下集群级与节点级指标规则的实现细节并学会按需自定义指标。工作原理与数据流SkyWalking 本身不直接抓取 Pulsar 的 Prometheus 指标而是借用 OpenTelemetry Collector 作为中间层。Pulsar 集群中的 Broker 在8080端口Admin API暴露 Prometheus 格式的 metricsCollector 通过 Prometheus Receiver 周期抓取再通过 OTLP gRPC exporter 推送到 OAP Server 的 OpenTelemetry receiver最终由 MALMetrics as a Language规则过滤、计算、聚合后存入 Meter System。整个数据流分三步Pulsar 通过 Prometheus endpoint 暴露指标OpenTelemetry Collector 通过 Prometheus Receiver 从 Pulsar 集群拉取指标并通过 OpenTelemetry gRPC exporter 推送到 SkyWalking OAP ServerOAP Server 按 MAL 表达式解析、过滤/计算/聚合并将结果存储。在 OAP 的实体模型中一个 Pulsar 集群被建模为一个Layer: PULSAR的Service集群维度集群中的每个节点broker、bookie则表现为该 Service 下的Instance节点维度。这个建模方式与 Kafka、RocketMQ 等其他消息中间件的监控方案一致。部署步骤第一步准备 Pulsar 集群Pulsar 集群由 Pulsar Broker 集群和 BookKeeper bookie 集群两部分组成二者都提供 Prometheus 格式的指标端点。仓库的 e2e 测试环境 docker-compose.yml 给出了一个最小可参考的拓扑zookeeper3.9.1bookieapachepulsar/pulsar:3.1.1暴露 8000 端口的 Prometheus metricsbroker同镜像暴露 8080 端口 两个pulsar-perf生产/消费负载容器 otel-collector。第二步配置 OpenTelemetry Collectore2e 用例中的 otel-collector-config.yaml 是官方示例配置关键点逐条说明receivers: prometheus: config: scrape_configs: - job_name: pulsar-monitoring # 关键该 job 名是 OAP 侧 MAL 规则的过滤条件 scrape_interval: 15s static_configs: - targets: [broker:8080] # Pulsar broker 的 Prometheus endpoint relabel_configs: - source_labels: [ __address__ ] regex: (.) target_label: node # 将地址如 broker:8080改写为 node 标签 replacement: $$1 - job_name: bookkeeper-monitoring scrape_interval: 10s static_configs: - targets: [bookie:8000] # BookKeeper bookie 的 Prometheus endpoint labels: cluster: pulsar-cluster # 显式打上 cluster 标签 relabel_configs: - source_labels: [ __address__ ] regex: (.) target_label: node replacement: $$1 processors: batch: exporters: otlp: endpoint: oap:11800 # OAP 的 OTLP gRPC 端口 tls: insecure: true logging: loglevel: debug service: pipelines: metrics: receivers: [prometheus] processors: [batch] exporters: [otlp, logging]配置中有三个细节直接决定 OAP 能否正确识别数据job_name必须与 OAP 侧 MAL 规则的filter一致。Pulsar broker 的抓取任务命名为pulsar-monitoringBookKeeper 的命名为bookkeeper-monitoring这两个值分别对应 OAP 规则文件中的过滤条件。node标签通过 relabel 从__address__生成。OAP 的 MAL 规则会用[cluster, node]作为聚合/实例维度node即实例名例如broker:8080。cluster标签标识集群归属。集群级指标会按cluster标签聚合规则中会把 Service 名改写为pulsar::cluster前缀形式。关于job_name为何能进入 MAL 的tagsOpenTelemetry Collector 的 Prometheus receiver 会把 Prometheus 的job标签转换为 OTLP 资源属性service.name而 OAP 的 OTel receiver 会将service.name兜底映射为job_name标签供 MAL 规则过滤使用。这一点在 OpenTelemetry receiver 文档的 Fallback label mappings 一节中有明确说明。第三步启用 OAP 的 OpenTelemetry receiverOAP 侧需要启用 OTLP 处理器并加载 Pulsar 相关的 MAL 规则文件。在 application.yml 中receiver-otel的配置形如receiver-otel: selector: ${SW_OTEL_RECEIVER:default} default: enabledHandlers: ${SW_OTEL_RECEIVER_ENABLED_HANDLERS:otlp-metrics} enabledOtelMetricsRules: ${SW_OTEL_RECEIVER_ENABLED_OTEL_METRICS_RULES:apisix,nginx/*,k8s/*,...,pulsar/*,bookkeeper/*,...}两个要点需要设置环境变量SW_OTEL_RECEIVERdefault或将receiver-otel/selector改为 defaultreceiver 才会激活这是 OpenTelemetry receiver 文档中特别标注的注意事项默认开启的规则列表中已包含pulsar/*与bookkeeper/*即仓库自带的 otel-rules/pulsar/ 与 otel-rules/bookkeeper/ 目录下所有规则默认全部生效通常无需修改。集群级指标pulsar-cluster.yaml 规则解析集群级指标规则位于 pulsar-cluster.yaml其头部结构为filter: { tags - tags.job_name pulsar-monitoring } # 只处理 job_name 为 pulsar-monitoring 的样本 expSuffix: tag({tags - tags.cluster pulsar:: tags.cluster}).service([cluster], Layer.PULSAR) metricPrefix: meter_pulsarfilterGroovy 风格表达式只让job_name pulsar-monitoring的样本进入本规则因此 Broker 与 BookKeeper 的指标在 Collector 侧必须使用不同的 job 名expSuffix先把cluster标签值改写为pulsar::前缀例如pulsar::pulsar-cluster再按[cluster]维度创建Layer.PULSAR的 Service 实体——这就是 UI/GraphQL 中看到的集群 Service 名metricPrefix所有指标名统一加meter_pulsar前缀。metricsRules部分对每一条规则都做了sum([cluster, node])聚合即把集群内所有 broker 节点的同名 Prometheus 指标求和。仓库支持的集群级指标如下与 官方文档的指标表一一对应监控面板指标名说明数据来源Total Topicsmeter_pulsar_total_topics该集群中 Pulsar topic 的数量Pulsar ClusterTotal Subscriptionsmeter_pulsar_total_subscriptions该集群中 Pulsar 订阅的数量Pulsar ClusterTotal Producersmeter_pulsar_total_producers连接到此集群的活跃 producer 数量Pulsar ClusterTotal Consumersmeter_pulsar_total_consumers连接到此集群的活跃 consumer 数量Pulsar ClusterMessage Rate Inmeter_pulsar_message_rate_in进入该集群的总消息速率条/秒Pulsar ClusterMessage Rate Outmeter_pulsar_message_rate_out从该集群发出的总消息速率条/秒Pulsar ClusterThroughput Inmeter_pulsar_throughput_in进入该集群的总吞吐量字节/秒Pulsar ClusterThroughput Outmeter_pulsar_throughput_out从该集群发出的总吞吐量字节/秒Pulsar ClusterStorage Sizemeter_pulsar_storage_size该 broker 所有 topic 的存储总大小字节Pulsar ClusterStorage Logical Sizemeter_pulsar_storage_logical_size该 broker 所有 topic 去除副本后的存储大小字节Pulsar ClusterStorage Write Ratemeter_pulsar_storage_write_rate该 broker 写入存储的消息批次速率batch/秒Pulsar ClusterStorage Read Ratemeter_pulsar_storage_read_rate该 broker 从存储读取的消息批次速率batch/秒Pulsar Cluster对应的 MAL 表达式示例exp中的指标名是 Pulsar 原生 Prometheus 指标名metricsRules: - name: total_topics exp: pulsar_broker_topics_count.sum([cluster, node]) - name: total_subscriptions exp: pulsar_broker_subscriptions_count.sum([cluster, node]) - name: total_producers exp: pulsar_broker_producers_count.sum([cluster, node]) - name: total_consumers exp: pulsar_broker_consumers_count.sum([cluster, node]) - name: message_rate_in exp: pulsar_broker_rate_in.sum([cluster, node]) - name: message_rate_out exp: pulsar_broker_rate_out.sum([cluster, node]) - name: throughput_in exp: pulsar_broker_throughput_in.sum([cluster, node]) - name: throughput_out exp: pulsar_broker_throughput_out.sum([cluster, node]) - name: storage_size exp: pulsar_broker_storage_size.sum([cluster, node]) - name: storage_logical_size exp: pulsar_broker_storage_logical_size.sum([cluster, node]) - name: storage_write_rate exp: pulsar_broker_storage_write_rate.sum([cluster, node]) - name: storage_read_rate exp: pulsar_broker_storage_read_rate.sum([cluster, node])节点级Broker指标pulsar-broker.yaml 规则解析Broker 节点级规则位于 pulsar-broker.yaml头部结构为filter: { tags - tags.job_name pulsar-monitoring } expSuffix: tag({tags - tags.cluster pulsar:: tags.cluster}).instance([cluster], [node], Layer.PULSAR) metricPrefix: meter_pulsar_broker与集群规则的区别在于expSuffix使用了instance([cluster], [node], Layer.PULSAR)在pulsar::cluster这个 Service 下按node标签创建 Instance 实体。从 e2e 用例 pulsar-cases.yaml 可以印证查询实例指标时使用的--instance-namebroker:8080正是 Collector 侧 relabel 写入的__address__值。支持的 Broker 节点级指标如下完整继承自官方文档监控面板指标名说明数据来源Active Connectionsmeter_pulsar_broker_active_connections活跃连接数Pulsar BrokerTotal Connectionsmeter_pulsar_broker_total_connections连接总数Pulsar BrokerConnection Create Success Countmeter_pulsar_broker_connection_create_success_count成功创建的连接数Pulsar BrokerConnection Create Fail Countmeter_pulsar_broker_connection_create_fail_count创建失败的连接数Pulsar BrokerConnection Closed Total Countmeter_pulsar_broker_connection_closed_total_count关闭的连接总数Pulsar BrokerJVM Buffer Pool Usedmeter_pulsar_broker_jvm_buffer_pool_used_bytesJVM buffer pool 用量Pulsar BrokerJVM Memory Pool Usedmeter_pulsar_broker_jvm_memory_pool_usedJVM 内存池用量Pulsar BrokerJVM Memorymeter_pulsar_broker_jvm_memory_init / meter_pulsar_broker_jvm_memory_used / meter_pulsar_broker_jvm_memory_committedJVM 内存用量Pulsar BrokerJVM Threadsmeter_pulsar_broker_jvm_threads_current / _daemon / _peak / _deadlockedJVM 线程数Pulsar BrokerGC Timemeter_pulsar_broker_jvm_gc_collection_seconds_sum各 GC 收集器耗时秒Pulsar BrokerGC Countmeter_pulsar_broker_jvm_gc_collection_seconds_count各 GC 收集器的收集次数Pulsar BrokerMAL 规则实现上有两个值得注意的写法连接类指标直接sum([cluster, node])例如pulsar_active_connections.sum([cluster, node])、pulsar_connection_created_total_count.sum([cluster, node])GC 相关指标jvm_gc_collection_seconds_count与jvm_gc_collection_seconds_sum是 Prometheus counter 类型规则上额外套了.rate(PT1M)即按 1 分钟窗口计算速率使 GC 时间与次数呈现为分钟级速率而非累计值内存池与线程类指标按[cluster, node, pool]或[cluster, node, gc]聚合保留了pool/gc维度标签便于在 UI 中区分具体内存池如 G1 Eden、G1 Old与具体 GC 收集器。BookKeeper 侧的配套监控Pulsar 的存储底座是 BookKeeper因此除pulsar/*规则外OAP 还自带了 otel-rules/bookkeeper/bookkeeper-cluster.yaml 与bookkeeper-node.yaml。以集群规则为例filter: { tags - tags.job_name bookkeeper-monitoring } expSuffix: tag({tags - tags.cluster bookkeeper:: tags.cluster}).service([cluster], Layer.BOOKKEEPER) metricPrefix: meter_bookkeeper即 BookKeeper 以独立的Layer.BOOKKEEPERServicebookkeeper::cluster建模与pulsar::cluster并列展示。e2e 用例 pulsar-cases.yaml 中同时校验了两类服务实体service ls期望返回pulsar::pulsar-cluster与bookkeeper::pulsar-cluster并分别通过instance ls验证 broker 实例broker:8080与 bookie 实例bookie:8000。集群级 BookKeeper 指标涵盖meter_bookkeeper_bookie_ledgers_count、meter_bookkeeper_bookie_entries_count、读写缓存_write_cache_size/count、_read_cache_size/count以及经rate(PT1M)处理的meter_bookkeeper_bookie_write_rate/meter_bookkeeper_bookie_read_rate等。端到端验证仓库 e2e 框架e2e.yaml以 docker-compose 拉起整套环境zookeeper、bookie、broker、pulsar-perf 生产/消费、otel-collector、OAP BanyanDB验证阶段以swctl对 OAP 的 GraphQL 接口12800 端口执行查询重试 60 次、间隔 3 秒直到指标出现。典型验证命令有三类# 1. 集群 Service 是否存在 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql service ls # 2. 节点 Instance 是否注册 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql instance ls --service-namepulsar::pulsar-cluster # 3. 集群指标是否有值按 cluster 标签聚合求和 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql metrics exec \ --expressionaggregate_labels(meter_pulsar_total_topics,sum) --service-namepulsar::pulsar-cluster # 4. 节点指标是否有值按实例查询 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql metrics exec \ --expressionmeter_pulsar_broker_active_connections --service-namepulsar::pulsar-cluster \ --instance-namebroker:8080e2e 用例对全部 12 条集群级指标和 12 条 broker 级指标逐一断言有值期望文件位于 expected/ 目录service.yml、broker_instance.yml、bookie_instance.yml、metrics-has-value.yml等可作为自建环境后的对照清单。自定义指标与面板当内置指标不够用时可以自定义指标、表达式或 dashboard 面板。指标定义与表达式规则就写在otel-rules/pulsar/pulsar-cluster.yaml与otel-rules/pulsar/pulsar-broker.yaml仓库内实际路径为 oap-server/server-starter/src/main/resources/otel-rules/pulsar/中。自定义时在对应文件中按既有格式追加规则即可例如把某条 Pulsar 原生 Prometheus 指标pulsar_xxx按集群聚合后命名为meter_pulsar_xxx- name: xxx exp: pulsar_xxx.sum([cluster, node])编写自定义规则时需注意两点约束均来自 OpenTelemetry receiver 文档receiver-otel采用推push模式因此规则文件中只支持 MAL schema 的group、defaultMetricLevel与metricsRules节点不支持完整 MAL 文件的其他节点规则文件在 OAP 启动时加载若新配置格式不正确OAP 可能启动失败因此修改后应校验 YAML 与表达式语法。另外Pulsar 的 dashboard 面板配置由 SkyWalking Horizon UI 发行包apache/skywalking-horizon-ui提供OAP 后端不再托管 UI 的 dashboard JSON如需扩展面板应在 UI 侧进行。小结SkyWalking 的 Pulsar 监控方案有三个核心设计值得复用一是职责分离——采集交给 OpenTelemetry Collector 的 Prometheus receiverOAP 只负责 OTLP 接入与 MAL 加工二是标签驱动的实体建模——job_name决定数据归属哪条 MAL 规则cluster与node标签决定 Service/Instance 实体的切分方式最终呈现为pulsar::cluster/bookkeeper::cluster两个 Service 及其下的节点实例三是规则即配置——所有指标定义集中在otel-rules/目录的 YAML 文件中新增或调整指标只需修改表达式并重启或重新加载OAP无需改动 Java 代码。【免费下载链接】skywalkingAPM, Application Performance Monitoring System项目地址: https://gitcode.com/gh_mirrors/sk/skywalking创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考