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

一、背景

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

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

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

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

針對(duì)以上列出的幾個(gè)問題，我們?cè)谙旅娣謩e展開敘述。

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

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

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

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

基于以上監(jiān)控系統(tǒng)的設(shè)計(jì)邏輯，我們?cè)诰唧w實(shí)現(xiàn)的過程中遇到了幾個(gè)比較關(guān)鍵的問題：

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

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

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

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

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

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

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

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

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

3.1 Pulsar消費(fèi)積壓指標(biāo)

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

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

效果

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

3.2 Pulsar bundle指標(biāo)

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

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

效果

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

3.3 kop消費(fèi)延遲指標(biāo)無法上報(bào)

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

（1）原因分析

由于kop的消費(fèi)延遲指標(biāo)是由Kafka lag exporter采集的，所以我們重點(diǎn)分析了Kafka lag exporter采集消費(fèi)延遲指標(biāo)的邏輯，下圖是Kafka-lag-exporter采集消費(fèi)延遲指標(biāo)的示意圖：

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

為什么kop/Kafka describeConsumerGroups接口返回member的assignment是空的？因?yàn)閏onsumer在啟動(dòng)消費(fèi)時(shí)會(huì)通過groupManager.storeGroup寫入__consumer_offset，在coordinator關(guān)閉時(shí)會(huì)轉(zhuǎn)移到另一個(gè)broker，但另一個(gè)broker并沒有把a(bǔ)ssignment字段反序列化出來（序列化為groupMetadataValue，反序列化為readGroupMessageValue），如下圖：

（2）解決方案

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

四、總結(jié)

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