共计 5130 个字符，预计需要花费 13 分钟才能阅读完成。
聚合在监控中的作用
- 简单来说: 需要将分散的大量监控数据按照一定的维度 (idc/service) 及一定的算法 (avg/sum/max/min/quantile 分位) 得到一个结果值
- 比如: 想知道服务 a 下面的 100 台机器的平均
- 又比如: 想知道查询接口 b 的 99 分位延迟值
open-falcon 原版聚合器
aggregator 介绍:
- aggregator 聚合器就是从 falcon_portal.cluster 表中取出用户在页面上配置的表达式, 然后解析后，通过 api 拿到对应机器组的所有机器，通过 api 查询 graph 数据算出一个值重新打回 transfer 作为一个新的点。
- 具体介绍可以看我的文章: open-falcon 聚合器 aggregator 代码解析
aggregator 问题:
endpoint 多的聚合断点问题
造成断点的原因有两个：
- 原来的接口在机器量超过 1k 时就效率就会很慢 2w+endpoint 需要 8s, 看了代码是用 orm 进行了多次查询而且附带了很多别的信息, 这里我只需要 group_id 对应 endpoint_list 所以我写了一个新的接口用一条 raw_sql 进行查询 HostnamesByID
- 数据查询 依赖 graph/lastpoint 这个 api 查询最近一个点的数据，如果 endpoint 多的时候 batch_query 会超时, 返回结果成功率也比较低
聚合器单点问题
这个是由于其代码架构导致的，如果单点聚合器挂了会导致数据断点
配置稍显复杂
基于以上问题我写了个新组件 polymetric
polymetric 代码
polymetric 介绍:
功能点说明
- 集群（sum,avg,max,min,tp50,tp90,tp99）指标及其 (_delta,_rate) 曲线
- 上述集群指标的同环比(grafana 图中默认为 avg 和 sum 两条)
- 实时 topk 和 bottomk 情况
- 动态获取过去一段段时间内分布在过高或过低区间次数（倒序）top20endpoint 曲线
- 动态获取过去一段时间内集群指标正态分布落在正常区间比率由低到 top10 时刻的集群指标分布图
架构说明
- 聚合策略按照 group_name + metric_name 保存在 db 中，hbs 从 db 中刷入 redis 集群中
- transfer 新增定时从 redis 集群更新 poly 策略到本地 map，目的是将 push 过来的 endpoint 直接按照 poly 策略 push 到 polymetric transfer 同步 redis 中的 poly 策略
- transfer 根据配置文件中的 poly 实例生成一致性哈希环，根据聚合策略哈希将要聚合的数据 push 到对应的 polymetric 实例上面 transfer rpc 对于 poly 的处理(agent rpc 中对于 polymetric 的处理)
- 这里还要注意一点: 需要将 counter 型的数据转化为 gauge 型所以需要在 map 中 hold 点
- 到这里 polymetric 实例就能源源不断的收到应该交给它聚合的数据，并将数据推入链表 polymetric rpc 接收 transfer 数据
- polymetric 启动 worker 处理数据 , 链表中 popall 数据, 排序后得到 slice 然后就可以轻松得出: max/min/sum/avg 和各个分位值
- 同时 max min 还需要知道是哪个 endpoint 提供的所以将数据推送到 kafka，写入 es 由 kibana 展示
- 同时将数据经由 pushgateway 打入 promethues 做 topk/ 同环比展示 push 到 promethues
- 异常分析: 将数据推送到异常分析组件做正态分布展示
- 数据处理主函数
func GeneralPolyMethods(Name string, Q *nlist.SafeListLimited) {Len := Q.Len()
item := Q.PopBackBy(Len)
count := len(item)
if count == 0 {return}
log.Infof("[GeneralPolyMethods]RunGroupPoly_called:Name:%s,len:%d", Name, count)
var dataList []float64
var numpList []SingleEnd
var sum, avg, max, min, tp50, tp90, tp99 float64
counterType := GAUGEType
// 为了给出 max、min 等极值对应的 endpoint
singMap := make(map[float64]string)
for _, i := range item {iF := i.(*cmodel.PolyRequest)
//if counterType == "" {
// counterType = iF.Type
//}
va := iF.Value.(float64)
endP := iF.EndPoint
t := SingleEnd{
Endpoint: endP,
Value: va,
}
numpList = append(numpList, t)
if singMap[va] == "" {singMap[va] = endP
}
sum += va
dataList = append(dataList, va)
}
realCount := len(dataList)
if realCount == 0 {return}
var pushSetp int64
pushSetp = PolyTimeStep
if realCount == 1 {sum = dataList[0]
avg = dataList[0]
max = dataList[0]
min = dataList[0]
tp50 = dataList[0]
tp90 = dataList[0]
tp99 = dataList[0]
} else {sort.Float64s(dataList)
max = dataList[realCount-1]
min = dataList[0]
avg = sum / float64(realCount)
tp50 = dataList[int(float64(realCount)*0.5)]
tp90 = dataList[int(float64(realCount)*0.95)]
tp99 = dataList[int(float64(realCount)*0.99)]
}
// 本地 map 做循环技术用
localDataMap := make(map[string]float64)
promeDataMap := make(map[string]float64)
localDataMap["sum"] = sum
localDataMap["avg"] = avg
localDataMap["max"] = max
localDataMap["min"] = min
localDataMap["tp50"] = tp50
localDataMap["tp90"] = tp90
localDataMap["tp99"] = tp99
names := strings.Split(Name, SEP)
polyType := names[0]
polyName := names[1]
metric := names[2]
endp := polyType + "_poly_" + polyName
log.Infof("poly_res:endp sum, avg, max, min, tp50, tp90, tp99", endp, sum, avg, max, min, tp50, tp90, tp99)
endNew := strings.Replace(endp, ".", "_", -1)
tagPre := "method="
//log.Infof("sum,avg,max,min,tp50,", sum, avg, max, min)
sender.Push(endNew, metric, tagPre+"sum", sum, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"avg", avg, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"max", max, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"min", min, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"tp50", tp50, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"tp90", tp90, counterType, int64(pushSetp))
sender.Push(endNew, metric, tagPre+"tp99", tp99, counterType, int64(pushSetp))
/*
根据内存中的值计算 rate 和 delta
*/
for k, v := range localDataMap {promeDataMap[k] = v
rate := 0.0
delta := 0.0
uniqueResultKey := endNew + metric + tagPre + k
if lastPoint, loaded := PolyHistoryDataMap.Load(uniqueResultKey); loaded {log.Debugf("[localDataMap_lastPoint] key,this_value,last_value,%+v,%+v,%+v", k, v, lastPoint)
lastP := lastPoint.(float64)
delta = v - lastP
if lastP == 0.0 {rate = 0.0} else {
//rate = delta / lastP * 100.0
rate = delta / lastP
}
}
// 本次计算完毕，更新 cache 中的值
PolyHistoryDataMap.Store(uniqueResultKey, v)
log.Debugf("[localDataMap] key,this_value,rate delta ,%+v,%+v,%+v,%+v", k, v, rate, delta)
sender.Push(endNew, metric+"_rate", tagPre+k, rate, counterType, int64(pushSetp))
sender.Push(endNew, metric+"_delta", tagPre+k, delta, counterType, int64(pushSetp))
promeDataMap[k+"_rate"] = rate
promeDataMap[k+"_delta"] = delta
}
// push to prome
if g.Config().Prome.Enabled {PushToProme(metric, polyName, promeDataMap)
}
// push 到 kafka
if kafka.KafkaAsyncProducer != nil {maxEnd := singMap[max]
minEnd := singMap[min]
tp50End := singMap[tp50]
tp90End := singMap[tp90]
tp99End := singMap[tp99]
AsyncPushKafka(polyType, polyName, maxEnd, metric, "max", max)
AsyncPushKafka(polyType, polyName, minEnd, metric, "min", min)
AsyncPushKafka(polyType, polyName, tp50End, metric, "tp50", tp50)
AsyncPushKafka(polyType, polyName, tp90End, metric, "tp90", tp90)
AsyncPushKafka(polyType, polyName, tp99End, metric, "tp99", tp99)
}
RpcCallNumpApi(metric, polyName, numpList)
////outlier check
//outlierStr := outlierCheck(dataList, singMap)
//outPoint := outlier.GrpOutlier{
// GrpName: polyName,
// PolyType: polyType,
// Counter: metric,
// Timestamp: time.Now().Unix(),
// Value: outlierStr,
//}
//saveOutlier2DB(&outPoint)
}
正文完
