前言

本系列将对夜莺平台各个模块的主要逻辑代码进行介绍，方便大家进行二次开发，本篇是系列的第二篇，judge模块的解读。

首先贴下夜莺的项目地址和架构图，正在使用夜莺的读者欢迎给夜莺加一个star

img

本篇主要讲解负责告警的judge模块，首先介绍下 transfer 到 judge 的数据流转

transfer 到 judge 数据流转

上文讲到的 transfer 模块，接收到数据之后，一份数据会转发给存储，另一份数据则会发给 judge，下面我们介绍下 transfer 到 judge 的数据流转

因为不是所有的监控数据都会配置告警策略，所以为了降低无效数据的转发，transfer 在把数据发给 judge 的之前，会对数据进行一次过滤，那过滤依据是什么呢？当然就是告警策略了，只有配置了告警策略的监控数据才会转发给judge组件，所以 transfer 其实会定期从 monapi 模块获取全量的告警策略。

拿到监控数据之后，因为 judge 是可以同时部署多个实例的，那 transfer 怎么判断把监控数据发送给哪个 judge 实例呢？这个工作其实由 monapi 和 transfer 配合一起来完成的。首先将一个 monapi 负责的工作

monapi 也需要把用户配置的告警策略分配给不同的 judge 实例，那 monapi 如何对告警策略进行分发呢，这里有三点要注意

  1. 每条策略只能发送到一个 judge 实例，不然会出现重复告警的问题
  2. 每个 judge 实例得到的告警策略要尽可能的均匀
  3. judge 实例扩容或者缩容（宕机）之后，告警策略要重新分配，不然会有部分分策略不生效，或者策略分发不均衡

为了满足上面3个要求，monapi 在给 judge 分发策略的时候，使用的是一致性hash 算法，并且每个 judge 实例都会定期上报自己的心跳，monapi 会定期的检查 judge 的心跳，一旦 judge 实例的状态发送变化，monapi 会重建 judge 的 hash环，对告警策略进行重新分配，通过这个机制 judge 具备了水平扩展和高可用部署的能力

transfer 拿到的告警策略，每个告警策略数据结构中有一个 JudgeInstance 字段，标记匹配到此条告警策略的数据要发给哪个 judge 时候，这样 transfer 不需要再管judge 实例的状态了，这个完成交给 monapi 来负责，monapi 下发告警数据发给哪个 judge 实例，transfer 就把相关的监控数据发给哪个 judge 实例。

judge模块

首先看一下judge的main函数，对主要功能都加了注释说明

func main() {
  cfg := config.Config
    identity.Parse()
    loggeri.Init(cfg.Logger)
    go stats.Init("n9e.judge")

    query.Init(cfg.Query, "rdb") //初始化查询数据的client
    redi.Init(cfg.Redis) //初始化 redis 用来推送告警事件

    cache.InitHistoryBigMap() //初始化bigMap，在内存中存放告警时用到的监控数据，数据会轮询更替，不会一直变大
    cache.Strategy = cache.NewStrategyMap() // 存放从monapi获取的告警策略
    cache.NodataStra = cache.NewStrategyMap() // 存放从monapi获取的nodata告警策略
    cache.SeriesMap = cache.NewIndexMap() // 存放监控数据的索引，用来查询内存中的数据

    go rpc.Start() //用来接收监控数据

    go stra.GetStrategy(cfg.Strategy) //定期从monapi获取告警策略
    go judge.NodataJudge(cfg.NodataConcurrency)
    go report.Init(cfg.Report, "rdb") //定期上报自己的心跳，让monapi知道自己是存活的


    r := gin.New()
    routes.Config(r)
    go http.Start(r, "judge", cfg.Logger.Level)

    ending()
}

judge 接收到监控数据之后，再加上从 monapi 拿到的告警策略就可以进行告警规则计算了。下面是 judge 的主要流程，从流程图可以看出来，整个告警处理过程是一个简单的流式计算。

img

judge 提供了一个 rpc 接口来接收监控数据，接收到监控数据之后，这里做了一个小的优化，judge 可以根据告警数据中携带的 sid 字段很快的获取到其对应的告警策略，sid 字段是在 transfer 模块加上去的，这样 judge 就不需要再根据metric + tag 来让告警数据和告警策略进行匹配，既可以提高匹配效率，也可以降低cpu消耗

stra, exists := GetStra(val.Sid)
if !exists {
        stats.Counter.Set("point.miss", 1)
        return
}

//...
history, info, lastValue, status := Judge(stra, expr, historyData, val, now)
statusArr = append(statusArr, status)

if value == "" {
    value = fmt.Sprintf("%s: %s", expr.Metric, lastValue)
} else {
    value += fmt.Sprintf("; %s: %s", expr.Metric, lastValue)
}

//...
historyArr = append(historyArr, history)
eventInfo += info

sendEventIfNeed(statusArr, event, stra)

judge函数中的 judgeItemWithStrategy() 会根据告警策略中配置的告警函数，调用对应的告警函数，来进行告警计算，如果想自己扩展告警函数，可以在通过在 src/modules/judge/judge/func.go 增加新的告警函数来实现。告警函数的接口如

type Function interface {
    Compute(vs []*dataobj.HistoryData) (leftValue dataobj.JsonFloat, isTriggered bool)
}

生成告警事件之后，sendEvent 负责将告警事件推送到 redis 中。

func sendEvent(event *dataobj.Event) {
    // update last event
    cache.LastEvents.Set(event.ID, event)

    err := redi.Push(event)
    if err != nil {
        stats.Counter.Set("redis.push.failed", 1)
        logger.Errorf("push event:%v err:%v", event, err)
    }
}

到此 judge 的整个工作就结束了，下篇将为大家带来数据存储模块的解读

作者简介

秦叶宁 企业级开源运维平台 Nightingale 主程，Urlooker 作者，现负责滴滴私有云运维产品方向的工作，如有运维平台的搭建需求，欢迎与我联系：）