本文是基于Pulsar 2.9.2/kop-2.9.2展開的。

一、背景

作為一種新型消息中間件，Pulsar在架構(gòu)設計及功能特性等方面要優(yōu)于Kafka，所以我們引入Pulsar作為我們新一代的消息中間件。在對Pulsar進行調(diào)研的時候（比如：性能測試、故障測試等），針對Pulsar提供一套可觀測系統(tǒng)是必不可少的。Pulsar的指標是面向云原生的，并且官方提供了Prometheus作為Pulsar指標的采集、存儲和查詢的方案，但是使用Prometheus采集指標面臨以下幾個問題：

1. Prometheus自帶的時序數(shù)據(jù)庫不是分布式的，它受單機資源的限制；
2. Prometheus 在存儲時序數(shù)據(jù)時消耗大量的內(nèi)存，并且Prometheus在實現(xiàn)高效查詢和聚合計算的時候會消耗大量的CPU。

除了以上列出的可觀測系統(tǒng)問題，Pulsar還有一些指標本身的問題，這些問題包括：

1. Pulsar的訂閱積壓指標單位是entry而不是條數(shù)，這會嚴重影響從Kafka遷移過來的用戶的使用體驗及日常運維工作；
2. Pulsar沒有bundle指標，因為Pulsar自動均衡的最小單位是bundle，所以bundle指標是調(diào)試Pulsar自動均衡參數(shù)時重要的觀測依據(jù)；
3. kop指標上報異常等問題。

針對以上列出的幾個問題，我們在下面分別展開敘述。

二、Pulsar監(jiān)控告警系統(tǒng)架構(gòu)

在上一章節(jié)我們列出了使用Prometheus作為觀測系統(tǒng)的局限，由于Pulsar的指標是面向云原生的，采用Prometheus采集Pulsar指標是最好的選擇，但對于指標的存儲和查詢我們使用第三方存儲來減輕Prometheus的壓力，整個監(jiān)控告警系統(tǒng)架構(gòu)如下圖所示：

在整個可觀測系統(tǒng)中，各組件的職能如下：

• Pulsar、bookkeeper等組件提供暴露指標的接口
• Prometheus訪問Pulsar指標接口采集指標
• adaptor提供了服務發(fā)現(xiàn)、Prometheus格式指標的反序列化和序列化以及指標轉(zhuǎn)發(fā)遠端存儲的能力，這里的遠端存儲可以是Pulsar或Kafka
• Druid消費指標topic并提供數(shù)據(jù)分析的能力
• vivo內(nèi)部的檢測告警平臺提供了動態(tài)配置檢測告警的能力

基于以上監(jiān)控系統(tǒng)的設計邏輯，我們在具體實現(xiàn)的過程中遇到了幾個比較關(guān)鍵的問題：

1、adaptor需要接收Pulsar所有線上服務的指標并兼容Prometheus格式數(shù)據(jù)，所以在調(diào)研Prometheus采集Pulsar指標時，我們基于Prometheus的官方文檔開發(fā)了adaptor，在adaptor里實現(xiàn)了服務加入集群的發(fā)現(xiàn)機制以及動態(tài)配置prometheus采集新新加入服務的指標：

• Prometheus動態(tài)加載配置：Prometheus配置-官方文檔
• Prometheus自定義服務發(fā)現(xiàn)機制：Prometheus自定義服務發(fā)現(xiàn)-官方文檔

在可以動態(tài)配置Prometheus采集所有線上正在運行的服務指標之后，由于Prometheus的指標是基于protobuf協(xié)議進行傳輸?shù)?，并且Prometheus是基于go編寫的，所以為了適配Java版本的adaptor，我們基于Prometheus和go提供的指標格式定義文件（remote.proto、types.proto和gogo.proto）生成了Java版本的指標接收代碼，并將protobuf格式的指標反序列化后寫入消息中間件。

2、Grafana社區(qū)提供的Druid插件不能很好的展示Counter類型的指標，但是bookkeeper上報的指標中又有很多是Counter類型的指標，vivo的Druid團隊對該插件做了一些改造，新增了計算速率的聚合函數(shù)。

druid插件的安裝可以參考官方文檔（詳情）

3、由于Prometheus比較依賴內(nèi)存和CPU，而我們的機器資源組又是有限的，在使用遠端存儲的基礎(chǔ)上，我們針對該問題優(yōu)化了一些Prometheus參數(shù)，這些參數(shù)包括：

• --storage.tsdb.retentinotallow=30m：該參數(shù)配置了數(shù)據(jù)的保留時間為30分鐘，在這個時間之后，舊的數(shù)據(jù)將會被刪除。
• --storage.tsdb.min-block-duratinotallow=5m：該參數(shù)配置了生成塊（block）的最小時間間隔為5分鐘。塊是一組時序數(shù)據(jù)的集合，它們通常被一起壓縮和存儲在磁盤上，該參數(shù)間接控制Prometheus對內(nèi)存的占用。
• --storage.tsdb.max-block-duratinotallow=5m：該參數(shù)配置了生成塊（block）的最大時間間隔為5分鐘。如果一個塊的時間跨度超過這個參數(shù)所設的時間跨度，則這個塊將被分成多個子塊。
• --enable-feature=memory-snapshot-on-shutdown：該參數(shù)配置了在Prometheus關(guān)閉時，自動將當前內(nèi)存中的數(shù)據(jù)快照寫入到磁盤中，Prometheus在下次啟動時讀取該快照從而可以更快的完成啟動。

三、Pulsar 指標優(yōu)化

Pulsar的指標可以成功觀測之后，我們在日常的調(diào)優(yōu)和運維過程中發(fā)現(xiàn)了一些Pulsar指標本身存在的問題，這些問題包括準確性、用戶體驗、以及性能調(diào)優(yōu)等方面，我們針對這些問題做了一些優(yōu)化和改造，使得Pulsar更加通用、易維護。

3.1 Pulsar消費積壓指標

原生的Pulsar訂閱積壓指標單位是entry，從Kafka遷移到Pulsar的用戶希望Pulsar能夠和Kafka一樣，提供以消息條數(shù)為單位的積壓指標，這樣可以方便用戶判斷具體的延遲大小并盡量不改變用戶使用消息中間件的習慣。

在確保配置brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor情況下，Pulsar broker端在往bookkeeper端寫入entry前，通過攔截器往entry的頭部添加索引元數(shù)據(jù)，該索引在同一分區(qū)內(nèi)單調(diào)遞增，entry頭部元數(shù)據(jù)示例如下：

biz-log-partition-1 -l 24622961 -e 6
Batch Message ID: 24622961:6:0
Publish time: 1676917007607
Event time: 0
Broker entry metadata index: 157398560244
Properties:
"X-Pulsar-batch-size    2431"
"X-Pulsar-num-batch-message    50"

以分區(qū)為指標統(tǒng)計的最小單位，基于last add confirmed entry和last consumed entry計算兩個entry中的索引差值，即是訂閱在每個分區(qū)的數(shù)據(jù)積壓。下面是cursor基于訂閱位置計算訂閱積壓的示意圖，其中l(wèi)ast add confirmed entry在攔截器中有記錄最新索引，對于last consumed entry，cursor需要從bookkeeper中讀取，這個操作可能會涉及到bookkeeper讀盤，所以在收集延遲指標的時候可能會增加采集的耗時。

效果

上圖是新訂閱積壓指標和原生積壓指標的對比，新增的訂閱積壓指標單位是條，原生訂閱積壓指標單位是entry。在客戶端指定單條發(fā)送100w條消息時，訂閱積壓都有明顯的升高，當客戶端指定批次發(fā)送100w條消息的時候，新的訂閱積壓指標會有明顯的升高，而原生訂閱積壓指標相對升高幅度不大，所以新的訂閱積壓指標更具體的體現(xiàn)了訂閱積壓的情況。

3.2 Pulsar bundle指標

Pulsar相比于Kafka增加了自動負載均衡的能力，在Pulsar里topic分區(qū)是綁定在bundle上的，而負載均衡的最小單位是bundle，所以我們在調(diào)優(yōu)負載均衡策略和參數(shù)的時候比較依賴bunlde的流量分布指標，并且該指標也可以作為我們切分bundle的參考依據(jù)。我們在開發(fā)bundle指標的時候做了下面兩件事情：

統(tǒng)計當前Pulsar集群非游離狀態(tài)bundle的負載情況對于處于游離狀態(tài)的bundle（即沒有被分配到任何broker上的bundle），我們指定Pulsar leader在上報自身bundle指標的同時，上報這些處于游離狀態(tài)的bundle指標，并打上是否游離的標簽。

效果

上圖就是bundle的負載指標，除了出入流量分布的情況，我們還提供了生產(chǎn)者/消費者到bundle的連接數(shù)量，以便運維同學從更多角度來調(diào)優(yōu)負載均衡策略和參數(shù)。

3.3 kop消費延遲指標無法上報

在我們實際運維過程中，重啟kop的Coordinator節(jié)點后會偶發(fā)消費延遲指標下降或者掉0的問題，從druid查看上報的數(shù)據(jù)，我們發(fā)現(xiàn)在重啟broker之后消費組就沒有繼續(xù)上報kop消費延遲指標。

（1）原因分析

由于kop的消費延遲指標是由Kafka lag exporter采集的，所以我們重點分析了Kafka lag exporter采集消費延遲指標的邏輯，下圖是Kafka-lag-exporter采集消費延遲指標的示意圖：

其中，kafka-lag-exporter計算消費延遲指標的邏輯會依賴kop的describeConsumerGroups接口，但是當GroupCoordinator節(jié)點重啟后，該接口返回的member信息中assignment數(shù)據(jù)缺失，kafka-lag-exporter會將assignment為空的member給過濾掉，所以最終不會上報對應member下的分區(qū)指標，代碼調(diào)試如下圖所示：

為什么kop/Kafka describeConsumerGroups接口返回member的assignment是空的？因為consumer在啟動消費時會通過groupManager.storeGroup寫入__consumer_offset，在coordinator關(guān)閉時會轉(zhuǎn)移到另一個broker，但另一個broker并沒有把assignment字段反序列化出來（序列化為groupMetadataValue，反序列化為readGroupMessageValue），如下圖：

（2）解決方案

在GroupMetadataConstants#readGroup-MessageValue()方法對coordinator反序列化消費組元數(shù)據(jù)信息時，將assignment字段讀出來并設置（序列化為groupMetadataValue，反序列化為readGroupMessageValue），如下圖：

四、總結(jié)

在Pulsar監(jiān)控系統(tǒng)構(gòu)建的過程中，我們解決了與用戶體驗、運維效率、Pulsar可用性等方面相關(guān)的問題，加快了Pulsar在vivo的落地進度。雖然我們在構(gòu)建Pulsar可觀測系統(tǒng)過程中解決了一部分問題，但是監(jiān)控鏈路仍然存在單點瓶頸等問題，所以Pulsar在vivo的發(fā)展未來還會有很多挑戰(zhàn)。