Open-falcon transfer发送judge的流程优化监控工具

【Open-falcon transfer发送judge的流程优化】transfer组件通过一致性hash分配，使用RPC接口将数据发送给judge组件，由judge组件做告警判定。
在transfer中，对于没有配置告警策略的指标，没有必要存储到judgeQueue，也没有必要发送给judge节点。
Open-falcon的处理是由transfer把所有的指标无脑发给judge，由judge判断该指标是否有告警规则关联。
nightingale作为升级版的Open-falcon，它的处理较为巧妙，它在transfer组件中就做了判断，若指标没有告警规则关联，则就无需存入judgeQueue，也不会发送到judge了，大大减少了网络发送数据量。
基于此，可对Open-falcon的transfer-->judge做一定的流程优化。
1. Open-falcon的transfer-->judge transfer将所有的指标无脑发送给judge，由judge做告警规则的判定；
看一下judge的代码，g.FilterMap中存储了metric及其告警规则信息，g.FilterMap的信息由judge请求hbs获得：

//modules/judge/rpc/receiver.go func (this *Judge) Send(items []*model.JudgeItem, resp *model.SimpleRpcResponse) error { remain := g.Config().Remain now := time.Now().Unix() for _, item := range items { exists := g.FilterMap.Exists(item.Metric)//过滤有metric的数据 if !exists { continue } pk := item.PrimaryKey() store.HistoryBigMap[pk[0:2]].PushFrontAndMaintain(pk, item, remain, now) } return nil }

流程如下图所示：

文章图片

2. nightingale的transfer-->judge nightingale在transfer这一层就把无关告警规则的metric都过滤掉了，不再发送给judge。
Transfer接收agent指标的代码入口：

//agent发送数据到transfer，transfer存储到queue // src/modules/transfer/rpc/push.go func (t *Transfer) Push(args []*dataobj.MetricValue, reply *dataobj.TransferResp) error { start := time.Now()items := make([]*dataobj.MetricValue, 0) for _, v := range args { if err := v.CheckValidity(start.Unix()); err != nil { ...... continue } items = append(items, v) } ...... if backend.Config.Enabled { backend.Push2JudgeSendQueue(items) }reply.Total = len(args) reply.Latency = (time.Now().UnixNano() - start.UnixNano()) / 1000000 return nil }

筛选metric到judgeQueue的代码：

先比对endpoint/metric是否关联告警策略；
再对比tag是否匹配；
若两者都匹配，则放入judgeQueue;

// src/modules/transfer/backend/judge.go func Push2JudgeSendQueue(items []*dataobj.MetricValue) { errCnt := 0 for _, item := range items { //判断是否有关联的endpoint/metric key := str.PK(item.Metric, item.Endpoint) stras := cache.StraMap.GetByKey(key)for _, stra := range stras { //是否tag匹配 if !TagMatch(stra.Tags, item.TagsMap) { continue } judgeItem := &dataobj.JudgeItem{ Endpoint:item.Endpoint, Metric:item.Metric, Value:item.Value, Timestamp: item.Timestamp, DsType:item.CounterType, Tags:item.Tags, TagsMap:item.TagsMap, Step:int(item.Step), Sid:stra.Id, Extra:item.Extra, }q, exists := JudgeQueues.Get(stra.JudgeInstance) q.PushFront(judgeItem) } } stats.Counter.Set("judge.queue.err", errCnt) }

cache.StraMap是transfer定期从mon-api组件查询得到的；
流程图如下：