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

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

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

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

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

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

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

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

　　圖3 eventESSink Capability

 

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

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

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

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

　　圖5 paas alertESSink Policy

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

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

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

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

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

　　了解元數(shù)據(jù)驅(qū)動后，讓我們來看看可以通過哪些方法實現(xiàn) Flink 作業(yè)的而優(yōu)化和監(jiān)控。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

　　通過以上配置，可以限定每個 TaskManager 獨占 CPU 和內(nèi)存的資源，且不會多個作業(yè)搶占，實現(xiàn)作業(yè)之間的隔離。

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

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

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

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

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

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

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

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

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

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

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

　　圖9 Single-Threshold1 Policy （點擊查看大圖）

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

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

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

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

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

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

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