夜莺二次开发指南 - 监控系统（2）

关注
发布于: 2020 年 12 月 24 日
前言本系列将对夜莺平台各个模块的主要逻辑代码进行介绍，方便大家进行二次开发，本篇是系列的第二篇，judge模块的解读。
﻿
首先贴下夜莺的项目地址和架构图，正在使用夜莺的读者欢迎给夜莺加一个star
github地址 https://github.com/didi/nightingale
v3.0架构图
﻿
本篇主要讲解负责告警的judge模块，首先介绍下 transfer 到 judge 的数据流转
transfer 到 judge 数据流转上文讲到的 transfer 模块，接收到数据之后，一份数据会转发给存储，另一份数据则会发给 judge，下面我们介绍下 transfer 到 judge 的数据流转
﻿
因为不是所有的监控数据都会配置告警策略，所以为了降低无效数据的转发，transfer 在把数据发给 judge 的之前，会对数据进行一次过滤，那过滤依据是什么呢？当然就是告警策略了，只有配置了告警策略的监控数据才会转发给judge组件，所以 transfer 其实会定期从 monapi 模块获取全量的告警策略。
﻿
拿到监控数据之后，因为 judge 是可以同时部署多个实例的，那 transfer 怎么判断把监控数据发送给哪个 judge 实例呢？这个工作其实由 monapi 和 transfer 配合一起来完成的。首先将一个 monapi 负责的工作
﻿
monapi 也需要把用户配置的告警策略分配给不同的 judge 实例，那 monapi 如何对告警策略进行分发呢，这里有三点要注意
﻿
每条策略只能发送到一个 judge 实例，不然会出现重复告警的问题
每个 judge 实例得到的告警策略要尽可能的均匀
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 的主要流程，从流程图可以看出来，整个告警处理过程是一个简单的流式计算。
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 作者，现负责滴滴私有云运维产品方向的工作，如有运维平台的搭建需求，欢迎与我联系：）
﻿
﻿