控實戰(zhàn)：從 OpenTelemetry Collector 到 MAL 指標(biāo)規(guī)則的全鏈路解析)
SkyWalking Pulsar 監(jiān)控實戰(zhàn)從 OpenTelemetry Collector 到 MAL 指標(biāo)規(guī)則的全鏈路解析【免費下載鏈接】skywalkingAPM, Application Performance Monitoring System項目地址: https://gitcode.com/gh_mirrors/sk/skywalking本篇基于 SkyWalking 倉庫中 Pulsar 監(jiān)控的官方文檔與配套源碼講解如何通過 OpenTelemetry Collector 采集 Pulsar 集群Broker BookKeeper的 Prometheus 指標(biāo)、經(jīng) OTLP gRPC 推送到 OAP Server并由 MAL 規(guī)則加工后以Layer: PULSAR的 Service/Instance 維度落庫。讀完本文你可以完整搭建一套 Pulsar 監(jiān)控系統(tǒng)理解otel-rules/pulsar/下集群級與節(jié)點級指標(biāo)規(guī)則的實現(xiàn)細(xì)節(jié)并學(xué)會按需自定義指標(biāo)。工作原理與數(shù)據(jù)流SkyWalking 本身不直接抓取 Pulsar 的 Prometheus 指標(biāo)而是借用 OpenTelemetry Collector 作為中間層。Pulsar 集群中的 Broker 在8080端口Admin API暴露 Prometheus 格式的 metricsCollector 通過 Prometheus Receiver 周期抓取再通過 OTLP gRPC exporter 推送到 OAP Server 的 OpenTelemetry receiver最終由 MALMetrics as a Language規(guī)則過濾、計算、聚合后存入 Meter System。整個數(shù)據(jù)流分三步Pulsar 通過 Prometheus endpoint 暴露指標(biāo)OpenTelemetry Collector 通過 Prometheus Receiver 從 Pulsar 集群拉取指標(biāo)并通過 OpenTelemetry gRPC exporter 推送到 SkyWalking OAP ServerOAP Server 按 MAL 表達(dá)式解析、過濾/計算/聚合并將結(jié)果存儲。在 OAP 的實體模型中一個 Pulsar 集群被建模為一個Layer: PULSAR的Service集群維度集群中的每個節(jié)點broker、bookie則表現(xiàn)為該 Service 下的Instance節(jié)點維度。這個建模方式與 Kafka、RocketMQ 等其他消息中間件的監(jiān)控方案一致。部署步驟第一步準(zhǔn)備 Pulsar 集群Pulsar 集群由 Pulsar Broker 集群和 BookKeeper bookie 集群兩部分組成二者都提供 Prometheus 格式的指標(biāo)端點。倉庫的 e2e 測試環(huán)境 docker-compose.yml 給出了一個最小可參考的拓?fù)鋤ookeeper3.9.1bookieapachepulsar/pulsar:3.1.1暴露 8000 端口的 Prometheus metricsbroker同鏡像暴露 8080 端口 兩個pulsar-perf生產(chǎn)/消費負(fù)載容器 otel-collector。第二步配置 OpenTelemetry Collectore2e 用例中的 otel-collector-config.yaml 是官方示例配置關(guān)鍵點逐條說明receivers: prometheus: config: scrape_configs: - job_name: pulsar-monitoring # 關(guān)鍵該 job 名是 OAP 側(cè) MAL 規(guī)則的過濾條件 scrape_interval: 15s static_configs: - targets: [broker:8080] # Pulsar broker 的 Prometheus endpoint relabel_configs: - source_labels: [ __address__ ] regex: (.) target_label: node # 將地址如 broker:8080改寫為 node 標(biāo)簽 replacement: $$1 - job_name: bookkeeper-monitoring scrape_interval: 10s static_configs: - targets: [bookie:8000] # BookKeeper bookie 的 Prometheus endpoint labels: cluster: pulsar-cluster # 顯式打上 cluster 標(biāo)簽 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]配置中有三個細(xì)節(jié)直接決定 OAP 能否正確識別數(shù)據(jù)job_name必須與 OAP 側(cè) MAL 規(guī)則的filter一致。Pulsar broker 的抓取任務(wù)命名為pulsar-monitoringBookKeeper 的命名為bookkeeper-monitoring這兩個值分別對應(yīng) OAP 規(guī)則文件中的過濾條件。node標(biāo)簽通過 relabel 從__address__生成。OAP 的 MAL 規(guī)則會用[cluster, node]作為聚合/實例維度node即實例名例如broker:8080。cluster標(biāo)簽標(biāo)識集群歸屬。集群級指標(biāo)會按cluster標(biāo)簽聚合規(guī)則中會把 Service 名改寫為pulsar::cluster前綴形式。關(guān)于job_name為何能進(jìn)入 MAL 的tagsOpenTelemetry Collector 的 Prometheus receiver 會把 Prometheus 的job標(biāo)簽轉(zhuǎn)換為 OTLP 資源屬性service.name而 OAP 的 OTel receiver 會將service.name兜底映射為job_name標(biāo)簽供 MAL 規(guī)則過濾使用。這一點在 OpenTelemetry receiver 文檔的 Fallback label mappings 一節(jié)中有明確說明。第三步啟用 OAP 的 OpenTelemetry receiverOAP 側(cè)需要啟用 OTLP 處理器并加載 Pulsar 相關(guān)的 MAL 規(guī)則文件。在 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/*,...}兩個要點需要設(shè)置環(huán)境變量SW_OTEL_RECEIVERdefault或?qū)eceiver-otel/selector改為 defaultreceiver 才會激活這是 OpenTelemetry receiver 文檔中特別標(biāo)注的注意事項默認(rèn)開啟的規(guī)則列表中已包含pulsar/*與bookkeeper/*即倉庫自帶的 otel-rules/pulsar/ 與 otel-rules/bookkeeper/ 目錄下所有規(guī)則默認(rèn)全部生效通常無需修改。集群級指標(biāo)pulsar-cluster.yaml 規(guī)則解析集群級指標(biāo)規(guī)則位于 pulsar-cluster.yaml其頭部結(jié)構(gòu)為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 風(fēng)格表達(dá)式只讓job_name pulsar-monitoring的樣本進(jìn)入本規(guī)則因此 Broker 與 BookKeeper 的指標(biāo)在 Collector 側(cè)必須使用不同的 job 名expSuffix先把cluster標(biāo)簽值改寫為pulsar::前綴例如pulsar::pulsar-cluster再按[cluster]維度創(chuàng)建Layer.PULSAR的 Service 實體——這就是 UI/GraphQL 中看到的集群 Service 名metricPrefix所有指標(biāo)名統(tǒng)一加meter_pulsar前綴。metricsRules部分對每一條規(guī)則都做了sum([cluster, node])聚合即把集群內(nèi)所有 broker 節(jié)點的同名 Prometheus 指標(biāo)求和。倉庫支持的集群級指標(biāo)如下與 官方文檔的指標(biāo)表一一對應(yīng)監(jiān)控面板指標(biāo)名說明數(shù)據(jù)來源Total Topicsmeter_pulsar_total_topics該集群中 Pulsar topic 的數(shù)量Pulsar ClusterTotal Subscriptionsmeter_pulsar_total_subscriptions該集群中 Pulsar 訂閱的數(shù)量Pulsar ClusterTotal Producersmeter_pulsar_total_producers連接到此集群的活躍 producer 數(shù)量Pulsar ClusterTotal Consumersmeter_pulsar_total_consumers連接到此集群的活躍 consumer 數(shù)量Pulsar ClusterMessage Rate Inmeter_pulsar_message_rate_in進(jìn)入該集群的總消息速率條/秒Pulsar ClusterMessage Rate Outmeter_pulsar_message_rate_out從該集群發(fā)出的總消息速率條/秒Pulsar ClusterThroughput Inmeter_pulsar_throughput_in進(jìn)入該集群的總吞吐量字節(jié)/秒Pulsar ClusterThroughput Outmeter_pulsar_throughput_out從該集群發(fā)出的總吞吐量字節(jié)/秒Pulsar ClusterStorage Sizemeter_pulsar_storage_size該 broker 所有 topic 的存儲總大小字節(jié)Pulsar ClusterStorage Logical Sizemeter_pulsar_storage_logical_size該 broker 所有 topic 去除副本后的存儲大小字節(jié)Pulsar ClusterStorage Write Ratemeter_pulsar_storage_write_rate該 broker 寫入存儲的消息批次速率batch/秒Pulsar ClusterStorage Read Ratemeter_pulsar_storage_read_rate該 broker 從存儲讀取的消息批次速率batch/秒Pulsar Cluster對應(yīng)的 MAL 表達(dá)式示例exp中的指標(biāo)名是 Pulsar 原生 Prometheus 指標(biāo)名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])節(jié)點級Broker指標(biāo)pulsar-broker.yaml 規(guī)則解析Broker 節(jié)點級規(guī)則位于 pulsar-broker.yaml頭部結(jié)構(gòu)為filter: { tags - tags.job_name pulsar-monitoring } expSuffix: tag({tags - tags.cluster pulsar:: tags.cluster}).instance([cluster], [node], Layer.PULSAR) metricPrefix: meter_pulsar_broker與集群規(guī)則的區(qū)別在于expSuffix使用了instance([cluster], [node], Layer.PULSAR)在pulsar::cluster這個 Service 下按node標(biāo)簽創(chuàng)建 Instance 實體。從 e2e 用例 pulsar-cases.yaml 可以印證查詢實例指標(biāo)時使用的--instance-namebroker:8080正是 Collector 側(cè) relabel 寫入的__address__值。支持的 Broker 節(jié)點級指標(biāo)如下完整繼承自官方文檔監(jiān)控面板指標(biāo)名說明數(shù)據(jù)來源Active Connectionsmeter_pulsar_broker_active_connections活躍連接數(shù)Pulsar BrokerTotal Connectionsmeter_pulsar_broker_total_connections連接總數(shù)Pulsar BrokerConnection Create Success Countmeter_pulsar_broker_connection_create_success_count成功創(chuàng)建的連接數(shù)Pulsar BrokerConnection Create Fail Countmeter_pulsar_broker_connection_create_fail_count創(chuàng)建失敗的連接數(shù)Pulsar BrokerConnection Closed Total Countmeter_pulsar_broker_connection_closed_total_count關(guān)閉的連接總數(shù)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 內(nèi)存池用量Pulsar BrokerJVM Memorymeter_pulsar_broker_jvm_memory_init / meter_pulsar_broker_jvm_memory_used / meter_pulsar_broker_jvm_memory_committedJVM 內(nèi)存用量Pulsar BrokerJVM Threadsmeter_pulsar_broker_jvm_threads_current / _daemon / _peak / _deadlockedJVM 線程數(shù)Pulsar BrokerGC Timemeter_pulsar_broker_jvm_gc_collection_seconds_sum各 GC 收集器耗時秒Pulsar BrokerGC Countmeter_pulsar_broker_jvm_gc_collection_seconds_count各 GC 收集器的收集次數(shù)Pulsar BrokerMAL 規(guī)則實現(xiàn)上有兩個值得注意的寫法連接類指標(biāo)直接sum([cluster, node])例如pulsar_active_connections.sum([cluster, node])、pulsar_connection_created_total_count.sum([cluster, node])GC 相關(guān)指標(biāo)jvm_gc_collection_seconds_count與jvm_gc_collection_seconds_sum是 Prometheus counter 類型規(guī)則上額外套了.rate(PT1M)即按 1 分鐘窗口計算速率使 GC 時間與次數(shù)呈現(xiàn)為分鐘級速率而非累計值內(nèi)存池與線程類指標(biāo)按[cluster, node, pool]或[cluster, node, gc]聚合保留了pool/gc維度標(biāo)簽便于在 UI 中區(qū)分具體內(nèi)存池如 G1 Eden、G1 Old與具體 GC 收集器。BookKeeper 側(cè)的配套監(jiān)控Pulsar 的存儲底座是 BookKeeper因此除pulsar/*規(guī)則外OAP 還自帶了 otel-rules/bookkeeper/bookkeeper-cluster.yaml 與bookkeeper-node.yaml。以集群規(guī)則為例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 中同時校驗了兩類服務(wù)實體service ls期望返回pulsar::pulsar-cluster與bookkeeper::pulsar-cluster并分別通過instance ls驗證 broker 實例broker:8080與 bookie 實例bookie:8000。集群級 BookKeeper 指標(biāo)涵蓋meter_bookkeeper_bookie_ledgers_count、meter_bookkeeper_bookie_entries_count、讀寫緩存_write_cache_size/count、_read_cache_size/count以及經(jīng)rate(PT1M)處理的meter_bookkeeper_bookie_write_rate/meter_bookkeeper_bookie_read_rate等。端到端驗證倉庫 e2e 框架e2e.yaml以 docker-compose 拉起整套環(huán)境zookeeper、bookie、broker、pulsar-perf 生產(chǎn)/消費、otel-collector、OAP BanyanDB驗證階段以swctl對 OAP 的 GraphQL 接口12800 端口執(zhí)行查詢重試 60 次、間隔 3 秒直到指標(biāo)出現(xiàn)。典型驗證命令有三類# 1. 集群 Service 是否存在 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql service ls # 2. 節(jié)點 Instance 是否注冊 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql instance ls --service-namepulsar::pulsar-cluster # 3. 集群指標(biāo)是否有值按 cluster 標(biāo)簽聚合求和 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql metrics exec \ --expressionaggregate_labels(meter_pulsar_total_topics,sum) --service-namepulsar::pulsar-cluster # 4. 節(jié)點指標(biāo)是否有值按實例查詢 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 條集群級指標(biāo)和 12 條 broker 級指標(biāo)逐一斷言有值期望文件位于 expected/ 目錄service.yml、broker_instance.yml、bookie_instance.yml、metrics-has-value.yml等可作為自建環(huán)境后的對照清單。自定義指標(biāo)與面板當(dāng)內(nèi)置指標(biāo)不夠用時可以自定義指標(biāo)、表達(dá)式或 dashboard 面板。指標(biāo)定義與表達(dá)式規(guī)則就寫在otel-rules/pulsar/pulsar-cluster.yaml與otel-rules/pulsar/pulsar-broker.yaml倉庫內(nèi)實際路徑為 oap-server/server-starter/src/main/resources/otel-rules/pulsar/中。自定義時在對應(yīng)文件中按既有格式追加規(guī)則即可例如把某條 Pulsar 原生 Prometheus 指標(biāo)pulsar_xxx按集群聚合后命名為meter_pulsar_xxx- name: xxx exp: pulsar_xxx.sum([cluster, node])編寫自定義規(guī)則時需注意兩點約束均來自 OpenTelemetry receiver 文檔receiver-otel采用推push模式因此規(guī)則文件中只支持 MAL schema 的group、defaultMetricLevel與metricsRules節(jié)點不支持完整 MAL 文件的其他節(jié)點規(guī)則文件在 OAP 啟動時加載若新配置格式不正確OAP 可能啟動失敗因此修改后應(yīng)校驗 YAML 與表達(dá)式語法。另外Pulsar 的 dashboard 面板配置由 SkyWalking Horizon UI 發(fā)行包apache/skywalking-horizon-ui提供OAP 后端不再托管 UI 的 dashboard JSON如需擴(kuò)展面板應(yīng)在 UI 側(cè)進(jìn)行。小結(jié)SkyWalking 的 Pulsar 監(jiān)控方案有三個核心設(shè)計值得復(fù)用一是職責(zé)分離——采集交給 OpenTelemetry Collector 的 Prometheus receiverOAP 只負(fù)責(zé) OTLP 接入與 MAL 加工二是標(biāo)簽驅(qū)動的實體建模——job_name決定數(shù)據(jù)歸屬哪條 MAL 規(guī)則cluster與node標(biāo)簽決定 Service/Instance 實體的切分方式最終呈現(xiàn)為pulsar::cluster/bookkeeper::cluster兩個 Service 及其下的節(jié)點實例三是規(guī)則即配置——所有指標(biāo)定義集中在otel-rules/目錄的 YAML 文件中新增或調(diào)整指標(biāo)只需修改表達(dá)式并重啟或重新加載OAP無需改動 Java 代碼?！久赓M下載鏈接】skywalkingAPM, Application Performance Monitoring System項目地址: https://gitcode.com/gh_mirrors/sk/skywalking創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 SkyWalking Pulsar 監(jiān)控實戰(zhàn)：從 OpenTelemetry Collector 到 MAL 指標(biāo)規(guī)則的全鏈路解析
SkyWalking Pulsar 監(jiān)控實戰(zhàn)從 OpenTelemetry Collector 到 MAL 指標(biāo)規(guī)則的全鏈路解析【免費下載鏈接】skywalkingAPM, Application Performance Monitoring System項目地址: https://gitcode.com/gh_mirrors/sk/skywalking本篇基于 SkyWalking 倉庫中 Pulsar 監(jiān)控的官方文檔與配套源碼講解如何通過 OpenTelemetry Collector 采集 Pulsar 集群Broker BookKeeper的 Prometheus 指標(biāo)、經(jīng) OTLP gRPC 推送到 OAP Server并由 MAL 規(guī)則加工后以Layer: PULSAR的 Service/Instance 維度落庫。讀完本文你可以完整搭建一套 Pulsar 監(jiān)控系統(tǒng)理解otel-rules/pulsar/下集群級與節(jié)點級指標(biāo)規(guī)則的實現(xiàn)細(xì)節(jié)并學(xué)會按需自定義指標(biāo)。工作原理與數(shù)據(jù)流SkyWalking 本身不直接抓取 Pulsar 的 Prometheus 指標(biāo)而是借用 OpenTelemetry Collector 作為中間層。Pulsar 集群中的 Broker 在8080端口Admin API暴露 Prometheus 格式的 metricsCollector 通過 Prometheus Receiver 周期抓取再通過 OTLP gRPC exporter 推送到 OAP Server 的 OpenTelemetry receiver最終由 MALMetrics as a Language規(guī)則過濾、計算、聚合后存入 Meter System。整個數(shù)據(jù)流分三步Pulsar 通過 Prometheus endpoint 暴露指標(biāo)OpenTelemetry Collector 通過 Prometheus Receiver 從 Pulsar 集群拉取指標(biāo)并通過 OpenTelemetry gRPC exporter 推送到 SkyWalking OAP ServerOAP Server 按 MAL 表達(dá)式解析、過濾/計算/聚合并將結(jié)果存儲。在 OAP 的實體模型中一個 Pulsar 集群被建模為一個Layer: PULSAR的Service集群維度集群中的每個節(jié)點broker、bookie則表現(xiàn)為該 Service 下的Instance節(jié)點維度。這個建模方式與 Kafka、RocketMQ 等其他消息中間件的監(jiān)控方案一致。部署步驟第一步準(zhǔn)備 Pulsar 集群Pulsar 集群由 Pulsar Broker 集群和 BookKeeper bookie 集群兩部分組成二者都提供 Prometheus 格式的指標(biāo)端點。倉庫的 e2e 測試環(huán)境 docker-compose.yml 給出了一個最小可參考的拓?fù)鋤ookeeper3.9.1bookieapachepulsar/pulsar:3.1.1暴露 8000 端口的 Prometheus metricsbroker同鏡像暴露 8080 端口 兩個pulsar-perf生產(chǎn)/消費負(fù)載容器 otel-collector。第二步配置 OpenTelemetry Collectore2e 用例中的 otel-collector-config.yaml 是官方示例配置關(guān)鍵點逐條說明receivers: prometheus: config: scrape_configs: - job_name: pulsar-monitoring # 關(guān)鍵該 job 名是 OAP 側(cè) MAL 規(guī)則的過濾條件 scrape_interval: 15s static_configs: - targets: [broker:8080] # Pulsar broker 的 Prometheus endpoint relabel_configs: - source_labels: [ __address__ ] regex: (.) target_label: node # 將地址如 broker:8080改寫為 node 標(biāo)簽 replacement: $$1 - job_name: bookkeeper-monitoring scrape_interval: 10s static_configs: - targets: [bookie:8000] # BookKeeper bookie 的 Prometheus endpoint labels: cluster: pulsar-cluster # 顯式打上 cluster 標(biāo)簽 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]配置中有三個細(xì)節(jié)直接決定 OAP 能否正確識別數(shù)據(jù)job_name必須與 OAP 側(cè) MAL 規(guī)則的filter一致。Pulsar broker 的抓取任務(wù)命名為pulsar-monitoringBookKeeper 的命名為bookkeeper-monitoring這兩個值分別對應(yīng) OAP 規(guī)則文件中的過濾條件。node標(biāo)簽通過 relabel 從__address__生成。OAP 的 MAL 規(guī)則會用[cluster, node]作為聚合/實例維度node即實例名例如broker:8080。cluster標(biāo)簽標(biāo)識集群歸屬。集群級指標(biāo)會按cluster標(biāo)簽聚合規(guī)則中會把 Service 名改寫為pulsar::cluster前綴形式。關(guān)于job_name為何能進(jìn)入 MAL 的tagsOpenTelemetry Collector 的 Prometheus receiver 會把 Prometheus 的job標(biāo)簽轉(zhuǎn)換為 OTLP 資源屬性service.name而 OAP 的 OTel receiver 會將service.name兜底映射為job_name標(biāo)簽供 MAL 規(guī)則過濾使用。這一點在 OpenTelemetry receiver 文檔的 Fallback label mappings 一節(jié)中有明確說明。第三步啟用 OAP 的 OpenTelemetry receiverOAP 側(cè)需要啟用 OTLP 處理器并加載 Pulsar 相關(guān)的 MAL 規(guī)則文件。在 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/*,...}兩個要點需要設(shè)置環(huán)境變量SW_OTEL_RECEIVERdefault或?qū)eceiver-otel/selector改為 defaultreceiver 才會激活這是 OpenTelemetry receiver 文檔中特別標(biāo)注的注意事項默認(rèn)開啟的規(guī)則列表中已包含pulsar/*與bookkeeper/*即倉庫自帶的 otel-rules/pulsar/ 與 otel-rules/bookkeeper/ 目錄下所有規(guī)則默認(rèn)全部生效通常無需修改。集群級指標(biāo)pulsar-cluster.yaml 規(guī)則解析集群級指標(biāo)規(guī)則位于 pulsar-cluster.yaml其頭部結(jié)構(gòu)為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 風(fēng)格表達(dá)式只讓job_name pulsar-monitoring的樣本進(jìn)入本規(guī)則因此 Broker 與 BookKeeper 的指標(biāo)在 Collector 側(cè)必須使用不同的 job 名expSuffix先把cluster標(biāo)簽值改寫為pulsar::前綴例如pulsar::pulsar-cluster再按[cluster]維度創(chuàng)建Layer.PULSAR的 Service 實體——這就是 UI/GraphQL 中看到的集群 Service 名metricPrefix所有指標(biāo)名統(tǒng)一加meter_pulsar前綴。metricsRules部分對每一條規(guī)則都做了sum([cluster, node])聚合即把集群內(nèi)所有 broker 節(jié)點的同名 Prometheus 指標(biāo)求和。倉庫支持的集群級指標(biāo)如下與 官方文檔的指標(biāo)表一一對應(yīng)監(jiān)控面板指標(biāo)名說明數(shù)據(jù)來源Total Topicsmeter_pulsar_total_topics該集群中 Pulsar topic 的數(shù)量Pulsar ClusterTotal Subscriptionsmeter_pulsar_total_subscriptions該集群中 Pulsar 訂閱的數(shù)量Pulsar ClusterTotal Producersmeter_pulsar_total_producers連接到此集群的活躍 producer 數(shù)量Pulsar ClusterTotal Consumersmeter_pulsar_total_consumers連接到此集群的活躍 consumer 數(shù)量Pulsar ClusterMessage Rate Inmeter_pulsar_message_rate_in進(jìn)入該集群的總消息速率條/秒Pulsar ClusterMessage Rate Outmeter_pulsar_message_rate_out從該集群發(fā)出的總消息速率條/秒Pulsar ClusterThroughput Inmeter_pulsar_throughput_in進(jìn)入該集群的總吞吐量字節(jié)/秒Pulsar ClusterThroughput Outmeter_pulsar_throughput_out從該集群發(fā)出的總吞吐量字節(jié)/秒Pulsar ClusterStorage Sizemeter_pulsar_storage_size該 broker 所有 topic 的存儲總大小字節(jié)Pulsar ClusterStorage Logical Sizemeter_pulsar_storage_logical_size該 broker 所有 topic 去除副本后的存儲大小字節(jié)Pulsar ClusterStorage Write Ratemeter_pulsar_storage_write_rate該 broker 寫入存儲的消息批次速率batch/秒Pulsar ClusterStorage Read Ratemeter_pulsar_storage_read_rate該 broker 從存儲讀取的消息批次速率batch/秒Pulsar Cluster對應(yīng)的 MAL 表達(dá)式示例exp中的指標(biāo)名是 Pulsar 原生 Prometheus 指標(biāo)名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])節(jié)點級Broker指標(biāo)pulsar-broker.yaml 規(guī)則解析Broker 節(jié)點級規(guī)則位于 pulsar-broker.yaml頭部結(jié)構(gòu)為filter: { tags - tags.job_name pulsar-monitoring } expSuffix: tag({tags - tags.cluster pulsar:: tags.cluster}).instance([cluster], [node], Layer.PULSAR) metricPrefix: meter_pulsar_broker與集群規(guī)則的區(qū)別在于expSuffix使用了instance([cluster], [node], Layer.PULSAR)在pulsar::cluster這個 Service 下按node標(biāo)簽創(chuàng)建 Instance 實體。從 e2e 用例 pulsar-cases.yaml 可以印證查詢實例指標(biāo)時使用的--instance-namebroker:8080正是 Collector 側(cè) relabel 寫入的__address__值。支持的 Broker 節(jié)點級指標(biāo)如下完整繼承自官方文檔監(jiān)控面板指標(biāo)名說明數(shù)據(jù)來源Active Connectionsmeter_pulsar_broker_active_connections活躍連接數(shù)Pulsar BrokerTotal Connectionsmeter_pulsar_broker_total_connections連接總數(shù)Pulsar BrokerConnection Create Success Countmeter_pulsar_broker_connection_create_success_count成功創(chuàng)建的連接數(shù)Pulsar BrokerConnection Create Fail Countmeter_pulsar_broker_connection_create_fail_count創(chuàng)建失敗的連接數(shù)Pulsar BrokerConnection Closed Total Countmeter_pulsar_broker_connection_closed_total_count關(guān)閉的連接總數(shù)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 內(nèi)存池用量Pulsar BrokerJVM Memorymeter_pulsar_broker_jvm_memory_init / meter_pulsar_broker_jvm_memory_used / meter_pulsar_broker_jvm_memory_committedJVM 內(nèi)存用量Pulsar BrokerJVM Threadsmeter_pulsar_broker_jvm_threads_current / _daemon / _peak / _deadlockedJVM 線程數(shù)Pulsar BrokerGC Timemeter_pulsar_broker_jvm_gc_collection_seconds_sum各 GC 收集器耗時秒Pulsar BrokerGC Countmeter_pulsar_broker_jvm_gc_collection_seconds_count各 GC 收集器的收集次數(shù)Pulsar BrokerMAL 規(guī)則實現(xiàn)上有兩個值得注意的寫法連接類指標(biāo)直接sum([cluster, node])例如pulsar_active_connections.sum([cluster, node])、pulsar_connection_created_total_count.sum([cluster, node])GC 相關(guān)指標(biāo)jvm_gc_collection_seconds_count與jvm_gc_collection_seconds_sum是 Prometheus counter 類型規(guī)則上額外套了.rate(PT1M)即按 1 分鐘窗口計算速率使 GC 時間與次數(shù)呈現(xiàn)為分鐘級速率而非累計值內(nèi)存池與線程類指標(biāo)按[cluster, node, pool]或[cluster, node, gc]聚合保留了pool/gc維度標(biāo)簽便于在 UI 中區(qū)分具體內(nèi)存池如 G1 Eden、G1 Old與具體 GC 收集器。BookKeeper 側(cè)的配套監(jiān)控Pulsar 的存儲底座是 BookKeeper因此除pulsar/*規(guī)則外OAP 還自帶了 otel-rules/bookkeeper/bookkeeper-cluster.yaml 與bookkeeper-node.yaml。以集群規(guī)則為例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 中同時校驗了兩類服務(wù)實體service ls期望返回pulsar::pulsar-cluster與bookkeeper::pulsar-cluster并分別通過instance ls驗證 broker 實例broker:8080與 bookie 實例bookie:8000。集群級 BookKeeper 指標(biāo)涵蓋meter_bookkeeper_bookie_ledgers_count、meter_bookkeeper_bookie_entries_count、讀寫緩存_write_cache_size/count、_read_cache_size/count以及經(jīng)rate(PT1M)處理的meter_bookkeeper_bookie_write_rate/meter_bookkeeper_bookie_read_rate等。端到端驗證倉庫 e2e 框架e2e.yaml以 docker-compose 拉起整套環(huán)境zookeeper、bookie、broker、pulsar-perf 生產(chǎn)/消費、otel-collector、OAP BanyanDB驗證階段以swctl對 OAP 的 GraphQL 接口12800 端口執(zhí)行查詢重試 60 次、間隔 3 秒直到指標(biāo)出現(xiàn)。典型驗證命令有三類# 1. 集群 Service 是否存在 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql service ls # 2. 節(jié)點 Instance 是否注冊 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql instance ls --service-namepulsar::pulsar-cluster # 3. 集群指標(biāo)是否有值按 cluster 標(biāo)簽聚合求和 swctl --display yaml --base-urlhttp://${oap_host}:${oap_12800}/graphql metrics exec \ --expressionaggregate_labels(meter_pulsar_total_topics,sum) --service-namepulsar::pulsar-cluster # 4. 節(jié)點指標(biāo)是否有值按實例查詢 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 條集群級指標(biāo)和 12 條 broker 級指標(biāo)逐一斷言有值期望文件位于 expected/ 目錄service.yml、broker_instance.yml、bookie_instance.yml、metrics-has-value.yml等可作為自建環(huán)境后的對照清單。自定義指標(biāo)與面板當(dāng)內(nèi)置指標(biāo)不夠用時可以自定義指標(biāo)、表達(dá)式或 dashboard 面板。指標(biāo)定義與表達(dá)式規(guī)則就寫在otel-rules/pulsar/pulsar-cluster.yaml與otel-rules/pulsar/pulsar-broker.yaml倉庫內(nèi)實際路徑為 oap-server/server-starter/src/main/resources/otel-rules/pulsar/中。自定義時在對應(yīng)文件中按既有格式追加規(guī)則即可例如把某條 Pulsar 原生 Prometheus 指標(biāo)pulsar_xxx按集群聚合后命名為meter_pulsar_xxx- name: xxx exp: pulsar_xxx.sum([cluster, node])編寫自定義規(guī)則時需注意兩點約束均來自 OpenTelemetry receiver 文檔receiver-otel采用推push模式因此規(guī)則文件中只支持 MAL schema 的group、defaultMetricLevel與metricsRules節(jié)點不支持完整 MAL 文件的其他節(jié)點規(guī)則文件在 OAP 啟動時加載若新配置格式不正確OAP 可能啟動失敗因此修改后應(yīng)校驗 YAML 與表達(dá)式語法。另外Pulsar 的 dashboard 面板配置由 SkyWalking Horizon UI 發(fā)行包apache/skywalking-horizon-ui提供OAP 后端不再托管 UI 的 dashboard JSON如需擴(kuò)展面板應(yīng)在 UI 側(cè)進(jìn)行。小結(jié)SkyWalking 的 Pulsar 監(jiān)控方案有三個核心設(shè)計值得復(fù)用一是職責(zé)分離——采集交給 OpenTelemetry Collector 的 Prometheus receiverOAP 只負(fù)責(zé) OTLP 接入與 MAL 加工二是標(biāo)簽驅(qū)動的實體建模——job_name決定數(shù)據(jù)歸屬哪條 MAL 規(guī)則cluster與node標(biāo)簽決定 Service/Instance 實體的切分方式最終呈現(xiàn)為pulsar::cluster/bookkeeper::cluster兩個 Service 及其下的節(jié)點實例三是規(guī)則即配置——所有指標(biāo)定義集中在otel-rules/目錄的 YAML 文件中新增或調(diào)整指標(biāo)只需修改表達(dá)式并重啟或重新加載OAP無需改動 Java 代碼?！久赓M下載鏈接】skywalkingAPM, Application Performance Monitoring System項目地址: https://gitcode.com/gh_mirrors/sk/skywalking創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考