2019-11-30

基于 Flink 构建关联分析引擎的挑战和实践

如何构建流式关联分析引擎？

本文转载自奇安信官网，作者：奇安信集团高级研发总监韩鹏

原文地址：https://www.aqniu.com/tools-tech/59894.html

随着云计算、大数据等新一代IT技术在各行业的深入应用，政企机构IT规模和复杂程度不断提高，网络流量、日志等各类数据规模大幅提升。与此同时，网络攻防日益激烈，网络安全威胁逐渐凸显出来，这对于SOC/SIEM产品的性能提出了一个很大的挑战。因此，奇安信独立研发了国内首款流式分布式关联分析引擎Sabre，搭载于公司旗下态势感知与安全运营平台（下文简称NGSOC），从而大幅提升NGSOC的数据分析能力和网络安全检测能力。

本文将从技术研发的角度，全面阐述Sabre的由来。

1.Sabre是什么？

Sabre是奇安信研发的新一代流式分布式关联分析引擎，是CEP（Complex Event Processing，复杂事件处理）技术在大数据领域的一个具体实现。奇安信研发关联引擎已有数年历史，中间经历了三次主要的技术演进，在2015年之前，奇安信使用的是基于开源CEP软件Esper研发的关联引擎，由于一些架构和设计上的问题，整体性能不是非常理想，也不支持多机扩展；在2016-2017年，用C++开发了一个高性能引擎，代号Dolphin，可以在单机上实现很高的性能；在2018年，从技术上全面转向Flink框架，极大增强了系统的可扩展性，推出了Sabre引擎作为NGSOC的核心检测引擎。

Sabre应用于奇安信的态势感知与安全运营平台（NGSOC）产品中，NGSOC主要服务于中大型政企客户，目前已经成功应用于200+大型政企机构，在国内安全管理平台市场占有率第一，其中搭载的Sabre引擎提供了核心的安全检测能力。和很多互联网公司内部自建数据处理平台不同的是，Sabre更注重的是技术的工程化交付，因此在设计和实现上和一般基于Flink的业务系统相比会有较大差异。

2.为什么要开发Sabre？

随着网络应用规模和复杂度的不断提高，网络中传输的数据量急剧上升，网络攻防对抗日趋激烈，企业内部新的安全问题开始显现，实时关联分析引擎，作为NGSOC检测体系中的核心组件，也遇到了越来越多的挑战：

(1) 性能优化问题。主要针对随着新型攻击的不断出现，关联分析规则规模不断上升导致的性能问题。传统开源关联引擎往往加载几十条规则即达到了性能瓶颈，而NGSOC的应用场景中，关联引擎需要支撑规模上千的关联规则。在有限的硬件资源条件下，如何避免系统整体性能随规则条数上升而发生线性下降，成为关联引擎的一个主要挑战。

(2) 规则的语义扩展问题。在网络安全事件井喷式发生的今天，安全需求迅速扩展。为了能够在有限时间内对特定语义的快速支持，关联引擎的整体架构必须异常灵活，才能适应未来安全分析场景的各种需求，而基于开源关联引擎实现的产品会在激烈的需求变化时遇到很多问题。

(3) 系统扩展性问题。主要指分布式环境下节点的扩展。随着企业网络流量和业务资产的不断扩容，NGSOC的系统处理能力必须能随企业业务规模的不断扩张而动态扩展。未来的分布式关联分析引擎需要支持数百节点的规模，以能够与现有的大数据平台无缝集成。

与Storm、Spark Streaming等流式计算框架相比，Flink具有编程接口丰富、自带多种Window算子、支持Exactly-Once、高性能分布式检查点、批流计算模式统一等优点。且Flink发展较为迅速，开源社区极为活跃，是目前最具发展潜力的流式计算框架，是未来实时计算执牛耳者。由于Flink为事件驱动的实时关联分析引擎在底层框架上提供了有力支持，因此奇安信的下一代关联分析引擎Sabre是基于Flink流式计算框架实现的。

在选择了Flink之后，发现Flink开源方案直接应用于安全检测领域，仍有很大的技术障碍。

和互联网企业内部使用的大型集群相比，NGSOC面向的企业级应用集群规模较小，硬件资源受限，且客户的定制需求较多，导致安全监测的规则要求更严格，引擎发布成本较高。但是，现有的Flink开源解决方案，或者需要根据业务需求进行改造，或者性能较差，均不能较好地解决上述问题。首先，原生Flink只提供了函数式编程模式，即需要直接编写复合特定业务需求的固定程序代码，由此导致开发测试周期较长，不便于动态更新规则，可复用性较弱，且不能从全局语义层面进行优化，性能较差。其次，Flink-CEP仅是一个受限的序列算子，在运行时需要将所有数据传输到CEP算子，然后在CEP算子中串行执行各个条件语句。这种汇集到单点的运行模式，较多的冗余数据需要执行条件匹配，产生了不必要的网络负载，而且降低了CPU利用率。再次，还存在一些非官方开源的轻量级CEP引擎，比如Flink-siddhi，功能简单，不是一个完整的解决方案。

面向企业级的网络安全监测引擎具有一些特定需求，当前解决方案对此支持较差。比如，现实情况，客户对算子实例和Taskmanager概念较为模糊，真正关心的运行状态的基本单位是规则。而Flink监控页面显示的是算子实例及Taskmanager进程整体内存的运行状态，而在网络安全监控的业务场景中，对运行状态和资源的监控均需要细化到规则层面。其次，在算子层面，Flink原生Window算子，没有较好的资源（CPU/内存）保护机制，且存在大量重复告警，不符合网络安全监测领域的业务需求。再次，Flink缺乏一些必要算子，例如不支持“不发生算子”。一个较为常见的应用场景，某条规则指定在较长时间内没收到某台服务器的系统日志，则认为此台服务器发生了异常，需要及时通知用户。

综上所述，现有解决方案应用于网络安全监测领域均会遇到问题，由此奇安信集团基于Flink构建了一种全新的CEP引擎。

3.Sabre如何处理数据？

上图为NGSOC的数据处理架构图，展示了整个系统的数据流。自下而上，NGSOC的数据处理过程由四部分组成，其核心是由“流式分布式关联分析引擎Sabre”构成的数据处理层PROCESS，且Sabre运行的硬件环境是由多个节点组成的分布式集群。右侧的规则配置管理模块供专业的安全人员使用，可通过类Visio图的界面较为友好便捷地配置规则；规则管理模块具有添加、删除、编辑和查找规则的功能，并可批量启动/停用多个规则，规则管理模块会将处于启动状态的有效规则统一发送给Sabre引擎。
最上方的绿色部分为结果处理层RESULT，Sabre会将处理结果“告警”或“关联事件”发送给下级响应模块，实现响应联动、分析调查及追踪溯源等功能。最下方的蓝色部分为日志采集层COLLECT，主要有“网络流量日志采集器”、“设备及系统日志采集器”和“其他类型的日志采集器（比如：防火墙、入侵检测系统IDS、入侵防护系统IPS、高级威胁监测系统APT等等）”三大类。中间部分为日志解析层PARSE，网络流量日志和系统安全日志格式多种多样，须将上述两类原始日志数据格式化，而其他类型的日志（比如：威胁情报、漏洞、资产）本身即为格式化数据，最终所有格式化数据均需统一存储到高性能消息队列Kafka。

4.Sabre的关键技术

(1) 系统架构

上图为Sabre系统整体架构图。Sabre整体架构包含三大核心模块，中间是Sabre-server，左侧是配置端，右侧是Sabre运行端。核心数据流存在两条主线，红线表示规则的提交、编译、发布和运行流程。绿线表示状态监控的生成、收集、统计和展示流程。如图所示，此架构与Hive极为相似，是一种通用的大数据OLAP系统架构。下面详细介绍三大核心模块和两大核心数据流。

首先，通过规则配置端创建规则，采用性能保护配置端修改性能保护策略，然后将任务所属的规则文件和性能保护策略文件一并推送到Sabre-server提供的REST接口，该接口会调用文件解析及优化方法构建规则有向无环图。接着执行词法语法分析方法，将规则有向无环图中各个节点的EPL转换为与其对应的AST（Abstract Syntax Tree，抽象语法树），再将AST翻译为任务java代码。最后调用maven命令打包java代码为任务jar包，并将任务jar包及基础运行库一并提交到Flink-on-YARN集群。

Flink有多种运行模式（例如 standalone Flink cluster、Flink cluster on YARN、Flink job on YARN等），Sabre采用了“Flink job on YARN”模式，在奇安信NGSOC应用的特定场景下，采用YARN可统一维护硬件资源，并且使用 Flink job on YARN 可与Hadoop平台进行无缝对接，以此实现了很好的任务间资源隔离。

在Sabre任务执行过程中，Kafka数据源向引擎提供原始事件。引擎处理结果分为回注事件和告警事件两类。告警事件会直接输出到目的Kafka，供下级应用消费。回注事件表示一条规则的处理结果可直接回注到下级规则，作为下级规则的数据源事件，由此可实现规则的相互引用。

绿线流程表示任务执行过程中会定时输出节点的运行监控消息到Sabre-server的监控消息缓存器，然后监控消息统计器再汇总各个规则实例的运行监控消息，统计为整条规则的运行监控状态，最后通过Sabre-server提供的REST接口推送给规则监控端。

(2) 功能设计

算子的设计和实现是构建CEP的重要组成部分。

上图展示了Flink和Sabre算子的比较关系。包含三列：Flink原生算子、Sabre算子、两者之间的比较结果（相同、实现、优化、新增）。Sabre共有13种完全自研的核心算子，其中Datasource、CustomKafkaSink和CustomDatabase按照 Flink接口要求做了具体实现，Filter、Key、Join和Aggregation按照Flink原有算子的语义做了重新实现，CustomWindow和Sequence在Flink原有算子语义的基础上做了优化实现。

由于Flink原有FilterFunction算子只能简单返回布尔值，以致输出结果的控制能力较差，而重新实现的Filter算子可同时执行多种业务逻辑，将一个“原始事件”输出一个或多个“处理事件”。Sabre还实现了一种针对窗口的全局触发器Trigger，Trigger能够将多个子计算性算子组合为复杂表达式，并实现了具有GroupBy/Distinct功能的Key算子以适配此Trigger算子。众所周知，Join和Aggregation的时间范围由Window限定，而Flink原有Window算子不适合网络安全监测需求，为此Sabre设计了一种“自定义Window算子”，且重新实现了与“自定义Window算子”相匹配的Join和Aggregation算子。

上图展示了Sabre算子间的关联关系。序列Sequence、聚合Aggregation、不发生NotOccur、流式机器学习StreamML和连接Join均属于Window执行时间包含的计算性算子。蓝色虚线表示引用动态数据，紫色虚线表示Filter无须经过Window可直连输出组件。

如上图所示，为满足复杂场景需求，一种规则的输出可直接作为另一种规则的输入。通过这种规则拆分的方式，能分层构造较为复杂的“多级规则”。如：上面的“暴力探测”规则结果可以直接回注到下面的“登陆成功 ”规则，而无须额外的通信组件。

(3) 性能优化

因为采用了Flink作为底层运行组件，所以Sabre具有与Flink等同的执行性能。并且，针对网络安全监测领域的特定需求，还做了如下的性能优化工作：

1）全局组件（数据源、动态表）引用优化。由于Kafka类型的数据源topic有限，而规则数量可动态扩展，导致多个规则会有极大概率共用同一个数据源，根据EPL语义等价原则合并相同的数据源，进而可以减少数据输入总量及线程总数。

2）全新的匹配引擎。序列Sequence算子采用了新颖的流式状态机引擎，复用了状态机缓存的状态，提升了匹配速度。类似优化还包含大规模IP匹配引擎和大规模串匹配引擎。

3）表计算表达式优化。对于规则中引用的动态表，会根据表达式的具体特性构建其对应的最优计算数据结构，以避免扫描全表数据，进而确保了执行的时间复杂度为常量值。

4）自定义流式Window算子。采用“时间槽”技术实现了乱序纠正功能，并具有可以实时输出无重复、无遗漏告警的特性。

5）字段自动推导，优化事件结构。根据规则前后逻辑关系，推导出规则中标注使用的原始日志相关字段，无须输出所有字段，以此优化输出事件结构，减少输出事件大小。

6）数据分区自动推导，优化流拓扑。由于功能需要，Window往往会缓存大量数据，以致消耗较多内存。通过对全局窗口Hash优化，避免所有全局窗口都分配到同一个Taskmanager进程，由此提高了引擎整体内存的利用率。

(4) 机器学习

机器学习在网络异常检测上已经越来越重要，为适应实时检测的需求，Sabre没有使用Flink MachineLearning，而是引入了自研的流式机器学习算子StreamML。Flink MachineLearning是一种基于批模式DataSetApi实现的机器学习函数库，而StreamML是一种流式的机器学习算子，其目的是为了满足网络安全监测的特定需求。与阿里巴巴开源的Alink相比，StreamML允许机器学习算法工程师通过配置规则的方式即可快速验证算法模型，无需编写任何程序代码。并且，流式机器学习算子StreamML实现了“模型训练/更新”与“模型使用”统一的理念。其核心功能是通过算法、技术及模型实现数据训练及对新数据检测。该流式机器学习算子StreamML引入的输入有三类，分别是：事件流、检测对象和对象属性；输出也包含三类，分别是：事件、告警和预警。

流式机器学习算子StreamML的组件栈包含三部分，从下往上依次为：机器学习方法、应用场景和产品业务。通过基本的机器学习算法（比如：统计学习算法、序列分析算法、聚类分析算法），流式机器学习算子StreamML可满足具体特定的安全监测应用场景（比如：行为特征异常检测、时间序列异常检测、群组聚类分析），进而为用户提供可理解的产品业务（比如：基线、用户及实体行为分析UEBA）。

行为特征异常检测：根据采集的样本数据（长时间）对统计分析对象建立行为基线，并以此基线为准，检测发现偏离正常行为模式的行为。例如：该用户通常从哪里发起连接？哪个运营商？哪个国家？哪个地区？这个用户行为异常在组织内是否为常见异常？

时间序列异常检测：根据某一个或多个统计属性，判断按时间顺序排列的数值序列是否异常，由此通过监测指标变化来发现安全事件。例如：监测某网站每小时的访问量以防止DDOS攻击；建模每个账号传输文件大小的平均值，检测出传输文件大小的平均值离群的账号。

群组聚类分析：对数据的特征属性间潜在相关性进行挖掘，将具有类似特征值的数据进行分组聚类。例如：该用户是否拥有任何特殊特征？可执行权限/特权用户？基于执行的操作命令和可访问的实体，来识别IT管理员、DBA和其它高权限用户。

5.Sabre如何快速适配复杂的客户环境?

由于客户规模较大，项目种类较多，部署环境较为复杂，或者存在多种Yarn集群版本，或者Sabre需作为单一Flink应用发布到客户已部署的Flink集群。如何节省成本及提高实施效率，快速适配上述复杂的部署环境是个亟需解决的问题，为此Sabre的设计原则是仅采用Flink的分布式计算能力，业务代码尽可能减少对API层的依赖，以便于兼容多种Flink版本。如图所示，Deploy、Core、APIs、Libraries四层是大家熟知的Flink基本的组件栈。Sabre对API层的依赖降到了最低，只引用了DataStream、KeyedStream和SplitStream三种数据流API。函数依赖则包括DataStream的assignTimestamps、flatMap、union、keyBy、split、process、addSink等函数，KeyedStream最基础的process函数，以及SplitStream的select函数。由于依赖的Flink API较少，Sabre可以很容易适配到各个Flink版本，从而具有良好的Flink版本兼容性。

6.如何保障Sabre稳定运行？

为减少引擎的维护成本，需要保障引擎在超限数据量的条件下亦然能够稳定运行，Sabre主要做了两个优化：流量控制和自我保护。

为了增强Sabre引擎的健壮性，避免因规则配置错误，导致生成大量无效告警，在输出端做了流量控制，以更好地保护下级应用。当下级抗压能力较弱时（例如数据库），整个系统会做输出降级。

另一个问题是，跑在JVM上的程序，经常会遇到由于长时间 Full GC导致OOM的错误，并且此时CPU占用率往往非常高，Flink同样存在上述问题。自我保护功能采用了同时兼顾“Window隶属规则的优先级”及“Window引用规则数量”两个条件的加权算法，以此根据全局规则语义实现自动推导Window优先级，并根据此优先级确定各个Window的自我保护顺序。实时监控CPU及内存占用，当超过一定阈值时，智能优化事件分布，以防出现CPU长期过高或内存使用率过大而导致的OOM问题。

目前，基于Flink构建的Sabre引擎还在继续开发新的功能，并会持续优化引擎性能。未来将总结凝练项目中的优秀实践，并及时回馈给Apache Flink社区。