前篇说到告警写入后被分发到dispatcher的aggrGroupsPerRoute中的aggrGroup里，然后每个aggrGroup会启动一个自己的goroutine按照group_wait和group_interval两种频率来定时调用dispatcher.stage.Exec方法来处理告警，实际上dispatcher.stage中存储的就是由多种处理函数编排成的一个告警处理流水线，也就是架构图中的下面这部分：
pipeline的构建是在main函数中创建dispatcher的时候，很容易找到，这里不赘述了，我们看看 pipeline 是怎样定义自己的 Exec 方法的，
// pipeline 就是 RoutingStage 类型，
// 它是基于 ctx 中的 receiver 进入这个 receiver 的 Stage
type RoutingStage map[string]Stage
// 看看流水线构建函数是如何为每个 receiver 配置 Stage 的
func (pb *PipelineBuilder) New(
receivers map[string][]Integration,
wait func() time.Duration,
inhibitor *inhibit.Inhibitor,
silencer *silence.Silencer,
muteTimes map[string][]timeinterval.TimeInterval,
notificationLog NotificationLog,
peer Peer,
) RoutingStage {
rs := make(RoutingStage, len(receivers))
ms := NewGossipSettleStage(peer)
is := NewMuteStage(inhibitor)
ss := NewMuteStage(silencer)
tms := NewTimeMuteStage(muteTimes)
// 基于 receiver 的 name 编排了一个包含多个 Stage 对象的 MultiStage
for name := range receivers {
st := createReceiverStage(name, receivers[name], wait, notificationLog, pb.metrics)
rs[name] = MultiStage{ms, is, tms, ss, st}
}
return rs
}
// 实现了 Exec 方法实际上是实现了 Stage 接口
// 流水线就是由各种 Stage 对象组合成的，后面再说 Stage 的设计，
// 先看看 RoutingStage 做了什么
func (rs RoutingStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
// 从 context 中获取当前路由下配置的告警接收器
receiver, ok := ReceiverName(ctx)
if !ok {
return ctx, nil, errors.New("receiver missing")
}
// 从 RoutingStage 中找到对应的 MultiStage 执行 MultiStage.Exec
s, ok := rs[receiver]
if !ok {
return ctx, nil, errors.New("stage for receiver missing")
}
return s.Exec(ctx, l, alerts...)
}
// 这个时候 aggGroup 经过 RoutingStage
// 为这些 Alerts 找到了 MultiStage
// 我们看看 MultiStage
// MultiStage 就是个包含了多个 Stage 的数组
type MultiStage []Stage
func (ms MultiStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
var err error
// 顺序执行使用数组中的每个 Stage.Exec()
for _, s := range ms {
if len(alerts) == 0 {
return ctx, nil, nil
}
ctx, alerts, err = s.Exec(ctx, l, alerts...)
if err != nil {
return ctx, nil, err
}
}
return ctx, alerts, nil
}
复制代码到这里总结一下，Dispatcher下的每个aggGroup先按照自己的receiver.Name通过调用RoutingStage.Exec中找到对应的MultiStage，然后顺序调用其中的每个Stage.Exec，接下来看下Stage的设计：
type Stage interface {
Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context context.Context, []*types.Alert, error)
}
// 举几个具体的 Stage 类型
type FanoutStage []Stage
type GossipSettleStage struct { peer Peer }
type MuteStage struct { muter types.Muter }
复制代码Stage这里是一个只约定了Exec函数的接口，所以任何一个对象只要定义了相同签名的Exec函数就是Stage类型，你会在源码中很容找到各种Stage，然后在对应的Exec方法中就知道告警在当前Stage中会被怎样处理，Exec的入参数中alerts表示哪些告警进入这个Stage，然后出参中的alerts就是经过当前Stage处理还剩哪些告警，ctx可以很方便各个Stage获取当前流水线上的参数，当然也可以写入参数让后面的Stage使用。
前面RoutingStage.Exec和MultiStage.Exec已经看过了我这里再找几个Stage看看里面的具体行为：
// 负责并发的执行一些 Stage
type FanoutStage []Stage
func (fs FanoutStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
var (
wg sync.WaitGroup
me types.MultiError
)
wg.Add(len(fs))
// FanoutStage 和 MultiStage 使用的相同结构 []Stage
// 但是 FanoutStage 是并发的执行
for _, s := range fs {
go func(s Stage) {
if _, _, err := s.Exec(ctx, l, alerts...); err != nil {
me.Add(err)
}
wg.Done()
}(s)
}
wg.Wait()
if me.Len() > 0 {
return ctx, alerts, &me
}
return ctx, alerts, nil
}
func createReceiverStage(
name string,
integrations []Integration,
wait func() time.Duration,
notificationLog NotificationLog,
metrics *Metrics,
) Stage {
// 这个是 FanoutStage 构建时
// 里面是多个可以并发的 MultiStage
var fs FanoutStage
for i := range integrations {
recv := &nflogpb.Receiver{
GroupName: name,
Integration: integrations[i].Name(),
Idx: uint32(integrations[i].Index()),
}
var s MultiStage
s = append(s, NewWaitStage(wait))
s = append(s, NewDedupStage(&integrations[i], notificationLog, recv))
s = append(s, NewRetryStage(integrations[i], name, metrics))
s = append(s, NewSetNotifiesStage(notificationLog, recv))
fs = append(fs, s)
}
return fs
}
复制代码再看看静默和抑制的Stage
type MuteStage struct {
muter types.Muter
}
func NewMuteStage(m types.Muter) *MuteStage {
return &MuteStage{muter: m}
}
func (n *MuteStage) Exec(ctx context.Context, _ log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
var filtered []*types.Alert
// 检查每个 alert 的 labelset 是否跟静默规则或者抑制规则的 labelSet 匹配
// 如果 alert 的 Labels 不匹配 Mute 就保留下来, 传给下一个 stage
for _, a := range alerts {
if !n.muter.Mutes(a.Labels) {
filtered = append(filtered, a)
}
}
return ctx, filtered, nil
}
复制代码MuteStage被用来实现SilenceStage和InhibitStage, 它包含了一个 muter，MuteStage.Exec最重要的就是调用muter.Mutes方法，那么muter就是一个包含Mutes方法的接口，Silencer和Inhibitor实现各自的 Mutes方法就可以作为MuteStage，那我们再看看它们各自是怎样实现Mutes方法的：
// 这个就是 Inhibitor 实现的 Muter 接口
// 抑制功能设计是这样的：
// 对于 a, b 两个 alert
// 如果 a 满足 SourceMatchers
// b 满足 TargetMatchers
// 则 Equal 成立时, 用 a 抑制 b
// Equal 成立的两个极端情况:
// 1. a 和 b 都没有 Equal 中的 labels, 成立
// 2. a 和 b 都有 Equal 中的 labels, 且都为空值, 成立
// 关于抑制不生效的极端情况:
// 1. a 同时满足 SourceMatchers, TargetMatchers, b 同时满足 SourceMatchers, TargetMatchers, 且 Equal 成立, 不生效
// 抑制不生效的极端情况是为了避免告警的自抑制
// 所以，告警写入阶段 Inhibitor 会通过 Sub 的方式监听新的 alert 并判断 source 侧是否匹配,
// 匹配的话表示这个 alert 可能会抑制其他的 alert, 就会被缓存起来
// 在 Inhibitor 对应的 MuteStage 中取出来检查
func (ih *Inhibitor) Mutes(lset model.LabelSet) bool {
fp := lset.Fingerprint()
// 检查内存中所有 rule 是否匹配 lset
for _, r := range ih.rules {
// target 不匹配就没必要计算了
// 因为我们就是为了抑制 target
if !r.TargetMatchers.Matches(lset) {
continue
}
// target 匹配就检查 source, 如果 source 也匹配
// 那么就需要排除两端都匹配的情况
if inhibitedByFP, eq := r.hasEqual(lset, r.SourceMatchers.Matches(lset)); eq {
ih.marker.SetInhibited(fp, inhibitedByFP.String())
return true
}
}
// 这个位置没传 ids, 那么这个 alert 被置为 "active"
ih.marker.SetInhibited(fp)
return false
}
// 调用这个函数之前, 被检查 alert 已经满足了规则的 target,
// 而 scache 中的 alert 已经满足了规则的 source
// 剩下要确认的是:
// scache 中的 alert 有没有标签和被检查 alert 标签一致的,
// 再避免 alert 自我抑制的场景就可以了
func (r *InhibitRule) hasEqual(lset model.LabelSet, excludeTwoSidedMatch bool) (model.Fingerprint, bool) {
Outer:
for _, a := range r.scache.List() {
// The cache might be stale and contain resolved alerts.
if a.Resolved() {
continue
}
// 检查规则标签
for n := range r.Equal {
if a.Labels[n] != lset[n] {
continue Outer
}
}
// a 在加入 r.scache 的时候已经满足了 r.Source, 如果再通过 target 检查, 那么 scache 中的这个 a 同时满足 source 和 target
// 而 excludeTwoSidedMatch 如果为 true, 表示当前 dispatcher 处理的 alert 在 source 和 target 都满足
// 所以这个条件变成了:
// 如果被检查的 alert 标签还和 a 标签相同, 即抑制规则生效, 而且 a 和被检查的 alert 都同时满足 source 和 target,
// 就忽略 a 对被检查 alert 的抑制, 这里防止了一个告警自己抑制自己情况
if excludeTwoSidedMatch && r.TargetMatchers.Matches(a.Labels) {
continue Outer
}
// 出现一个抑制生效, 剩下的就不继续检查
return a.Fingerprint(), true
}
return model.Fingerprint(0), false
}
// 这个就是 Silencer 实现的 Muter 接口
func (s *Silencer) Mutes(lset model.LabelSet) bool {
fp := lset.Fingerprint()
activeIDs, pendingIDs, markerVersion, _ := s.marker.Silenced(fp)
var (
err error
allSils []*pb.Silence
newVersion = markerVersion
)
// 找出现在正在生效的静默规则
// 用来判断当前 alerts 哪些需要被静默掉
// version 相同表示 fp 标记时的 silences 到现在没有新的静默规则加入
if markerVersion == s.silences.Version() {
totalSilences := len(activeIDs) + len(pendingIDs)
if totalSilences == 0 {
return false
}
allIDs := append(append(make([]string, 0, totalSilences), activeIDs...), pendingIDs...)
allSils, _, err = s.silences.Query(
// 静默规则是用户在 web 端写入
// 这个位置使用 ids 和两种状态来过滤出需要判断的静默规则
// 这个 query 的封装也很特别，我后面会在golang代码设计的文章中聊
QIDs(allIDs...),
QState(types.SilenceStateActive, types.SilenceStatePending),
)
} else {
allSils, newVersion, err = s.silences.Query(
QState(types.SilenceStateActive, types.SilenceStatePending),
QMatches(lset),
)
}
if err != nil {
level.Error(s.logger).Log("msg", "Querying silences failed, alerts might not get silenced correctly", "err", err)
}
if len(allSils) == 0 {
s.marker.SetSilenced(fp, newVersion, nil, nil)
return false
}
activeIDs, pendingIDs = nil, nil
now := s.silences.now()
// 这里仅根据搜索结果的数量就判断是否需要静默当前的 alert
// 并没有计算 silence 的时间区间和当前时间是否重合,
// 因为 silence 有效的计算是在 Maintenance 过程中使用 GC 来维护的
// 所以匹配的一定是现在就生效的
for _, sil := range allSils {
switch getState(sil, now) {
case types.SilenceStatePending:
pendingIDs = append(pendingIDs, sil.Id)
case types.SilenceStateActive:
activeIDs = append(activeIDs, sil.Id)
default:
// Do nothing, silence has expired in the meantime.
}
}
level.Debug(s.logger).Log(
"msg", "determined current silences state",
"now", now,
"total", len(allSils),
"active", len(activeIDs),
"pending", len(pendingIDs),
)
sort.Strings(activeIDs)
sort.Strings(pendingIDs)
// activeIDs 为空且没有 inhibitBy 的话, fp 仍然会是 active 的
// pendingIDs 不会对 fp 的状态有影响
s.marker.SetSilenced(fp, newVersion, activeIDs, pendingIDs)
return len(activeIDs) > 0
}
复制代码到这里，流水线的大致情况就介绍的差不多了，总结一下：
- 先约定
Stage接口， - 再定义一些控制流程的
Stage，比如RoutingStage，MultiStage，FanoutStage等 - 然后根据需要定义一些对
alerts做真正处理的的Stage，比如InhibitStage，SilenceStage，TimeMuteStage等 - 最后把这些处理
alerts的Stage使用流程控制的Stage进行编排，就成了流水线
近期评论