谈谈复杂事件处理与IT运维监控，复杂事件管理规范

本文讲述了谈谈复杂事件处理与IT运维监控，复杂事件管理规范。

CEP简介

CEP全称为 Complex Event Processing 复杂事件处理，其可以通过在流式数据中发现符合某种特征的模式进而触发对应的后续动作，既支持基于单条事件的简单无状态的模式匹配（例如基于事件中的某个字段进行筛选过滤），也可以支持基于关联／聚合／时间窗口等跨事件的复杂有状态模式匹配（例如计算滑动时间窗口移动均值）。受益于其直接作用于流式数据，无需查询持久化数据库，对底层数据库不会产生任何压力，以及其强大的模式发现能力，在监控系统中，如果把CEP与流处理引擎结合，在IT运维管理中，可以大大增强告警的实时性以及适用范围。

CEP对IT运维的价值

传统的ITOM主要对底层的软硬件基础架构、对单一指标基于静态的阈值进行监控告警，这里有两个关键词：基础架构 以及 单一， 这其实正好对应了传统IT监控的两大痛点。

单一导致的结果就是误报。服务器的CPU利用率上升有时是因为交易量的上升带来的正常现象，只要在合理区间内就无需告警，但是CPU利用率的孤立上升就可能是因为代码缺陷造成的。有经验的工程师一眼就能看出是否是故障，是因为工程师在一瞬间就综合分析了各个相关指标。

针对基础架构则让运维人员的生活非常苦逼，有功劳都是其他部门的，出了故障都是运维的兄弟在顶包。所以近些年来基本上所有的企业都在做APM，通过网络抓包或者日志埋点等方式可以提取交易成功率/交易量/成功率等反映业务性能的指标，做了不少漂亮工程。不过不管是交易量还是成功率都还是从系统的角度去看问题，真正能带来多少业务价值其实也很难说，大屏上那些五颜六色的图表可能更多时候也是在领导检查或者参观时才体现其价值。

从数据来看，其实IT运维过程中的数据是最完整的，既有包含了服务器，网络设备等基础设施的底层运行信息，也有包含中间件和数据库的中间层系统信息，还有包含了全部业务过程的上层应用日志信息。在这个数据时代，IT运维正在向IT运营转变，理应能够发挥更大的业务价值。

举例：对于一个金融企业，如果能从应用日志中提取到同一个账户在1分钟内在距离50公里的两地柜员机取款的类欺诈信息来支持风控，IT运维所承担的就不只是保障作用，而是直接参与到了业务决策过程当中。

虽然CEP不是为IT运维管理而生，但是在一定程度上CEP确实可以解决上述两个问题。CEP最强大的就是其模式匹配引擎，其不仅可以作用于不同类型的事件，更可以按照时间窗口，发生顺序和次数以及其他状态聚合结果进行模式匹配，可以和各种业务规则进行对应。另外CEP是直接作用于流式数据，而非通过定期查询数据库的方式，因此最实时，且对数据库没有任何压力。

目前的Flink流引擎已经自带了CEP模块。Flink官方给出的CEP例子正好就是针对数据中心监控的场景，案例中需要对Rack的温度进行监控，对于同一个Rack，当10秒内连续两次温度超过温度阈值时预警，当20秒内产生连续两个预警，且第二次预警温度高于第一次时进行告警。该模式中有不同状态的切换，有时间窗口的滑动等。如果没有CEP，需要自行构建对应的状态机进行处理，但是基于CEP强大的模式挖掘能力之上的实现非常简单，由此可见CEP的威力。不过Flink中的CEP模块是与其流失API紧密结合的，如果我们的前置流引擎不是Flink，则无法直接使用。

我们在构建智能化监控系统时，从性能方面去考虑，在流引擎与CEP之间添加了基于AKKA的多层路由模块，可以按照业务系统，服务器，实例以及指标级别进行消息的路由，CEP引擎内嵌在每个Actor内部，可以在不同的路由级别对不同范围的消息进行模式匹配。下图是对应的架构图。 

监控系统的架构以及对应实现不在我们这篇文章范围之内，本文接下来介绍下如何使用CEP来完成一个复杂监控告警从生成到关闭的过程。

CEP作为监控告警的规则引擎

本文中使用了一个开源的CEP组件 esper (https://github.com/espertechinc/esper)， 目前该组件已经被weblogic等中间件厂商集成到自己的日志异常检测功能中。其本身是一个JAR库，无需其他运行时框架，所以使用起来非常方便。其支持一种DSL语言EPL，与SQL类似，用户通过输入不同的EPL来定义不同的匹配模式。

本文不包含基本的EPL语法介绍，有兴趣的同学可以移步这里(http://blog.csdn.net/luonanqin/article/details/21300263) 以及esper的官网(http://www.espertech.com/esper/)。

本文主要从一个实际的监控需求案例出发，来介绍esper的相关功能。

监控告警需求

监控原始事件定义如下：

//name代表监控指标的名称，例如CPU.CPUUtil

//timestamp指某一个具体的时间点

//value代表在timestamp时间点的指标值

//tags里面存储了一些其他描述信息，例如服务器名，ip地址，所属业务系统等等

case class GenericMetric(timestamp:Long, name:String, value:Any, tags:Map[String, Any])

//timestamp代表Warning产生的时间点

//name代表指标名称

//value代表产生Warning前最后一个事件的值

case class Warning(timestamp:Long, name:String, value:Any, tags:Map[String, Any])

//timestamp代表关闭事件产生的时间点

//name代表指标名称

case class Close(timestamp:Long, name:String)

//timestamp代表Critical产生的时间点

//name代表指标名称

//value代表产生Critical前最后一个事件的值

case class Critical(timestamp:Long, name:String, value:Any, tags:Map[String, Any])

告警规则1

当value连续10次 >= 0.5，< 0.8时产生Warning级别的告警，当value连续10次 >= 0.8时产生Critical级别的告警。

当产生Warning级别的告警后，当出现不在 >= 0.5 < 0.8区间的事件数据时，产生Close事件把Warning告警关闭。

当产生Critical级别的告警后，当出现不在 < 0.8区间的事件数据时，产生Close事件把Critical告警关闭。

Warning create epl:

insert into Warn select (select value from GenericMetric.std:lastevent()) as value, (select name from GenericMetric.std:lastevent()) as name, (select tags from GenericMetric.std:lastevent()) as tags,current_timestamp as timestamp from pattern[ every [10] (GenericMetric(value >= 0.5 and value < 0.8) and not GenericMetric(value < 0.5 or value >= 0.8))]

Critical create epl:

insert into Critical select (select value from GenericMetric.std:lastevent()) as value, (select name from GenericMetric.std:lastevent()) as name, (select tags from GenericMetric.std:lastevent()) as tags, current_timestamp as timestamp from pattern[ every [10] (GenericMetric(value >= 0.8) and not GenericMetric(value < 0.8))]

Warn close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every [1] (Warn -> (GenericMetric(value < 0.5 or value >= 0.8)))]

Critical close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every [1] (Critical -> (GenericMetric(value < 0.8)))]

告警规则2

当value连续10次 >= 0.5，< 0.8时产生Warning级别的告警，当value连续10次 >= 0.8时产生Critical级别的告警。

当产生Warning级别的告警后，当连续出现10次< 0.5的事件数据时，产生Close事件把Warning告警关闭。

当产生Critical级别的告警后，当连续出现10次 < 0.8区间的事件数据时，产生Close事件把Critical告警关闭。

告警的生成规则和规则1是相同的，这里就不再重复给出，需要关注的是关闭的规则与之前不同：

Warn close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every [1] (Warn -> every [10](GenericMetric(value < 0.5)))]

Critical close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every [1] (Critical -> every [10](GenericMetric(value < 0.8)))]

告警规则3

当value连续1分钟 >= 0.5，< 0.8时产生Warning级别的告警，当value连续1分钟 >= 0.8时产生Critical级别的告警。

当产生Warning级别的告警后，当连续出现1分钟< 0.5的事件数据时，产生Close事件把Warning告警关闭。

当产生Critical级别的告警后，当连续出现1分钟 < 0.8区间的事件数据时，产生Close事件把Critical告警关闭。

Warn create epl:

insert into Warn select (select value from GenericMetric.std:lastevent()) as value, (select name from GenericMetric.std:lastevent()) as name, (select tags from GenericMetric.std:lastevent()) as tags,current_timestamp as timestamp from pattern[ every  (GenericMetric(value >= 0.5 and value < 0.8)  -> ( timer:interval(60 sec) and not GenericMetric(value < 0.5 or value >= 0.8)))]

Critical create epl:

insert into Critical select (select value from GenericMetric.std:lastevent()) as value, (select name from GenericMetric.std:lastevent()) as name, (select tags from GenericMetric.std:lastevent()) as tags,current_timestamp as timestamp from pattern[ every  (GenericMetric(value >= 0.8)  -> ( timer:interval(60 sec) and not GenericMetric(value <  0.8)))]

Warn close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every (Warn -> ( timer:interval(60 sec) and GenericMetric(value < 0.5) and not GenericMetric(value >= 0.5)))]

Critical close epl:

insert into Close select (select name from GenericMetric.std:lastevent()) as name,current_timestamp as timestamp from pattern [every (Critical -> ( timer:interval(60 sec) and GenericMetric(value < 0.8) and not GenericMetric(value >= 0.8)))]

如果 没有CEP，要实现上述告警规则需要构建状态机，以及对状态跳转过程中的历史数据的记录，实现起来比较复杂。但是基于esper，对应每个规则只是一句简单的EPL，而且多个规则可以同时起作用，代码可读性以及扩展性都比构建状态机要好很多。

小结

本篇是从基础架构监控的层面来介绍了CEP。当然CEP更强大的功能是对结构化后的日志数据进行模式匹配，与复杂的业务规则进行对应，发挥更大的业务价值，后续的CEP系列文章里面会更多从日志数据挖掘的方面去做相关介绍。

此外，基于AKKA构建分布式的监控告警系统也是一个很有意思的主题，充分利用了actor的轻量级优势，通过创建百万甚至千万级别的actor与各指标进行对应，充分利用了服务器的计算资源，在相对普通的服务器集群上可以支撑海量的指标实时有状态监控，后续也会单独进行介绍。

复杂事件处理

复杂事件处理（CEP，Complex Event Processing）是一种基于动态环境中事件流的分析技术，事件在这里通常是有意义的状态变化，通过分析事件间的关系，利用过滤、关联、聚合等技术，根据事件间的时序关系和聚合关系制定检测规则，持续地从事件流中查询出符合要求的事件序列，最终分析得到更复杂的复合事件，主要用于网络诈欺识别等防止犯罪，银行等金融行业防止，以及风险规避和营销决策等.

复杂事件处理面临多方面的挑战：

减少应用存储数据（在分析数据之前）造成的延迟

能够持续，实时地分析多个数据流

能够关联不同数据流中的事件，从而发现新的相关情形

能够迅速响应新发现的危险或机会

能够迅速的将先前发现的规律应用到新的数据流分析模型中

能够利用已有的应用开发能力快速开放新的高性能，高扩展度的应用

确保应用和系统的连贯性。

事件处理语言

事件处理语言（EPL，Event processing language）用于系统中制定和查询感兴趣的事件序列，通常是类SQL的语句，从SQL语句中扩展而来.

事件关系

事件应该包含一些基本的要素：类型、发生事件以及更多的一些定义属性

通常需要关联多个事件进行分析处理，其中事件间的关系主要有5种：

时间关联

空间关联

依赖关系

事物的状态属性之间彼此的依赖关系和约束关系。

因果关系

事件处理过程

复杂事件处理过程包括：

格式化：将事件获取模块得到的事件信息转化为内部处理的形式

预处理：将事件按照字段内容进行处理

模式侦测：将数个事件复合起来，找出复合事件

事件发派：将复合事件发送到相应的处理模块

执行动作：处理模型按照事件状况执行相应的动作

以下是复杂事件处理系统中的关键模块组成图（来自《深入浅出复合事件处理》），

EPL解析器：复杂事件处理系统中EPL语言被解析器解析为处理引擎能理解的语言（类SQL解析器）。

规则管理：管理EPL。

事件接入：通过SOA、ESB、MOM、读取日志等方式将消息接入。

预处理：将事件依据字段内容进行处理。

CEP引擎：找出事件关联。

数据模型：维护内部数据。

事件发派：将已经发现的复合事件发派到负责处理的行动模块中。

行动模块：对复合事件采取行动 。

如何评价一个CEP平台：

开发工具中的编程模型，是否能吸引众多不同类型的开发人员？

是否提供了高可用性和高安全性？

是否可以部署在多个使用阵列配置的服务器？

是否提供了监视功能？

其引擎是否可以易于扩展，从而来支持公司特有的高级的逻辑？

是否有恢复机制，即使一旦服务器故障，也可以防止数据丢失？

进出处理引擎的消息传递，其机制是否可靠？

它同数据库以及其他系统能否很好的集成？

这个平台是否支持连续型实时数据管理的各个生命周期，包括分析，存储，建模和一个新的实时分析应用的部署？

事件驱动架构

事件驱动架构(Event-Driven Architecture，EDA) 基于消息传递，现在主要是异步通讯的消息队列的架构，依据“发布/订阅”模式，通过特定模式来对业务事件作出响应，耦合度较低.

面向服务架构

面向服务架构 (Service-Oriented Architecture, SOA) 基于“请求/响应”的形式

下面这幅图可以较好地体现基于SOA和EDA两种体系结构的差异处，SOA更多地是面向垂直系统的请求和响应处理，EDA则是应对横向的系统通.

上文就是小编为大家整理的谈谈复杂事件处理与IT运维监控，复杂事件管理规范。

国内(北京、上海、广州、深圳、成都、重庆、杭州、西安、武汉、苏州、郑州、南京、天津、长沙、东莞、宁波、佛山、合肥、青岛)睿象云智能运维平台软件分析、比较及推荐。