eBay 的監控平臺 Sherlock.IO 每天處理著(zhù)上百億條日志（log），事件（event）和指標（metric）。通過(guò)構建 Flink Streaming job 實(shí)時(shí)處理系統，監控團隊能夠及時(shí)將日志和事件的處理結果反饋給用戶(hù)。當前，監控團隊維護著(zhù) 8 個(gè) Flink 集群，最大的集群規模達到上千個(gè) TaskManager，總共運行著(zhù)上百個(gè)作業(yè)（job），一些作業(yè)已經(jīng)穩定運行了半年以上。

　　為了讓用戶(hù)和管理員能夠更加快捷地創(chuàng )建Flink作業(yè)并調整參數，監控團隊在 Flink 上搭建了一套元數據微服務(wù)（metadata service），該服務(wù)能夠用Json來(lái)描述一個(gè)作業(yè)的 DAG，且相同的 DAG 共用同一個(gè)作業(yè)，能夠更加方便地創(chuàng )建作業(yè)，無(wú)需調用 Flink API。Sherlock.IO 流處理整體的架構如圖1所示。

　　圖1 Sherlock.IO 流處理整體架構

　　目前，用這套元數據微服務(wù)創(chuàng )建的作業(yè)僅支持以 Kafka 作為數據源，只要數據接入到 Kafka，用戶(hù)就可以定義 Capability 來(lái)處理邏輯從而通過(guò) Flink Streaming 處理數據。

　　元數據微服務(wù)框架如圖 2 所示，最上層是元數據微服務(wù)提供的 Restful API, 用戶(hù)通過(guò)調用 API 來(lái)描述和提交作業(yè)。描述作業(yè)的元數據包含三個(gè)部分：Resource，Capability 和 Policy。Flink 適配器（Adaptor）連接了 Flink Streaming API 和元數據微服務(wù) API，且會(huì )根據元數據微服務(wù)描述的作業(yè)調用 Flink Streaming API 來(lái)創(chuàng )建作業(yè)，從而屏蔽 Flink StreamAPI。

　　因此，用戶(hù)不用了解 Flink Streaming API 就可以創(chuàng )建 Flink 作業(yè)。未來(lái)如果需要遷移到其他的流處理框架，只要增加一個(gè)適配器，就可以將現有的作業(yè)遷移到新的流處理框架上。

　　圖2 元數據微服務(wù)框架

　　Capability 定義了作業(yè)的 DAG 以及每個(gè)算子（Operator）所用的 Class，圖 3 是事件處理（eventProcess） Capability，它最終會(huì )生成如圖 4 的 DAG。事件處理 Capability 先從 Kafka 讀出數據，再寫(xiě)到 Elasticsearch 中。該 Capability 將該作業(yè)命名為“eventProcess”，并定義其并行度為“5”，其算子為“EventEsIndexSinkCapability”， 其數據流為“Source –> sink”。

　　圖3 eventESSink Capability

 

　　圖4 生成的Flink作業(yè)

　　每個(gè)命名空間（Namespace）需要定義一個(gè)或多個(gè) Policy，每個(gè) Policy 指定了相應的 Capability，即指定了用哪一套 DAG 來(lái)運行這個(gè) Policy。Policy 還定義了這個(gè)作業(yè)的相關(guān)配置，例如從哪個(gè) Kafka topic 中讀取數據，寫(xiě)到 ElasticSearch 的哪個(gè)索引（Index）中，中間是否要跳過(guò)某些算子等等。

　　其次，Policy 還能作為一個(gè)簡(jiǎn)易的過(guò)濾器（Filter），可以通過(guò)配置 Jexl 表達式過(guò)濾掉一些不需要的數據，提高作業(yè)的吞吐量。

　　另外，我們還實(shí)現了 Zookeeper 定時(shí)更新的機制，使得 Policy 修改后不再需要重啟作業(yè)，只要是在更新時(shí)間間隔內，該命名空間的 Policy 修改就會(huì )被自動(dòng)應用到作業(yè)上。圖 5 是命名空間為 paas 的 Policy 示例。

　　圖5 paas alertESSink Policy

　　Resource 定義了某個(gè)命名空間所需要的資源，比如 Flink 集群， Kafka broker，ES 集群等等。我們有多個(gè) Flink 集群和 ES 集群，通過(guò) Resource 配置，作業(yè)可以知道某個(gè)命名空間的日志應該寫(xiě)到哪個(gè) ES 集群，并可以判斷該命名空間的數據應該從哪個(gè) Kafka 集群讀取。

　　為了減少作業(yè)數量，我們可以讓相同的 DAG 復用同一個(gè)作業(yè)。我們先給不同的 Policy 指定相同的 Capability，在該 Capability 資源足夠的情況下，這些 Policy 就會(huì )被調度到同一個(gè)作業(yè)上。

　　以 SQL 的 Capability 為例，每個(gè) Policy 的 SQL 語(yǔ)句不盡相同，如果為每個(gè) Policy 都創(chuàng )建一個(gè)作業(yè)， Job Manager 的開(kāi)銷(xiāo)就會(huì )很大，且不好管理。因此，我們可以為 SQL Capability 配置 20 個(gè) Slot，每個(gè) Policy 占用一個(gè) Slot。那么該 Capability 生成的作業(yè)就可以運行 20 個(gè) Policy。

　　作業(yè)運行時(shí)，從 Source 讀進(jìn)來(lái)的數據會(huì )被打上相應 Policy 的標簽，并執行該 Policy 定義的 SQL 語(yǔ)句，從而實(shí)現不同 Policy 共享同一個(gè)作業(yè)，大大減少了作業(yè)的數量。

　　用共享作業(yè)還有一個(gè)好處：如果多個(gè)命名空間的數據在一個(gè) Kafka topic 里，那么只要讀一遍數據即可，不用每個(gè)命名空間都讀一次 topic 再過(guò)濾，這樣就大大提高了處理的效率。

　　了解元數據驅動(dòng)后，讓我們來(lái)看看可以通過(guò)哪些方法實(shí)現 Flink 作業(yè)的而優(yōu)化和監控。

　　在 Flink 集群的運維過(guò)程中，我們很難監控作業(yè)的運行情況。即使開(kāi)啟了檢查點(diǎn)（checkpoint），我們也無(wú)法確定是否丟失數據或丟失了多少數據。因此，我們?yōu)槊總�(gè)作業(yè)注入了 Heartbeat 以監控其運行情況。

　　Heartbeat 就像 Flink 中用來(lái)監控延遲的“LatencyMarker”一樣，它會(huì )流過(guò)每個(gè)作業(yè)的管道。但與 LatencyMarker 不同的是，當 Heartbeat 遇到 DAG 的分支時(shí)，它會(huì )分裂并流向每個(gè)分支，而不像 LatencyMarker 那樣隨機流向某一個(gè)分支。另一個(gè)不同點(diǎn)在于 Heartbeat 不是由 Flink 自身產(chǎn)生，而是由元數據微服務(wù)定時(shí)產(chǎn)生，而后由每個(gè)作業(yè)消費。

　　如圖 4 所示，每個(gè)作業(yè)在啟動(dòng)的時(shí)候會(huì )默認加一個(gè) Heartbeat 的數據源。Heartbeat 流入每個(gè)作業(yè)后，會(huì )隨數據流一起經(jīng)過(guò)每個(gè)節點(diǎn)，在每個(gè)節點(diǎn)上打上當前節點(diǎn)的標簽，然后跳過(guò)該節點(diǎn)的處理邏輯流向下個(gè)節點(diǎn)。直到 Heartbeat 流到最后一個(gè)節點(diǎn)時(shí)，它會(huì )以指標（Metric）的形式發(fā)送到 Sherlock.IO（eBay 監控平臺）。

　　該指標包含了 Heartbeat 產(chǎn)生的時(shí)間，流入作業(yè)的時(shí)間以及到達每個(gè)節點(diǎn)的時(shí)間。通過(guò)這個(gè)指標，我們可以判斷該作業(yè)在讀取 kafka 時(shí)是否延時(shí)，以及一條數據被整個(gè)管道處理所用的時(shí)間和每個(gè)節點(diǎn)處理數據所用的時(shí)間，進(jìn)而判斷該作業(yè)的性能瓶頸。

　　由于 Heartbeat 是定時(shí)發(fā)送的，因此每個(gè)作業(yè)收到的 Heartbeat 個(gè)數應該一致。若最后發(fā)出的指標個(gè)數與期望不一致，則可以進(jìn)一步判斷是否有數據丟失。

　　圖 6 描述了某 Flink 作業(yè)中的數據流以及 Heartbeat 的運行狀態(tài)：

　　圖6 Heartbeat在作業(yè)中的運行過(guò)程

　　有了 Heartbeat，我們就可以用來(lái)定義集群的可用性。首先，我們需要先定義在什么情況下屬于不可用的：

　　當內存不足（OutofMemory）或代碼運行錯誤時(shí)，作業(yè)就可能會(huì )意外重啟。我們認為重啟過(guò)程中造成的數據丟失是不可用的情況之一。因此我們的目標之一是讓 Flink 作業(yè)能夠長(cháng)時(shí)間穩定運行。

　　有時(shí)因為基礎設施的問(wèn)題導致物理機或者容器沒(méi)啟動(dòng)起來(lái)，或是在 Flink 作業(yè)發(fā)生重啟時(shí)由于 Slot 不夠而無(wú)法啟動(dòng)，或者是因為 Flink 作業(yè)的重啟次數已經(jīng)超過(guò)了最大重啟次數（rest.retry.max-attempts）, Flink 作業(yè)就會(huì )中止。此時(shí)需要人工干預才能將作業(yè)重新啟動(dòng)起來(lái)。

　　我們認為 Flink 作業(yè)中止時(shí)，也是不可用的情況之一。

　　發(fā)生這種情況，一般是因為遇到了反壓（BackPressure）。造成反壓的原因有很多種，比如上游的流量過(guò)大，或者是中間某個(gè)算子的處理能力不夠，或者是下游存儲節點(diǎn)遇到性能瓶頸等等。雖然短時(shí)間內的反壓不會(huì )造成數據丟失，但它會(huì )影響數據的實(shí)時(shí)性，最明顯的變化是延遲這個(gè)指標會(huì )變大。

　　我們認為反壓發(fā)生時(shí)是不可用的情況之一。

　　針對以上三種情況，我們都可以用 Heartbeat 來(lái)監控，并計算可用性。比如第一種情況，如果作業(yè)重啟時(shí)發(fā)生了數據丟失，那么相應的那段管道的 Heartbeat 也會(huì )丟失，從而我們可以監測出是否有數據丟失以及粗粒度地估算數據丟了多少。對于第二種情況，當作業(yè)中止時(shí)，HeartBeat 也不會(huì )被處理，因此可以很快發(fā)現作業(yè)停止運行并讓 on-call 及時(shí)干預。第三種情況當反壓發(fā)生時(shí)，HeartBeat 也會(huì )被阻塞在發(fā)生反壓的上游，因此 on-call 也可以很快地發(fā)現反壓發(fā)生并進(jìn)行人工干預。

　　綜上，Heartbeat 可以很快監測出 Flink 作業(yè)的運行情況。那么，如何評估可用性呢？由于 Heartbeat 是定時(shí)發(fā)生的，默認情況下我們設置每 10 秒發(fā)一次。1 分鐘內我們期望每個(gè)作業(yè)的每條管道能夠發(fā)出 6 個(gè)帶有作業(yè)信息的 heartbeat，那么每天就可以收到 8640 個(gè) Heartbeat。

　　因此，一個(gè)作業(yè)的可用性可以定義為：

　　Slot 是 Flink 運行作業(yè)的最小單位[1]，每個(gè) TaskManager 可以分配一個(gè)至多個(gè) Slot（一般分配的個(gè)數為該 TaskManager 的 CPU 數）。根據 Flink 作業(yè)的并行度，一個(gè)作業(yè)可以分配到多個(gè) TaskManager 上，而一個(gè) TaskManager 也可能運行著(zhù)多個(gè)作業(yè)。然而，一個(gè) TaskManager 就是一個(gè) JVM，當多個(gè)作業(yè)分配到一個(gè) TaskManager 上時(shí)，就會(huì )有搶奪資源的情況發(fā)生。

　　例如，我一個(gè) TaskManager 分配了 3 個(gè) Slot（3 個(gè) CPU）和 8G 堆內存。當 JobManager 調度作業(yè)的時(shí)候，有可能將 3 個(gè)不同作業(yè)的線(xiàn)程調度到該 TaskManager 上，那么這 3 個(gè)作業(yè)就會(huì )同時(shí)搶奪 CPU 和內存的資源。當其中一個(gè)作業(yè)特別耗 CPU 或內存的時(shí)候，就會(huì )影響其他兩個(gè)作業(yè)。

　　在這種情況下，我們通過(guò)配置 Flink 可以實(shí)現作業(yè)的隔離，如圖 7 所示：

　　圖7 Flink 作業(yè)隔離前后的調度圖

　　通過(guò)以上配置，可以限定每個(gè) TaskManager 獨占 CPU 和內存的資源，且不會(huì )多個(gè)作業(yè)搶占，實(shí)現作業(yè)之間的隔離。

　　我們運維 Flink 集群的時(shí)候發(fā)現，出現最多的問(wèn)題就是反壓。在 3.2 中提到過(guò)，發(fā)生反壓的原因有很多種，但無(wú)論什么原因，數據最終都會(huì )被積壓在發(fā)生反壓上游的算子的本地緩沖區（localBuffer）中。

　　我們知道，每一個(gè) TaskManager 有一個(gè)本地緩沖池， 每一個(gè)算子數據進(jìn)來(lái)后會(huì )把數據填充到本地緩沖池中，數據從這個(gè)算子出去后會(huì )回收這塊內存。當被反壓后，數據發(fā)不出去，本地緩沖池內存就無(wú)法釋放，導致一直請求緩沖區（requestBuffer）。

　　由于 Heartbeat 只能監控出是否發(fā)生了反壓，但無(wú)法定位到是哪個(gè)算子出了問(wèn)題，因此我們定時(shí)地將每個(gè)算子的 StackTrace 打印出來(lái)，當發(fā)生反壓時(shí)，通過(guò) StackTrace 就可以知道是哪個(gè)算子的瓶頸。

　　如圖8所示，我們可以清晰地看到發(fā)生反壓的 Flink 作業(yè)及其所在的 Taskmanager。再通過(guò) Thread Dump，我們就可以定位到代碼的問(wèn)題。

　　圖8 發(fā)生反壓的StackTrace （點(diǎn)擊觀(guān)看大圖）

　　Flink 本身提供了很多有用的指標[2]來(lái)監控 Flink 作業(yè)的運行情況，在此基礎上我們還加了一些業(yè)務(wù)上的指標。除此之外，我們還使用了以下工具監控 Flink 作業(yè)。

　　Flink 的 History server[3]可以查詢(xún)已完成作業(yè)的狀態(tài)和指標。比如一個(gè)作業(yè)的重啟次數、它運行的時(shí)間。我們常常用它找出運行不正常的作業(yè)。比如，我們可以通過(guò) History server 的 attempt 指標知道每個(gè)作業(yè)重啟的次數，從而快速去現場(chǎng)找到重啟的原因，避免下次再發(fā)生。

　　雖然 Flink 有 HA 的模式，但在極端情況下，例如整個(gè)集群出現問(wèn)題時(shí)，需要 on-call 即時(shí)發(fā)覺(jué)并人工干預。我們在元數據微服務(wù)中保存了最后一次提交作業(yè)成功的元數據，它記錄了在每個(gè) Flink 集群上應該運行哪些作業(yè)。守護線(xiàn)程（Daemon thread）會(huì )每分鐘去比較這個(gè)元數據和 Flink 上運行的作業(yè)，若發(fā)現 JobManager 連不通或者有作業(yè)運行不一致則立刻發(fā)出告警（Alert）通知 on-call。

　　下面介紹幾個(gè)已經(jīng)運行在監控系統上的 Flink 流處理系統的應用：

　　當前監控團隊是基于 Flink Streaming 做事件告警（Event alerting），我們定義了一個(gè)告警算子 EventAlertingCapability，該 Capability 可以處理每個(gè) Policy 自定義的規則。如圖 9 定義的一條性能監控規則：

　　該規則的含義是當性能檢測器的應用為“r1rover”， 主機以“r1rover”開(kāi)頭，且數值大于 90 時(shí)，就觸發(fā)告警。且生成的告警會(huì )發(fā)送到指定的 Kafka topic 中供下游繼續處理。

　　圖9 Single-Threshold1 Policy （點(diǎn)擊查看大圖）

　　Eventzon 就像 eBay 的事件中心，它收集了從各個(gè)應用，框架，基礎架構發(fā)過(guò)來(lái)的事件，最后通過(guò)監控團隊的 Flink Streaming 實(shí)時(shí)生成告警。由于各個(gè)事件的數據源不同，它們的元數據也不同，因此無(wú)法用一條統一的規則來(lái)描述它。

　　我們專(zhuān)門(mén)定義了一套作業(yè)來(lái)處理 Eventzon 的事件，它包含了多個(gè) Capability，比如 Filter Capability，用來(lái)過(guò)濾非法的或者不符合條件的事件； 又比如 Deduplicate Capability，可以用來(lái)去除重復的事件。Eventzon 的所有事件經(jīng)過(guò)一整套作業(yè)后，會(huì )生成有效的告警，并根據通知機制通過(guò) E-mail、Slack 或 Pagerduty 發(fā)給相關(guān)團隊。

　　Netmon 的全稱(chēng)為 Network Monitoring, 即網(wǎng)絡(luò )監控，它可以用來(lái)監控整個(gè) eBay 網(wǎng)絡(luò )設備的健康狀態(tài)。它的數據源來(lái)自 eBay 的交換機，路由器等網(wǎng)絡(luò )設備的日志。Netmon 的作用是根據這些日志找出一些特定的信息，往往是一些錯誤的日志，以此來(lái)生成告警。

　　eBay 的每一臺設備都要“登記造冊”，每臺設備將日志發(fā)過(guò)來(lái)后，我們通過(guò) EnrichCapability 從“冊子”中查詢(xún)這臺設備的信息，并把相關(guān)信息比如 IP 地址，所在的數據中心，所在的機架等填充到日志信息中作為事件保存。當設備產(chǎn)生一些特定的錯誤日志時(shí)， 它會(huì )被相應的規則匹配然后生成告警，該告警會(huì )被 EventProcess Capability 保存到 Elasticsearch 中實(shí)時(shí)顯示到 Netmon 的監控平臺（dashboard）上。有時(shí)因為網(wǎng)絡(luò )抖動(dòng)導致一些短暫的錯誤發(fā)生，但系統過(guò)一會(huì )兒就會(huì )自動(dòng)恢復。

　　當上述情況發(fā)生時(shí)，Netmon 會(huì )有相應的規則將發(fā)生在網(wǎng)絡(luò )抖動(dòng)時(shí)生成的告警標記為“已解決”（Resolved）。對于一些必須人工干預的告警，運維人員可以通過(guò)網(wǎng)絡(luò )監控平臺（Netmon dashboard）手動(dòng)點(diǎn)擊“已解決”，完成該告警的生命周期。

　　eBay 的監控團隊希望能根據用戶(hù)提供的指標、事件和日志以及相應的告警規則實(shí)時(shí)告警用戶(hù)。Flink Streaming 能夠提供低延時(shí)的處理從而能夠達到我們低延時(shí)的要求，并且它適合比較復雜的處理邏輯。

　　然而在運維 Flink 的過(guò)程中，我們也發(fā)現了由于作業(yè)重啟等原因導致誤報少報告警的情況發(fā)生，從而誤導客戶(hù)。因此今后我們會(huì )在 Flink 的穩定性和高可用性上投入更多。我們也希望在監控指標、日志上能夠集成一些復雜的 AI 算法，從而能夠生成更加有效精確的告警，成為運維人員的一把利器。