关于golang:alertmanager-源码分析一-告警的处理

上篇说到告警曾经写入到内存构造中 AlertsProvider 中,并且 Dispatcher 通过订阅 AlertsProvider 获取一个 chan，可能实时读到新写入的 alert。

Dispatcher 的作用就是把 alert 分派给正确的 route 来解决告警，route 是一个树状构造，每个节点有 Matchers 用于判断和以后正在解决的 alert 是否匹配，匹配就对这个 alert 利用这个 route 上的规定，先看看 route 构造:

type Route struct {    parent *Route    Routes []*Route    RouteOpts RouteOpts    // alert 就是由 alert.LabelSet 和 Route.Matchers 两个是否匹配来决定是否应用这个 Route 来解决    Matchers labels.Matchers    // 以后层呈现一个 route 匹配 alert 后，是否在同级上持续应用这个 alert 尝试匹配其它 route    Continue bool}// RouteOpts 保留了跟以后 Route 绑定的一些规定，alert 匹配后就会应用这些规定解决type RouteOpts struct {    Receiver string    GroupBy map[model.LabelName]struct{}    GroupByAll bool    GroupWait      time.Duration    GroupInterval  time.Duration    RepeatInterval time.Duration    MuteTimeIntervals []string}

对树状构造数据，源码中应用了递归来遍历节点

// lset 就是传入的告警的 label 汇合func (r *Route) Match(lset model.LabelSet) []*Route {    // 以后节点匹配    if !r.Matchers.Matches(lset) {        return nil    }    var all []*Route    for _, cr := range r.Routes {        matches := cr.Match(lset) // 深度优先        all = append(all, matches...)        // 如果曾经匹配到同级的一个, 且 Continue=false 就完结以后层的匹配        if matches != nil && !cr.Continue {            break        }    }    // 如果以后 route 的子 route.Routes 为空或者 route.Routes 中没有匹配 lset 时, 那么以后 route 就是匹配的    if len(all) == 0 {        all = append(all, r)    }    return all}

找到了所有跟以后告警匹配的 route, 接下来就是解决告警了，看看 processAlert,

func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {    // 获取以后 route 上用于分组的标签 kv, 计算一个组的 fp    // 如果 route 的 groupBy 是 ["a", "b"], 那么 a=1,b="s" 的告警和 a=1,b="e" 的告警就会在一个 route 下的不同分组内    groupLabels := getGroupLabels(alert, route)    fp := groupLabels.Fingerprint()    d.mtx.Lock()    defer d.mtx.Unlock()        // 没 routeGroups 就建一个, route 对应的是 routeGroups, fp 对应的是 aggrGroup    routeGroups, ok := d.aggrGroupsPerRoute[route]    if !ok {        routeGroups = map[model.Fingerprint]*aggrGroup{}        d.aggrGroupsPerRoute[route] = routeGroups    }    ag, ok := routeGroups[fp]    if ok {        // groupBy 中的 label 呈现了一个新的取值组合, 间接写入        ag.insert(alert)        return    }    ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger)    routeGroups[fp] = ag    d.aggrGroupsNum++    d.metrics.aggrGroups.Inc()    // 终于, 一个新的告警被 Dispatcher 分派到了一个 routeGroup 下的一个 aggrGroup 中    ag.insert(alert)    // 每个 aggrGroup 会有个独立的解决以后分组告警的 groutine, 这里传给 ag.run 的匿名函数最终会是真正解决每个告警的函数    go ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {        // 处理函数就是应用 Dispatcher 的 stage 来解决        // stage 是在 Dispatcher 新建时编排的流水线, 外面都是 stage 接口类型的对象, stage 接口只有一个 Exec 办法        _, _, err := d.stage.Exec(ctx, d.logger, alerts...)        // ...        return err == nil    })}

所以 processAlert 就是把告警分派给某个 route 下的某个 aggrGroup, 而每个 aggrGroup 会有个独立的 groutine 专门解决本人 aggrGroup 内的告警，在持续看每个 aggrGroup 是怎么解决本人组内的告警之前，先理解一下 aggrGroup 的构造

type aggrGroup struct {    // 分组应用的 labels，比方下面例子中的 a=1,b="s"    labels   model.LabelSet    opts     *RouteOpts    logger   log.Logger    routeKey string    // 这个构造和后面告警写入时存储的构造 AlertsProvider 底层应用的雷同    // 实际上就是一个 mutex.Lock 守护的一个 map[fp]alert    alerts  *store.Alerts    // 退出构造, aggrGroup 启动后在逻辑上是属于 Dispatcher, 所以须要监听 Dispatcher 的退出    // 这个 ctx 就是 Dispatcher 的 ctx    ctx     context.Context    cancel  func()    done    chan struct{}    // 这个 timer 新建时应用的周期就是 group_wait, 即一个分组的首次告警等待时间    // 在这个组首次 flush 之后，会被置为 GroupInterval 即一个更长的期待周期    next    *time.Timer    timeout func(time.Duration) time.Duration    mtx        sync.RWMutex    hasFlushed bool}// 写入 aggrGroup 一个告警func (ag *aggrGroup) insert(alert *types.Alert) {    // 和 AlertsProvider 的 Put 时一样    if err := ag.alerts.Set(alert); err != nil {        level.Error(ag.logger).Log("msg", "error on set alert", "err", err)    }    ag.mtx.Lock()    defer ag.mtx.Unlock()    // 写入 aggrGroup 胜利后, 如果以后分组在新建后执行过 flush, 而且以后这个告警会尝试触发一次 flush,     // flush 是每个 aggrGroup 用来解决本人分组内告警的函数, 稍后会说    //     if !ag.hasFlushed && alert.StartsAt.Add(ag.opts.GroupWait).Before(time.Now()) {        ag.next.Reset(0)    }}

接下来就是 aggrGroup 是如何解决本人组内的告警的

func (ag *aggrGroup) run(nf notifyFunc) {    defer close(ag.done)    defer ag.next.Stop()    for {        select {        case now := <-ag.next.C:            // 如果以后告警没有在 GroupInterval 内实现, 就立刻终止, 因为下个 flush 来了            ctx, cancel := context.WithTimeout(ag.ctx, ag.timeout(ag.opts.GroupInterval))            // flush 中应用的 now 都是以这个工夫为准            ctx = notify.WithNow(ctx, now)            // Populate context with information needed along the pipeline.            ctx = notify.WithGroupKey(ctx, ag.GroupKey())            ctx = notify.WithGroupLabels(ctx, ag.labels)            ctx = notify.WithReceiverName(ctx, ag.opts.Receiver)            ctx = notify.WithRepeatInterval(ctx, ag.opts.RepeatInterval)            ctx = notify.WithMuteTimeIntervals(ctx, ag.opts.MuteTimeIntervals)            ag.mtx.Lock()            // ag.next 这个 timer 在 aggrGroup 新建的时候应用 group_await            // 首次触发后, 就被重置为 GroupInterval, 个别这个更长一点            ag.next.Reset(ag.opts.GroupInterval)            ag.hasFlushed = true            ag.mtx.Unlock()            ag.flush(func(alerts ...*types.Alert) bool {                return nf(ctx, alerts...)            })            cancel()        case <-ag.ctx.Done():            return        }    }}func (ag *aggrGroup) flush(notify func(...*types.Alert) bool) {    if ag.empty() { return }    var (        alerts      = ag.alerts.List()        alertsSlice = make(types.AlertSlice, 0, len(alerts))        now         = time.Now()    )    for _, alert := range alerts {        a := *alert        // 没解决的就把完结工夫置零, 不然后续流程如果很长, 可能就解决到一半 EndsAt 就到期了        if !a.ResolvedAt(now) {            a.EndsAt = time.Time{}        }        alertsSlice = append(alertsSlice, &a)    }    sort.Stable(alertsSlice)    // 这个 notify(alertsSlice...) 就是音讯解决流水线入口,     // if body 中的逻辑就是跑完流水线并且没有谬误再执行的逻辑    if notify(alertsSlice...) {        for _, a := range alertsSlice {            fp := a.Fingerprint()            got, err := ag.alerts.Get(fp)            if err != nil {                // This should never happen.                level.Error(ag.logger).Log("msg", "failed to get alert", "err", err, "alert", a.String())                continue            }            // 从以后的 aggrGroup 中清理曾经解决的告警            if a.Resolved() && got.UpdatedAt == a.UpdatedAt {                if err := ag.alerts.Delete(fp); err != nil {                    level.Error(ag.logger).Log("msg", "error on delete alert", "err", err, "alert", a.String())                }            }        }    }}

到这个地位告警曾经被解决掉了，实际上就是跑完流水线后告警的状态变了，aggrGroup 中清理告警只是清理告警的一个指针，还有其它中央的组件还存储者这个告警的指针，比方 AlertsProvider 中，不过那边就由 AlertsProvider 的 GC 函数解决了。

下面的流水线的入口函数被传递了很多层，我把它收集在一起就看的比较清楚:

// 最后是 aggrGroup 启动时定义, 传给 aggrGroup.run 函数, 实际上这个函数是一个闭包, 因为援用了 d.logger, d.stagego ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {    _, _, err := d.stage.Exec(ctx, d.logger, alerts...)    return err == nil})// aggrGroup.run 对它进一步封装, 把 aggrGroup.run 中计算好的 ctx 传给这个函数, 减少了闭包中的信息func (ag *aggrGroup) run(nf notifyFunc) {    for {        select {        case now := <-ag.next.C:            ag.flush(func(alerts ...*types.Alert) bool {                return nf(ctx, alerts...)            })            cancel()        }    }}// 最终是在 flush 中应用本组的 alert 调用这个函数func (ag *aggrGroup) flush(notify func(...*types.Alert) bool) {    if notify(alertsSlice...) {}}

总结下来，就是为了调用 d.stage.Exec(ctx, d.logger, alerts...)，应用了 Dispatcher.logger, aggrGroup 的 ctx 和 alerts。