open-falcon 聚合器aggregator代碼解析
總結:aggregator聚合器就是從falcon_portal.cluster表中取出用戶在頁面上配置的表達式,然後解析後,通過api拿到對應機器組的所有機器,通過api查詢graph數據算出一個值重新打回transfer作爲一個新的點。
- 定時從db中拿出所有的聚合器配置放到一個map中
- 第一次啓動時遍歷聚合器map生成workers map 這兩個map的key都是id+updatetime
- 同時下一次拿出db生成map 對workers這個map進行增量更新 和刪除操作刪除是通過 worker.Quit chan通信的
- workers這個map 通過 ticker跑cron 運行WorkerRun這個方法
- WorkerRun這個方法解析分子分母的配置
- 調用api 根據grp_id拿出所有機器列表
- 調用graph的last接口拿出所有endpoint的counter 的值然後進行計算
- 計算後重新打回 一個線程安全的雙向鏈表隊列
- 另外一個goroutine異步pop隊列中的值發生給 transfer的http接口(不是給agent用的rpc接口)
- 機器量很多時獲取機器列表和查詢最新的值都是瓶頸
- 我在想如果直接在transfer中直接做數據的聚合速度上不存在瓶頸
下面我們來看下代碼:
- main.go中核心的兩個地方
//查詢db 調api算值 push 到push的隊列中 go cron.UpdateItems() //從push隊列push到transfer sender.StartSender()
2.看下go cron.UpdateItems()
func updateItems() { //從db中查詢出結果 items, err := db.ReadClusterMonitorItems() if err != nil { return } //對比key(id+uptime),將已經變更的項刪除 deleteNoUseWorker(items) //啓動新的worker createWorkerIfNeed(items) } //看下這個讀db的func func ReadClusterMonitorItems() (M map[string]*g.Cluster, err error){ ...... /*看到這個funcreturn的是個map key是 每個聚合項的id和他更新時間的字符串 value 就是Cluster結構體指針 type Cluster struct { Id int64 GroupId int64 Numerator string Denominator string Endpoint string Metric string Tags string DsType string Step int LastUpdate time.Time } */ M[fmt.Sprintf("%d%v", c.Id, c.LastUpdate)] = &c return M, err }
3.看下 deleteNoUseWorker 和createWorkerIfNeed 這兩個func都是圍繞 Worker這個struct的進行增刪
func deleteNoUseWorker(m map[string]*g.Cluster) { del := []string{} for key, worker := range Workers { //遍歷已經創建的work,如果key在新的map中沒有了說明這條記錄在db中被更改或刪除了 //所以刪掉它 給Workers這個map縮容 if _, ok := m[key]; !ok { //將worker 中的Quit chan關閉 會調用ticker.stop 真正關閉 worker.Drop() del = append(del, key) } } for _, key := range del { delete(Workers, key) } } func createWorkerIfNeed(m map[string]*g.Cluster) { for key, item := range m { if _, ok := Workers[key]; !ok { //如果配置中step小於0 丟棄這條 if item.Step <= 0 { log.Println("[W] invalid cluster(step <= 0):", item) continue } //初始化worker worker := NewWorker(item) Workers[key] = worker worker.Start() } } }
4. 看下Worker這個結構體包含三個域
- ticker作爲一個計時器實現類似cron的功能每隔一段時間執行一次Start 中的func
- ClusterItem作爲每個聚合器的配置
- Quit是一個chan用來外部關閉 key在新的map中沒有了說明這條記錄在db中被更改或刪除了
type Worker struct { Ticker *time.Ticker ClusterItem *g.Cluster Quit chan struct{} } func NewWorker(ci *g.Cluster) Worker { w := Worker{} w.Ticker = time.NewTicker(time.Duration(ci.Step) * time.Second) w.Quit = make(chan struct{}) w.ClusterItem = ci return w } func (this Worker) Start() { go func() { for { select { case <-this.Ticker.C: WorkerRun(this.ClusterItem) case <-this.Quit: if g.Config().Debug { log.Println("[I] drop worker", this.ClusterItem) } this.Ticker.Stop() return } } }() } func (this Worker) Drop() { close(this.Quit) } var Workers = make(map[string]Worker)
到這裏我們已經看明白聚合器的流程了:
- 定時從db中拿出所有的聚合器配置放到一個map中
- 第一次啓動時遍歷聚合器map生成workers map 這兩個map的key都是id+updatetime
- 同時下一次拿出db生成map 對workers這個map進行增量更新 和刪除操作刪除是通過 worker.Quit chan通信的
- workers這個map 通過 ticker跑cron 運行WorkerRun這個方法
5.下面看下最重要的方法 WorkerRun
func WorkerRun(item *g.Cluster) { debug := g.Config().Debug /* Numerator代表分子 例如 $(cpu.user)+$(cpu.system) 代表求cpu.user和cpu.system的和 Denominator代表分母 例如 $# 代表所有機器 */ //cleanParam去除\r等字符 numeratorStr := cleanParam(item.Numerator) denominatorStr := cleanParam(item.Denominator) //判斷分子分母是否合法 if !expressionValid(numeratorStr) || !expressionValid(denominatorStr) { log.Println("[W] invalid numerator or denominator", item) return } //判斷分子分母是否需要計算 needComputeNumerator := needCompute(numeratorStr) needComputeDenominator := needCompute(denominatorStr) //如果分子分母都不需要計算就不需要用到聚合器了 if !needComputeNumerator && !needComputeDenominator { log.Println("[W] no need compute", item) return } //比如分子是這樣的: "($(cpu.busy)+$(cpu.idle)-$(cpu.nice))>80" //那麼parse的返回值爲 [cpu.busy cpu.idle cpu.nice] [+ -] >80 numeratorOperands, numeratorOperators, numeratorComputeMode := parse(numeratorStr, needComputeNumerator) denominatorOperands, denominatorOperators, denominatorComputeMode := parse(denominatorStr, needComputeDenominator) if !operatorsValid(numeratorOperators) || !operatorsValid(denominatorOperators) { log.Println("[W] operators invalid", item) return } /*add retry for gethostname bygid 這裏源碼是動過sdk根據group_id查找組裏面機器列表 這裏我進行了兩點優化: 1.sdk調用時沒有加重試,http失敗導致這次沒有get到機器所以這個點就不算了導致斷點 2.原來的接口在機器量超過1k時就效率就會很慢 2w+機器需要8s,看了代碼是用orm進行了多次查詢而且附帶了很多別的信息 這裏我只需要group_id對應endpoint_list所以我寫了一個新的接口用一條raw_sql進行查詢 測試2w+的機器0.2s就能返回 */ retry_limit :=3 r_s :=0 var hostnames []string for r_s <retry_limit{ hostnames_tmp, err_tmp := sdk.HostnamesByID(item.GroupId) if err_tmp != nil { log.Println("[E] get hostlist err",err_tmp) r_s+=1 time.Sleep(time.Second) }else{ hostnames = hostnames_tmp break } } //沒有機器當然不用算了 if len(hostnames)==0{ log.Println("[E] get 0 record hostname item:",item) return } now := time.Now().Unix() /*這裏是調用graph/lastpoint這個api 查詢最近一個點的數據 1.機器是上面查到的主機列表 2.counter這裏做了合併 把所有要查的metirc都放在一個請求裏面查詢了 3.查詢的時候在api那邊做了for循環 逐個item查詢 估計這裏也會拖慢速度 4.查完之後計算下值推到發送隊列 */ valueMap, err := queryCounterLast(numeratorOperands, denominatorOperands, hostnames, now-int64(item.Step*2), now) if err != nil { log.Println("[E] get queryCounterLast", err, item) return } .......... sender.Push(item.Endpoint, item.Metric, item.Tags, numerator/denominator, item.DsType, int64(item.Step)) }
6.最後看下發送的代碼
- MetaDataQueue是個線程安全的雙向鏈表
- 上面說的WorkerRun方法中會將轉化好的監控項數據PushFront入鏈表
- startSender這個goroutine 每200毫秒會將隊列中的數據取出發送到transfer的http接口
func Push(endpoint, metric, tags string, val interface{}, counterType string, step_and_ts ...int64) { md := MakeMetaData(endpoint, metric, tags, val, counterType, step_and_ts...) MetaDataQueue.PushFront(md) } const LIMIT = 200 var MetaDataQueue = NewSafeLinkedList() var PostPushUrl string var Debug bool func StartSender() { go startSender() } func startSender() { for { L := MetaDataQueue.PopBack(LIMIT) if len(L) == 0 { time.Sleep(time.Millisecond * 200) continue } err := PostPush(L) if err != nil { log.Println("[E] push to transfer fail", err) } } }
歡迎關注我們的微信公衆號,每天學習Go知識