基于 Flink 構(gòu)建 CEP 引擎的挑戰(zhàn)和實踐

背景及現(xiàn)狀

奇安信集團作為一家網(wǎng)絡(luò)安全公司，專門為政府、企業(yè)，教育、金融等機構(gòu)和組織提供企業(yè)級網(wǎng)絡(luò)安全技術(shù)、產(chǎn)品和服務(wù)，奇安信的 NGSOC 產(chǎn)品的核心引擎是一個 CEP 引擎，用于實時檢測網(wǎng)絡(luò)攻擊，其技術(shù)演進過程如下圖所示。

2015 年開始使用基于 Esper 的 CEP 方案，但是當時遇到了很多問題，其中最顯著的是性能問題，因為 Esper 對于規(guī)則條目的支持數(shù)量不多，一般情況下超過幾十條就會受到嚴重影響；

2017 年奇安信的技術(shù)方案演進到了使用 C++ 實現(xiàn)的 Dolphin 1.0，其在單機上的性能表現(xiàn)大幅度提升；

2018 年奇安信決定將技術(shù)方案全面轉(zhuǎn)向基于 Flink 的 Sabre。

奇安信產(chǎn)品具體的應(yīng)用場景是企業(yè)系統(tǒng)的安全檢測和數(shù)據(jù)分析，其自下而上分為四個業(yè)務(wù)處理流程，分別是數(shù)據(jù)的采集、解析、處理和展示結(jié)果，這其中最核心的是第三層數(shù)據(jù)處理。該產(chǎn)品的用戶主要是安全規(guī)則團隊，其可以使用規(guī)則編輯器來對安全規(guī)則進行添加、刪除、編輯和查找操作，并可批量啟動/停用多個規(guī)則，同時可以將處于啟動狀態(tài)的有效規(guī)則統(tǒng)一發(fā)送給產(chǎn)品。

在數(shù)據(jù)規(guī)模方面，產(chǎn)品解決的不是一個或幾個大型數(shù)據(jù)集群的問題，而是數(shù)以百計的中小型數(shù)據(jù)集群的運維問題。在 B2B 領(lǐng)域，由于產(chǎn)品是直接部署到客戶方，很多客戶使用的是內(nèi)部隔離網(wǎng)，無法連接外網(wǎng)，且沒有專門人員負責集群的運維，這種情況下哪怕一個小升級都會耗費大量時間。因此，產(chǎn)品更多關(guān)注該領(lǐng)域下數(shù)據(jù)集群可運維性問題的解決。

奇安信在最初計劃使用 Flink 作為技術(shù)方案并進行調(diào)研的過程中，發(fā)現(xiàn)了其一系列的痛點問題。由于企業(yè)級硬件資源環(huán)境受限，規(guī)則集數(shù)量及種類不確定，使得 Flink 程序運行難以控制，并且現(xiàn)有的庫“Flink SQL”和“Flink CEP”均不能滿足其業(yè)務(wù)性能需求。具體的痛點如下：

**1.不能進行語義優(yōu)化、不便于動態(tài)更新規(guī)則。 ** 網(wǎng)絡(luò)安全事件井噴式發(fā)生的今天，安全需求迅速擴展。為了能夠在有限時間內(nèi)對特定語義的快速支持，關(guān)聯(lián)引擎的整體架構(gòu)必須異常靈活，才能適應(yīng)未來安全分析場景的各種需求，而基于開源關(guān)聯(lián)引擎實現(xiàn)的產(chǎn)品會在激烈的需求變化時遇到很多問題。

2.狀態(tài)監(jiān)控 & 高可用支持不足。

面向企業(yè)級的網(wǎng)絡(luò)安全監(jiān)測引擎具有一些特定需求，當前解決方案對此支持較差。

比如，現(xiàn)實情況中客戶對算子實例和 Taskmanager 概念較為模糊，真正關(guān)心的運行狀態(tài)的基本單位是規(guī)則。Flink 監(jiān)控頁面顯示的是算子實例及 Task manager 進程整體內(nèi)存的運行狀態(tài)，而在網(wǎng)絡(luò)安全監(jiān)控的業(yè)務(wù)場景中，對運行狀態(tài)和資源的監(jiān)控均需要細化到規(guī)則層面。

其次，在算子層面，F(xiàn)link 原生 Window 算子，沒有較好的資源（CPU / 內(nèi)存）保護機制，且存在大量重復告警，不符合網(wǎng)絡(luò)安全監(jiān)測領(lǐng)域的業(yè)務(wù)需求。

再次，F(xiàn)link 缺乏一些必要算子，例如不支持“不發(fā)生算子”。一個較為常見的應(yīng)用場景，某條規(guī)則指定在較長時間內(nèi)沒收到某臺服務(wù)器的系統(tǒng)日志，則認為此臺服務(wù)器發(fā)生了異常，需要及時通知用戶。

3.CEP 網(wǎng)絡(luò)負載高、CPU 利用率低。

和互聯(lián)網(wǎng)企業(yè)內(nèi)部使用的大型集群相比，奇安信面向的企業(yè)級應(yīng)用集群規(guī)模較小，硬件資源受限，且客戶的定制需求較多，導致安全監(jiān)測的規(guī)則要求更嚴格，引擎發(fā)布成本較高。但是，現(xiàn)有的 Flink 開源解決方案，或者需要根據(jù)業(yè)務(wù)需求進行改造，或者性能較差，均不能較好地解決上述問題。

首先，原生 Flink 只提供了函數(shù)式編程模式，即需要手動編寫復合特定業(yè)務(wù)需求的固定程序代碼，由此導致開發(fā)測試周期較長，不便于動態(tài)更新規(guī)則，可復用性較弱，且不能從全局語義層面進行優(yōu)化，性能較差。

其次，F(xiàn)link-CEP 僅是一個受限的序列算子，在運行時需要將所有數(shù)據(jù)傳輸?shù)?CEP 算子，然后在 CEP 算子中串行執(zhí)行各個條件語句。這種匯集到單點的運行模式，較多的冗余數(shù)據(jù)需要執(zhí)行條件匹配，由此產(chǎn)生了不必要的網(wǎng)絡(luò)負載，而且降低了 CPU 利用率。

再次，還存在一些非官方開源的輕量級 CEP 引擎，比如 Flink-siddhi，功能簡單，不是一個完整的解決方案。

其他的痛點問題還包括不支持空值窗口出發(fā)、以及聚合不保存原始數(shù)據(jù)等。

為了解決上述問題，奇安信在 Flink 的基礎(chǔ)上推出了一種全新的 CEP 引擎， Sabre。其整體架構(gòu)如下圖所示，其中包含三大核心模塊，左側(cè)是配置端，中間是 Sabre-server，右側(cè)是 Sabre 運行端。核心數(shù)據(jù)流存在兩條主線，紅線表示規(guī)則的提交、編譯、發(fā)布和運行流程。綠線表示狀態(tài)監(jiān)控的生成、收集、統(tǒng)計和展示流程。如圖所示，此架構(gòu)與 Hive 極為相似，是一種通用的大數(shù)據(jù) OLAP 系統(tǒng)架構(gòu)。下面詳細介紹三大核心模塊和兩大核心數(shù)據(jù)流。

首先，通過規(guī)則配置端創(chuàng)建規(guī)則，采用性能保護配置端修改性能保護策略；

然后，將任務(wù)所屬的規(guī)則文件和性能保護策略文件一并推送到 Sabre-server 提供的 REST 接口，該接口會調(diào)用文件解析及優(yōu)化方法構(gòu)建規(guī)則有向無環(huán)圖。

接著，執(zhí)行詞法語法分析方法，將規(guī)則有向無環(huán)圖中各個節(jié)點的 EPL 轉(zhuǎn)換為與其對應(yīng)的 AST（AbstractSyntax Tree，抽象語法樹），再將 AST 翻譯為任務(wù) java 代碼。

最后，調(diào)用 maven 命令打包 java 代碼為任務(wù) jar 包，并將任務(wù) jar 包及基礎(chǔ)運行庫一并提交到 Flink-on-YARN 集群。

Flink 有多種運行模式（例如 standalone Flink cluster、Flink cluster on YARN、Flink job on YARN 等），Sabre 采用了“Flink job on YARN”模式，在奇安信 NGSOC 應(yīng)用的特定場景下，采用 YARN 可統(tǒng)一維護硬件資源，并且使用 Flink job on YARN 可與 Hadoop 平臺進行無縫對接，以此很好的實現(xiàn)了任務(wù)間資源隔離。

在 Sabre 任務(wù)執(zhí)行過程中，Kafka 數(shù)據(jù)源向引擎提供原始事件。引擎處理結(jié)果分為回注事件和告警事件兩類。告警事件會輸出到目的 Kafka，供下級應(yīng)用消費?；刈⑹录硎疽粭l規(guī)則的處理結(jié)果可直接回注到下級規(guī)則，作為下級規(guī)則的數(shù)據(jù)源事件，由此可實現(xiàn)規(guī)則的相互引用。

綠線流程表示任務(wù)執(zhí)行過程中會定時輸出節(jié)點的運行監(jiān)控消息到 Sabre-server 的監(jiān)控消息緩存器，然后監(jiān)控消息統(tǒng)計器再匯總各個規(guī)則實例的運行監(jiān)控消息，統(tǒng)計為整條規(guī)則的運行監(jiān)控狀態(tài)，最后通過 Sabre-server 提供的 REST 接口推送給規(guī)則監(jiān)控端。

技術(shù)架構(gòu)

Sabre 的組件依賴與版本兼容情況如下圖所示。

大多數(shù)情況下，奇安信會以黑盒的方式發(fā)布產(chǎn)品，但是如果用戶方已經(jīng)部署大數(shù)據(jù)處理平臺，則產(chǎn)品會以 APP 的方式提供使用。

由于客戶規(guī)模較大，項目種類較多，部署環(huán)境較為復雜，或者存在多種 Yarn 集群版本，或者 Sabre 需作為單一 Flink 應(yīng)用發(fā)布到客戶已部署的 Flink 集群。

如何節(jié)省成本及提高實施效率，快速適配上述復雜的部署環(huán)境是個亟需解決的問題，為此 Sabre 的設(shè)計原則是僅采用 Flink 的分布式計算能力，業(yè)務(wù)代碼盡可能減少對 API 層的依賴，以便于兼容多種 Flink 版本。

如圖所示，Deploy、Core、APIs、Libraries 四層是大家熟知的 Flink 基本的組件棧。Sabre 對 API 層的依賴降到了最低，只引用了 DataStream、KeyedStream 和 SplitStream 三種數(shù)據(jù)流 API。函數(shù)依賴只包括 DataStream 的 assignTimestamps、flatMap、union、keyBy、split、process、addSink 等函數(shù)，KeyedStream 最基礎(chǔ)的 process 函數(shù)，以及 SplitStream 的 select 函數(shù)。由于依賴的 Flink API 較少，Sabre 可以很容易適配到各個 Flink 版本，從而具有良好的 Flink 版本兼容性。

在算子方面，Sabre 對 Flink 進行了一系列的重構(gòu)，下圖展示了這 Flink 和 Sabre 這二者之間的對比關(guān)系，其中主要包含三列，即 Flink 原生算子、Sabre 算子和兩者之間的比較結(jié)果。比較結(jié)果主要有四種情況，相同（Same）、實現(xiàn)（Implement）、優(yōu)化(優(yōu)化)和新增（New）。Sabre 共有 13 種完全自研的核心算子，其中 Datasource、CustomKafkaSink 和 CustomDatabase 按照 Flink 接口要求做了具體實現(xiàn)，F(xiàn)ilter、Key、Join 和 Aggregation 按照 Flink 原有算子的語義做了重新實現(xiàn)，CustomWindow 和 Sequence 在 Flink 原有算子語義的基礎(chǔ)上做了優(yōu)化實現(xiàn)。

下圖展示了 Sabre 的規(guī)則與 EPL 設(shè)計。序列 Sequence、聚合 Aggregation、不發(fā)生 NotOccur、流式機器學習 StreamML 和連接 Join 均屬于 Window 執(zhí)行時間包含的計算性算子。藍色虛線表示引用動態(tài)數(shù)據(jù)（Dynamic data），紫色虛線表示 Filter 無須經(jīng)過 Window 可直連輸出組件。

Window 算子

眾所周知，Join 和 Aggregation 的時間范圍由 Window 限定，而 Flink 原有 Window 算子不適合網(wǎng)絡(luò)安全監(jiān)測需求，為此 Sabre 設(shè)計了一種“自定義 Window 算子”，且重新實現(xiàn)了與“自定義 Window 算子”相匹配的 Join 和 Aggregation 算子。全新的 Window 具有以下六個主要特點：

實時觸發(fā)、即刻匹配：其目的是為了滿足自動化實時響應(yīng)的需求，一旦告警發(fā)出，會及時觸發(fā)響應(yīng)；

匹配不重復：重復告警對于規(guī)則引擎來講是一個常見問題，大量重復告警會增加安全人員的工作量，而該算子會將整個窗口與告警相關(guān)的事件全部清空，以此減少重復告警的數(shù)量；

糾正亂序：將 Window 窗口以特定單位為邊界切成一個個的時間槽，一旦發(fā)現(xiàn)亂序情況，插入亂序事件時可直接定位時間槽，基于流式狀態(tài)機進行局部計算，并且窗口事件超時，同步更新計算性算子的值，并入 count 算子，刪除超時事件的同時，同步減少 count 值；

實時資源和狀態(tài)監(jiān)控：由于 Window 對與內(nèi)存和 CPU 的影響比較大，因此需要對該類資源進行特別監(jiān)控以及保護；

流量控制：主要是為了更好地保護下級應(yīng)用。

Sequence 序列算子

Sabre 用 EPL 對 Flink CEP 實現(xiàn)的序列算子進行了重新設(shè)計，左邊是 Flink CEP 官方代碼展示，采用程序代碼的方式拼湊“NFA 自動機”。右邊是 Sabre 中 Sequence 算子的實現(xiàn)方式，其中包含了三個不同的 filter,通過正則表達式的使用來提升其表達的能力，并且，Sabre 將 filter 前置，無效事件不會傳輸?shù)?window 算子，從而較少了不必要的網(wǎng)絡(luò)負載。并且，只有較少的有效數(shù)據(jù)需要執(zhí)行正則匹配，降低了 CPU 利用率（filter 可以并行）。

NotOccur 不發(fā)生算子

NotOccur 是 Sabre 在 Flink 基礎(chǔ)上新增的一個算子，支持空事件觸發(fā)。

Trigger 全局算子

Sabre 還實現(xiàn)了一種針對窗口的全局觸發(fā)器 Trigger，Trigger 能夠?qū)⒍鄠€子計算性算子組合為復雜表達式，并實現(xiàn)了具有 GroupBy/Distinct 功能的 Key 算子以適配此 Trigger 算子。

Dynamic Data

Dynamicdata 可以映射為數(shù)據(jù)庫中的一個表，但是對這個表要進行特別的優(yōu)化，具體來講，如果一個事件的 IP 在威脅情報列表中，而這個威脅情報有可能比較長，比如十幾萬行甚至更長，這種情況下需要對該表數(shù)據(jù)結(jié)構(gòu)進行優(yōu)化以提升效率。Dynamic data 可以在其他算子中使用，如 Filter、Join 等。

流式統(tǒng)計與機器學習 StreamML

機器學習在網(wǎng)絡(luò)異常檢測上已經(jīng)越來越重要，為適應(yīng)實時檢測的需求，Sabre 沒有使用 Flink MachineLearning，而是引入了自研的流式機器學習算子 StreamML。

Flink MachineLearning 是一種基于批模式 DataSetApi 實現(xiàn)的機器學習函數(shù)庫，而 StreamML 是一種流式的機器學習算子，其目的是為了滿足網(wǎng)絡(luò)安全監(jiān)測的特定需求。與阿里巴巴開源的 Alink 相比，StreamML 允許機器學習算法工程師通過配置規(guī)則的方式即可快速驗證算法模型，無需編寫任何程序代碼。并且，流式機器學習算子 StreamML 實現(xiàn)了“模型訓練/更新”與“模型使用”統(tǒng)一的理念。其核心功能是通過算法、技術(shù)及模型實現(xiàn)數(shù)據(jù)訓練及對新數(shù)據(jù)檢測。該流式機器學習算子 StreamML 引入的輸入有三類，分別是：事件流、檢測對象和對象屬性；輸出也包含三類，分別是：事件、告警和預警。

流式機器學習算子 StreamML 的組件棧包含三部分，從下往上依次為：機器學習方法、應(yīng)用場景和產(chǎn)品業(yè)務(wù)。通過基本的機器學習算法（比如：統(tǒng)計學習算法、序列分析算法、聚類分析算法），流式機器學習算子 StreamML 可滿足具體特定的安全監(jiān)測應(yīng)用場景（比如：行為特征異常檢測、時間序列異常檢測、群組聚類分析），進而為用戶提供可理解的產(chǎn)品業(yè)務(wù)（比如：基線、用戶及實體行為分析 UEBA）。

行為特征異常檢測：根據(jù)采集的樣本數(shù)據(jù)（長時間）對統(tǒng)計分析對象建立行為基線，并以此基線為準，檢測發(fā)現(xiàn)偏離正常行為模式的行為。例如：該用戶通常從哪里發(fā)起連接？哪個運營商？哪個國家？哪個地區(qū)？這個用戶行為異常在組織內(nèi)是否為常見異常？

時間序列異常檢測：根據(jù)某一個或多個統(tǒng)計屬性，判斷按時間順序排列的數(shù)值序列是否異常，由此通過監(jiān)測指標變化來發(fā)現(xiàn)安全事件。例如：監(jiān)測某網(wǎng)站每小時的訪問量以防止 DDOS 攻擊；建模每個賬號傳輸文件大小的平均值，檢測出傳輸文件大小的平均值離群的賬號。

群組聚類分析：對數(shù)據(jù)的特征屬性間潛在相關(guān)性進行挖掘，將具有類似特征值的數(shù)據(jù)進行分組聚類。例如：該用戶是否擁有任何特殊特征？可執(zhí)行權(quán)限/特權(quán)用戶？基于執(zhí)行的操作命令和可訪問的實體，來識別IT管理員、DBA 和其它高權(quán)限用戶。

因為采用了 Flink 作為底層運行組件，所以 Sabre 具有與 Flink 等同的執(zhí)行性能。并且，針對網(wǎng)絡(luò)安全監(jiān)測領(lǐng)域的特定需求，Sabre 還在以下方面進行了性能優(yōu)化：

全局組件（數(shù)據(jù)源、動態(tài)表）引用優(yōu)化。由于 Kafka 類型的數(shù)據(jù)源 topic 有限，而規(guī)則數(shù)量可動態(tài)擴展，導致多個規(guī)則會有極大概率共用同一個數(shù)據(jù)源，根據(jù) EPL 語義等價原則合并相同的數(shù)據(jù)源，進而可以減少數(shù)據(jù)輸入總量及線程總數(shù)。

全新的匹配引擎。序列 Sequence 算子采用了新穎的流式狀態(tài)機引擎，復用了狀態(tài)機緩存的狀態(tài)，提升了匹配速度。類似優(yōu)化還包含大規(guī)模 IP 匹配引擎和大規(guī)模串匹配引擎。在流量、日志中存在大規(guī)模 IP 和字符串匹配需求，通過 IP 匹配引擎和大規(guī)模串匹配引擎進行優(yōu)化以提高效率。

表計算表達式優(yōu)化。對于規(guī)則中引用的動態(tài)表，會根據(jù)表達式的具體特性構(gòu)建其對應(yīng)的最優(yōu)計算數(shù)據(jù)結(jié)構(gòu)，以避免掃描全表數(shù)據(jù)，進而確保了執(zhí)行的時間復雜度為常量值。

自定義流式 Window 算子。采用“時間槽”技術(shù)實現(xiàn)了亂序糾正功能，并具有可以實時輸出無重復、無遺漏告警的特性。

圖上字段自動推導，優(yōu)化事件結(jié)構(gòu)。根據(jù)規(guī)則前后邏輯關(guān)系，推導出規(guī)則中標注使用的原始日志相關(guān)字段，無須輸出所有字段，以此優(yōu)化輸出事件結(jié)構(gòu)，減少了輸出事件大小。

圖上數(shù)據(jù)分區(qū)自動推導，優(yōu)化流拓撲。由于特定的功能需要，Window 往往會緩存大量數(shù)據(jù)，以致消耗較多內(nèi)存。通過對全局窗口 Hash 優(yōu)化，避免所有全局窗口都分配到同一個 Taskmanager 進程，由此提高了引擎整體內(nèi)存的利用率。

上圖是 Sabre 流式狀態(tài)機引擎的表示，其主要負責的功能是序列匹配。圖中左邊是標準的正則引擎，通常的流程可以從 Pattern 到語法樹到 NFA 再到 DFA，也可以從 Paterrn 直接到 NFA；圖左下側(cè)是一個正則表達式的 NFA 表示，右側(cè)是該正則表達式的 DFA 表示，使用該 DFA 的時候進行了改進（如圖中綠色線）。其目的是為了在出現(xiàn)亂序的時候提升處理性能，在亂序發(fā)生在正則表達式后半段的時候，該改進對于性能提升的效果最好。

大規(guī)模正則引擎主要使用了兩種互補的方法（圖上半側(cè)和下半側(cè)）。在將 NFA 轉(zhuǎn)向 DFA 的時候，很多情況下是不成功的，這種情況下往往會生成 DFA 的半成品，稱為Unfinished-DFA，第一種方法屬于混合狀態(tài)自動機，包含 NFA 和 DFA，其適用于Pattern 量少于 1000 的情況。而第二種方法適用于 Pattern 量大于 1000 甚至上萬的情況，該方法中首先需要尋找錨點，再做匹配，以降低整體的時間復雜度。這兩種方法相結(jié)合能夠較好地解決大規(guī)模正則匹配的問題。

產(chǎn)品運維

多級規(guī)則

多級規(guī)則是產(chǎn)品運維的一個顯著特點。如下圖所示，為滿足復雜場景需求，一種規(guī)則的輸出可直接作為另一種規(guī)則的輸入。通過這種規(guī)則拆分的方式，能分層構(gòu)造較為復雜的“多級規(guī)則”。如：圖中的“暴力探測”規(guī)則結(jié)果可以直接回注到下面的“登陸成功 ”規(guī)則，而無須額外的通信組件，由此實現(xiàn)更為復雜的“暴力破解”規(guī)則。

服務(wù)化/多租戶/資源監(jiān)控

產(chǎn)品采用微服務(wù)架構(gòu)，使用多租戶、多任務(wù)來滿足多個規(guī)則引擎的使用場景，同時對資源進行了實時監(jiān)控來保證系統(tǒng)的穩(wěn)定運行。

規(guī)則級的狀態(tài)/資源監(jiān)控

規(guī)則級的狀態(tài)和資源監(jiān)控是非常重要的產(chǎn)品需求，產(chǎn)品采用分布式監(jiān)控，提供三級分布式監(jiān)控能力（用戶、任務(wù)和規(guī)則），并支持吞吐量、EPS、CPU 和內(nèi)存的監(jiān)控。

整體系統(tǒng)保護

整體系統(tǒng)保護主要涉及兩方面，即流量控制和自我保護。

流量控制：為了增強 Sabre 引擎的健壯性，避免因規(guī)則配置錯誤，導致生成大量無效告警，在輸出端做了流量控制，以更好地保護下級應(yīng)用。當下級抗壓能力較弱時（例如數(shù)據(jù)庫），整個系統(tǒng)會做輸出降級。

自我保護：跑在 JVM 上的程序，經(jīng)常會遇到由于長時間 Full GC 導致 OOM 的錯誤，并且此時 CPU 占用率往往非常高，F(xiàn)link 同樣存在上述問題。自我保護功能采用了同時兼顧“Window隸屬規(guī)則的優(yōu)先級”及“Window引用規(guī)則數(shù)量”兩個條件的加權(quán)算法，以此根據(jù)全局規(guī)則語義實現(xiàn)自動推導 Window 優(yōu)先級，并根據(jù)此優(yōu)先級確定各個 Window 的自我保護順序。實時監(jiān)控 CPU 及內(nèi)存占用，當超過一定閾值時，智能優(yōu)化事件分布，以防出現(xiàn) CPU 長期過高或內(nèi)存使用率過大而導致的 OOM 問題。

未來發(fā)展與思考