马蜂窝大交通业务监控报警系统架构设计与实现

部门的业务线越来越多，任何一个线上运行的应用，都可能因为各种各样的原因出现问题：比如业务层面，订单量比上周减少了，流量突然下降了；技术层面的问题，系统出现 ERROR ，接口响应变慢了。拿大交通业务来说，一个明显的特点是依赖很多供应商的服务，所以我们还需要关注调用供应商接口是否出现异常等等。

为了让大交通下的各业务线都能够通过报警尽早发现问题、解决问题，进而提升业务系统的服务质量，我们决定构建统一的监控报警系统。一方面在第一时间发现已经出现的系统异常，及时解决；另一方面尽早发现一些潜在的问题，比如某个系统目前来看没有影响业务逻辑的正常运转，但是一些操作耗时已经比较长等，这类问题如果不及时处理，将来就很可能影响业务的发展。

本文主要介绍马蜂窝大交通业务监控报警系统的定位、整体架构设计，以及我们在落地实践过程中的一些踩坑经验。

我们希望监控报警系统主要具备以下三个能力：

1. 常用组件自动报警 ：对于各业务系统常用的框架组件（如 RPC ，HTTP 等）创建默认报警规则，来方便框架层面的统一监控。

2. 业务自定义报警 ：业务指标由业务开发自定义埋点字段，来记录每个业务和系统模块的特殊运行状况。

3. 快速定位问题 ：发现问题并不是目的，解决才是关键。我们希望在完成报警消息发送后，可以让开发者一目了然地发现问题出现在什么地方，从而快速解决。

在这样的前提下，报警中心的整体架构图和关键流程如下图所示：

纵向来看，Kafka 左侧是报警中心，右侧是业务系统。

报警中心的架构共分为三层，最上层是 WEB 后台管理页面，主要完成报警规则的维护和报警记录的查询；中间层是报警中心的核心；最下面一层是数据层。业务系统通过一个叫做 mes-client-starter 的 jar 包完成报警中心的接入。

我们可以将报警中心的工作划分为五个模块：

1. 数据收集

我们采用指标采集上报的方式来发现系统问题，就是将系统运行过程中我们关注的一些指标进行记录和上传。上传的方式可以是日志、 UDP 等等。

数据收集模块我们没有重复造轮子，是直接基于 MES （ 马蜂窝内部的大数据分析工具 ）来实现，主要考虑下面几个方面的原因：一来数据分析和报警在数据来源上是相似的；二来可以节省很多开发成本；同时也方便报警的接入。

那具体应该采集哪些指标呢？以大交通业务场景下用户的一次下单请求为例，整个链路可能包括 HTTP 请求、Dubbo 调用、SQL 操作，中间可能还包括校验、转换、赋值等环节。一整套调用下来，会涉及到很多类和方法，我们不可能对每个类、每个方法调用都做采集，既耗时也没有意义。

为了以最小的成本来尽可能多地发现问题，我们选取了一些系统常用的框架组件自动打点，比如 HTTP、SQL、我们使用的 RPC 框架 Dubbo 等 ，来实现框架层面的统一监控。

而对于业务来说，每个业务系统关注的指标都不一样。对于不同业务开发人员需要关注的不同指标，比如支付成功订单数量等，开发人员可以通过系统提供的 API 进行手动埋点，自己定义不同业务和系统模块需要记录的状况。

2. 数据存储

对于采集上来的动态指标数据，我们选择使用 Elasticsearch 来存储，主要基于两点原因：

一是动态字段存储。 每个业务系统关注的指标可能都不一样，每个中间件的关注点也不同，所以埋哪些字段、每个字段的类型都无法预知，这就需要一个可以动态添加字段的数据库来存储埋点。Elasticsearch 不需要预先定义字段和类型，埋点数据插入的时候可以自动添加。

二是能够经得起海量数据的考验 。每个用户请求经过每个监控组件都会产生多条埋点，这个数据量是非常庞大的。Elasticsearch 可以支持大数据量的存储，具有良好的水平扩展性。

此外，Elasticsearch 还支持聚合计算，方便快速执行 count , sum , avg 等任务。

3. 报警规则

有了埋点数据，下一步就需要定义一套报警规则，把我们关注的问题量化为具体的数据，验证是否超出了预设的阈值。这是整个报警中心最复杂的问题，也最为核心。

之前的整体架构图中，最核心的部分就是「规则执行引擎」，它通过执行定时任务来驱动系统的运行。首先，执行引擎会去查询所有生效的规则，然后根据规则的描述到 Elasticsearch 中进行过滤和聚合计算，最后将上一步聚合计算得结果跟规则中预先设定的阈值做比较，如果满足条件则发送报警消息。

这个过程涉及到了几个关键的技术点：

(1). 定时任务

为了保证系统的可用性，避免由于单点故障导致整个监控报警系统失效，我们以「分钟」为周期，设置每一分钟执行一次报警规则。这里用的是 Elastic Job 来进行分布式任务调度，方便操控任务的启动和停止。

(2). 「三段式」报警规则

我们将报警规则的实现定义为「过滤、聚合、比较」这三个阶段。举例来说，假设这是一个服务 A 的 ERROR 埋点日志：

app_name=B   is_error=false  warn_msg=aa   datetime=2019-04-01 11:12:00
app_name=A   is_error=false                datetime=2019-04-02 12:12:00
app_name=A   is_error=true   error_msg=bb  datetime=2019-04-02 15:12:00
app_name=A   is_error=true   error_msg=bb  datetime=2019-04-02 16:12:09

报警规则定义如下：

• 过滤 ：通过若干个条件限制来圈定一个数据集。对于上面的问题，过滤条件可能是：app_name=A , is_error=true , datetime between '2019-14-02 16:12:00' and '2019-14-02 16:13:00'.

  
  

• 聚合 ：通过 count,avg,sum,max 等预先定义的聚合类型对上一步的数据集进行计算，得到一个唯一的数值。对于上面的问题，我们选择 count 来计算出现 ERROR 的次数。

  
  

• 比较 ：把上一步得到的结果与设定的阈值比较。

对于一些复杂条件的报警 ，比如我们上边提到的失败率和流量波动，应该如何实现呢？

假设有这样一个问题：如果调用的 A 服务失败率超过 80%，并且总请求量大于 100，发送报警通知。

我们知道，失败率其实就是失败的数量除以总数量，而失败的数量和总数量可以通过前面提到的 「过滤+聚合」 的方式得到，那么其实这个问题就可以通过如下的公式描述出来：

failedCount/totalCount>0.8&&totalCount>100

然后我们使用表达式引擎 fast-el 对上面的表达式进行计算，得到的结果与设定的阈值比较即可。

(3) 自动创建默认报警规则

对于常用的 Dubbo, HTTP 等，由于涉及的类和方法比较多，开发人员可以通过后台管理界面维护报警规则，报警规则会存储到 MySQL 数据库中，同时在 Redis 中缓存。

以 Dubbo 为例，首先通过 Dubbo 的 ApplicationModel 获取所有的 provider 和 consumer，将这些类和方法的信息与规则模板结合（规则模板可以理解为剔除掉具体类和方法信息的规则），创建出针对某个类下某个方法的规则。

比如：A 服务对外提供的 dubbo 接口/ order / getOrderById 每分钟平均响应时间超过 1 秒则报警；B 服务调用的 dubbo 接口/ train / grabTicket /每分钟范围 false 状态个数超过 10 个则报警等等。

4. 报警行为

目前在报警规则触发后主要采用两种方式来发生报警行为：

• 邮件报警 ：通过对每一类报警制定不同的负责人，使相关人员第一时间获悉系统异常。

• 微信报警 ：作为邮件报警的补充。

之后我们会持续完善报警行为的策略，比如针对不同等级采用不同的报警方式，使开发人员既可以迅速发现问题，又不过多牵扯在新功能研发上的精力。

5. 辅助定位

为了能够快速帮助开发人员定位具问题，我们设计了命中抽样的功能：

首先，把命中规则的 tracer_id 提取出来，提供一个链接可以直接跳转到 kibana 查看相关日志，实现链路的还原。

其次，开发人员也可以自己设置他要关注的字段，然后把这个字段对应的值也抽取出来，问题出在哪里就可以一目了然地看到。

技术实现上，定义一个命中抽样的字段，这个字段里面允许用户输入一个或者多个 dollar 大括号。比如我们可能关注某个供应商的接口运行情况，则命中抽样的字段可能为下图中上半部分。在需要发送报警消息的时候，提取出里面的字段，到 ES 中查询对应的值，用 freemarker 来完成替换，最终发送给开发人员的消息是如下所示，开发人员可以快速知道系统哪里出了问题。

大交通业务监控报警系统的搭建是一个从 0 到 1 的过程，在整过开发过程中，我们遇到了很多问题，比如：内存瞬间被打满、ES 越来越慢、频繁 Full GC ，下面具体讲一下针对以上几点我们的优化经验。

踩过的坑

1. 内存瞬间被打满

任何一个系统，都有它能承受的极限，所以都需要这么一座大坝，在洪水来的时候能够拦截下来。

报警中心也一样，报警中心对外面临最大的瓶颈点在接收 Kafka 中传过来的 MES 埋点日志。上线初期出现过一次由于业务系统异常导致瞬间大量埋点日志打到报警中心，导致系统内存打满的问题。

解决办法是评估每个节点的最大承受能力，做好系统保护。针对这个问题，我们采取的是限流的方式，由于 Kafka 消费消息使用的是拉取的模式，所以只需要控制好拉取的速率即可，比如使用 Guava 的 RateLimiter ：

messageHandler = (message) -> {
  RateLimiter messageRateLimiter = RateLimiter.create(20000);
  final double acquireTime = messageRateLimiter.acquire();
  /**
   save..
  */
}