open-falcon-agent源码学习

最近学习falcon,看了源码和极客学院的视频解析,画了调用结构、关系,对主要的代码进行了注释

代码地址:https://github.com/beyondskyw...

标签(空格分隔): falcon go

监控数据

• 机器性能指标:cpu,mem,网卡,磁盘......

• 业务监控

• 开源软件状态:Nginx,Redis,MySQL

• snmp采集网络设备指标

设计原理

• 自发现采集值

• 不同类型数据采集分不同goroutine

• 进程和端口通过用户配置进行监控

配置文件

• hostname和ip默认留空,agent自动探测

• hbs和transfer都是配置其rpc地址

• collector网卡采集前缀

• ignore为true时取消上报

组织结构

• cron:间隔执行的代码,即定时任务

• funcs:信息采集

• g:全局数据结构

• http:简单的dashboard的server,获取单机监控指标数据

• plugins:插件处理机制

• public:静态资源文件

心跳机制

• 了解agent、plugin版本信息,方便升级

• 获取监听的进程和端口

• 获取本机执行的插件列表

与HBS、Transfer交互

调用关系

代码解读

• main入口

go cron.InitDataHistory()
// 上报本机状态
cron.ReportAgentStatus()
// 同步插件
cron.SyncMinePlugins()
// 同步监控端口、路径、进程和URL
cron.SyncBuiltinMetrics()
// 后门调试agent,允许执行shell指令的ip列表
cron.SyncTrustableIps()
// 开始数据次采集
cron.Collect()
// 启动dashboard server
go http.Start()

• ReportAgentStatus:汇报agent本身状态

// 判断hbs配置是否正常,正常则上报agent状态
if g.Config().Heartbeat.Enabled && g.Config().Heartbeat.Addr != "" {
 // 根据配置的interval间隔上报信息
 go reportAgentStatus(time.Duration(g.Config().Heartbeat.Interval) * time.Second)
}
func reportAgentStatus(interval time.Duration) {
 for {
 // 获取hostname, 出错则错误赋值给hostname
 hostname, err := g.Hostname()
 if err != nil {
 hostname = fmt.Sprintf("error:%s", err.Error())
 }
 // 请求发送信息
 req := model.AgentReportRequest{
 Hostname: hostname,
 IP: g.IP(),
 AgentVersion: g.VERSION,
 // 通过shell指令获取plugin版本,能否go实现
 PluginVersion: g.GetCurrPluginVersion(),
 }
 var resp model.SimpleRpcResponse
 // 调用rpc接口
 err = g.HbsClient.Call("Agent.ReportStatus", req, &resp)
 if err != nil || resp.Code != 0 {
 log.Println("call Agent.ReportStatus fail:", err, "Request:", req, "Response:", resp)
 }
 time.Sleep(interval)
 }
}

• SyncMinePlugins:同步插件

func syncMinePlugins() {
 var (
 timestamp int64 = -1
 pluginDirs []string
 )
 duration := time.Duration(g.Config().Heartbeat.Interval) * time.Second
 for {
 time.Sleep(duration)
 hostname, err := g.Hostname()
 if err != nil {
 continue
 }
 req := model.AgentHeartbeatRequest{
 Hostname: hostname,
 }
 var resp model.AgentPluginsResponse
 // 调用rpc接口,返回plugin
 err = g.HbsClient.Call("Agent.MinePlugins", req, &resp)
 if err != nil {
 log.Println("ERROR:", err)
 continue
 }
 // 保证时间顺序正确
 if resp.Timestamp <= timestamp {
 continue
 }
 pluginDirs = resp.Plugins
 // 存放时间保证最新
 timestamp = resp.Timestamp
 if g.Config().Debug {
 log.Println(&resp)
 }
 // 无插件则清空plugin
 if len(pluginDirs) == 0 {
 plugins.ClearAllPlugins()
 }
 desiredAll := make(map[string]*plugins.Plugin)
 // 读取所有plugin
 for _, p := range pluginDirs {
 underOneDir := plugins.ListPlugins(strings.Trim(p, "/"))
 for k, v := range underOneDir {
 desiredAll[k] = v
 }
 }
 // 停止不需要的插件,启动增加的插件
 plugins.DelNoUsePlugins(desiredAll)
 plugins.AddNewPlugins(desiredAll)
 }
}

• SyncBuiltinMetrics:同步内置metric,包括端口、目录和进程信息

func syncBuiltinMetrics() {
 var timestamp int64 = -1
 var checksum string = "nil"
 duration := time.Duration(g.Config().Heartbeat.Interval) * time.Second
 for {
 time.Sleep(duration)
 // 监控端口、目录大小、进程
 var ports = []int64{}
 var paths = []string{}
 var procs = make(map[string]map[int]string)
 var urls = make(map[string]string)
 hostname, err := g.Hostname()
 if err != nil {
 continue
 }
 req := model.AgentHeartbeatRequest{
 Hostname: hostname,
 Checksum: checksum,
 }
 var resp model.BuiltinMetricResponse
 err = g.HbsClient.Call("Agent.BuiltinMetrics", req, &resp)
 if err != nil {
 log.Println("ERROR:", err)
 continue
 }
 if resp.Timestamp <= timestamp {
 continue
 }
 if resp.Checksum == checksum {
 continue
 }
 timestamp = resp.Timestamp
 checksum = resp.Checksum
 for _, metric := range resp.Metrics {
 if metric.Metric == g.URL_CHECK_HEALTH {
 arr := strings.Split(metric.Tags, ",")
 if len(arr) != 2 {
 continue
 }
 url := strings.Split(arr[0], "=")
 if len(url) != 2 {
 continue
 }
 stime := strings.Split(arr[1], "=")
 if len(stime) != 2 {
 continue
 }
 if _, err := strconv.ParseInt(stime[1], 10, 64); err == nil {
 urls[url[1]] = stime[1]
 } else {
 log.Println("metric ParseInt timeout failed:", err)
 }
 }
 // {metric: net.port.listen, tags: port=22}
 if metric.Metric == g.NET_PORT_LISTEN {
 arr := strings.Split(metric.Tags, "=")
 if len(arr) != 2 {
 continue
 }
 if port, err := strconv.ParseInt(arr[1], 10, 64); err == nil {
 ports = append(ports, port)
 } else {
 log.Println("metrics ParseInt failed:", err)
 }
 continue
 }
 // metric: du.bs tags: path=/home/works/logs
 // du -bs /home/works/logs
 if metric.Metric == g.DU_BS {
 arr := strings.Split(metric.Tags, "=")
 if len(arr) != 2 {
 continue
 }
 paths = append(paths, strings.TrimSpace(arr[1]))
 continue
 }
 //mereic: proc.num tags: name=crond
 //或者metric: proc.num tags: cmdline=cfg.json
 if metric.Metric == g.PROC_NUM {
 arr := strings.Split(metric.Tags, ",")
 tmpMap := make(map[int]string)
 for i := 0; i < len(arr); i++ {
 if strings.HasPrefix(arr[i], "name=") {
 tmpMap[1] = strings.TrimSpace(arr[i][5:])
 } else if strings.HasPrefix(arr[i], "cmdline=") {
 tmpMap[2] = strings.TrimSpace(arr[i][8:])
 }
 }
 procs[metric.Tags] = tmpMap
 }
 }
 g.SetReportUrls(urls)
 g.SetReportPorts(ports)
 g.SetReportProcs(procs)
 g.SetDuPaths(paths)
 }
}

• SyncTrustableIps:同步可信IP列表
  请求获取远程访问执行shell命令的IP白名单,在通过http/run.go调用shell命令是会判断请求IP是否可信

func syncTrustableIps() {
 duration := time.Duration(g.Config().Heartbeat.Interval) * time.Second
 for {
 time.Sleep(duration)
 var ips string
 err := g.HbsClient.Call("Agent.TrustableIps", model.NullRpcRequest{}, &ips)
 if err != nil {
 log.Println("ERROR: call Agent.TrustableIps fail", err)
 continue
 }
 // 设置到本地可信IP列表
 g.SetTrustableIps(ips)
 }
}

• FuncsAndInterval:拆分不同的采集函数集,方便通过不同goroutine运行

// 间隔internal时间执行fs中的函数
type FuncsAndInterval struct {
 Fs []func() []*model.MetricValue
 Interval int
}
var Mappers []FuncsAndInterval
// 根据调用指令类型和是否容易被挂起而分类(通过不同的goroutine去执行,避免相互之间的影响)
func BuildMappers() {
 interval := g.Config().Transfer.Interval
 Mappers = []FuncsAndInterval{
 FuncsAndInterval{
 Fs: []func() []*model.MetricValue{
 AgentMetrics,
 CpuMetrics,
 NetMetrics,
 KernelMetrics,
 LoadAvgMetrics,
 MemMetrics,
 DiskIOMetrics,
 IOStatsMetrics,
 NetstatMetrics,
 ProcMetrics,
 UdpMetrics,
 },
 Interval: interval,
 },
 // 容易出问题
 FuncsAndInterval{
 Fs: []func() []*model.MetricValue{
 DeviceMetrics,
 },
 Interval: interval,
 },
 // 调用相同指令
 FuncsAndInterval{
 Fs: []func() []*model.MetricValue{
 PortMetrics,
 SocketStatSummaryMetrics,
 },
 Interval: interval,
 },
 FuncsAndInterval{
 Fs: []func() []*model.MetricValue{
 DuMetrics,
 },
 Interval: interval,
 },
 FuncsAndInterval{
 Fs: []func() []*model.MetricValue{
 UrlMetrics,
 },
 Interval: interval,
 },
 }
}

• Colleet:配置信息读取,读取Mapper中的FuncsAndInterval,根据func调用采集函数,采集所有信息(并非先过滤采集项),从所有采集到的数据中过滤ignore的项,并上报到transfer。

func Collect() {
 // 配置信息判断
 if !g.Config().Transfer.Enabled {
 return
 }
 if len(g.Config().Transfer.Addrs) == 0 {
 return
 }
 // 读取mapper中的FuncsAndInterval集,并通过不同的goroutine运行
 for _, v := range funcs.Mappers {
 go collect(int64(v.Interval), v.Fs)
 }
}
// 间隔采集信息
func collect(sec int64, fns []func() []*model.MetricValue) {
 // 启动断续器,间隔执行
 t := time.NewTicker(time.Second * time.Duration(sec)).C
 for {
 <-t
 hostname, err := g.Hostname()
 if err != nil {
 continue
 }
 mvs := []*model.MetricValue{}
 // 读取忽略metric名单
 ignoreMetrics := g.Config().IgnoreMetrics
 // 从funcs的list中取出每个采集函数
 for _, fn := range fns {
 // 执行采集函数
 items := fn()
 if items == nil {
 continue
 }
 if len(items) == 0 {
 continue
 }
 // 读取采集数据,根据忽略的metric忽略部分采集数据
 for _, mv := range items {
 if b, ok := ignoreMetrics[mv.Metric]; ok && b {
 continue
 } else {
 mvs = append(mvs, mv)
 }
 }
 }
 // 获取上报时间
 now := time.Now().Unix()
 // 设置上报采集项的间隔、agent主机、上报时间
 for j := 0; j < len(mvs); j++ {
 mvs[j].Step = sec
 mvs[j].Endpoint = hostname
 mvs[j].Timestamp = now
 }
 // 调用transfer发送采集数据
 g.SendToTransfer(mvs)
 }
}

• 采集信息结构

type MetricValue struct {
 Endpoint string // 主机名
 Metric string // 信息标识cpu.idle、mem.memtotal等
 Value interface{} // 采集结果
 Step int64 // 该项上报间隔
 Type string // GAUGE或COUNTER
 Tags string // 配置报警策略
 Timestamp int64 // 此次上报时间
}

• 采集信息组成metricValue结构

func NewMetricValue(metric string, val interface{}, dataType string, tags ...string) *model.MetricValue {
 mv := model.MetricValue{
 Metric: metric,
 Value: val,
 Type: dataType,
 }
 size := len(tags)
 if size > 0 {
 mv.Tags = strings.Join(tags, ",")
 }
 return &mv
}
// 原值类型
func GaugeValue(metric string, val interface{}, tags ...string) *model.MetricValue {
 return NewMetricValue(metric, val, "GAUGE", tags...)
}
// 计数器类型
func CounterValue(metric string, val interface{}, tags ...string) *model.MetricValue {
 return NewMetricValue(metric, val, "COUNTER", tags...)
}

• rpc组件

// 简单封装rpc.Cilent
type SingleConnRpcClient struct {
 sync.Mutex
 rpcClient *rpc.Client
 RpcServer string
 Timeout time.Duration
}
// 关闭rpc
func (this *SingleConnRpcClient) close() {
 if this.rpcClient != nil {
 this.rpcClient.Close()
 this.rpcClient = nil
 }
}
// 保证rpc存在,为空则重新创建, 如果server宕机, 死循环????
func (this *SingleConnRpcClient) insureConn() {
 if this.rpcClient != nil {
 return
 }
 var err error
 var retry int = 1
 for {
 if this.rpcClient != nil {
 return
 }
 // 根据timeout和server地址去连接rpc的server
 this.rpcClient, err = net.JsonRpcClient("tcp", this.RpcServer, this.Timeout)
 if err == nil {
 return
 }
 log.Printf("dial %s fail: %v", this.RpcServer, err)
 if retry > 6 {
 retry = 1
 }
 time.Sleep(time.Duration(math.Pow(2.0, float64(retry))) * time.Second)
 retry++
 }
}
// rpc client调用hbs函数
func (this *SingleConnRpcClient) Call(method string, args interface{}, reply interface{}) error {
 // 加锁保证一个agent只与server有一个连接,保证性能
 this.Lock()
 defer this.Unlock()
 // 保证rpc连接可用
 this.insureConn()
 timeout := time.Duration(50 * time.Second)
 done := make(chan error)
 go func() {
 err := this.rpcClient.Call(method, args, reply)
 done <- err
 }()
 // 超时控制
 select {
 case <-time.After(timeout):
 log.Printf("[WARN] rpc call timeout %v => %v", this.rpcClient, this.RpcServer)
 this.close()
 case err := <-done:
 if err != nil {
 this.close()
 return err
 }
 }
 return nil
}

• Transfer部件

// 定义transfer的rpcClient对应Map, transferClients读写锁
var (
 TransferClientsLock *sync.RWMutex = new(sync.RWMutex)
 TransferClients map[string]*SingleConnRpcClient = map[string]*SingleConnRpcClient{}
)
// 发送数据到随机的transfer
func SendMetrics(metrics []*model.MetricValue, resp *model.TransferResponse) {
 rand.Seed(time.Now().UnixNano())
 // 随机transferClient发送数据,直到发送成功
 for _, i := range rand.Perm(len(Config().Transfer.Addrs)) {
 addr := Config().Transfer.Addrs[i]
 if _, ok := TransferClients[addr]; !ok {
 initTransferClient(addr)
 }
 if updateMetrics(addr, metrics, resp) {
 break
 }
 }
}
// 初始化addr对应的transferClient
func initTransferClient(addr string) {
 TransferClientsLock.Lock()
 defer TransferClientsLock.Unlock()
 TransferClients[addr] = &SingleConnRpcClient{
 RpcServer: addr,
 Timeout: time.Duration(Config().Transfer.Timeout) * time.Millisecond,
 }
}
// 调用rpc接口发送metric
func updateMetrics(addr string, metrics []*model.MetricValue, resp *model.TransferResponse) bool {
 TransferClientsLock.RLock()
 defer TransferClientsLock.RUnlock()
 err := TransferClients[addr].Call("Transfer.Update", metrics, resp)
 if err != nil {
 log.Println("call Transfer.Update fail", addr, err)
 return false
 }
 return true
}

• 采集插件同步

// 插件信息: 路径、修改时间、运行周期(来自plugin插件)
type Plugin struct {
 FilePath string
 MTime int64
 Cycle int
}
// 插件map和调度器map
var (
 Plugins = make(map[string]*Plugin)
 PluginsWithScheduler = make(map[string]*PluginScheduler)
)
// 删除不需要的plugin
func DelNoUsePlugins(newPlugins map[string]*Plugin) {
 for currKey, currPlugin := range Plugins {
 newPlugin, ok := newPlugins[currKey]
 if !ok || currPlugin.MTime != newPlugin.MTime {
 deletePlugin(currKey)
 }
 }
}
// 添加同步时增加的plugin
func AddNewPlugins(newPlugins map[string]*Plugin) {
 for fpath, newPlugin := range newPlugins {
 // 去除重复插件
 if _, ok := Plugins[fpath]; ok && newPlugin.MTime == Plugins[fpath].MTime {
 continue
 }
 // 为新添加的插件新建调度器
 Plugins[fpath] = newPlugin
 sch := NewPluginScheduler(newPlugin)
 PluginsWithScheduler[fpath] = sch
 // 启动plugin调度
 sch.Schedule()
 }
}
func ClearAllPlugins() {
 for k := range Plugins {
 deletePlugin(k)
 }
}
func deletePlugin(key string) {
 v, ok := PluginsWithScheduler[key]
 if ok {
 // 暂停调度plugin
 v.Stop()
 delete(PluginsWithScheduler, key)
 }
 delete(Plugins, key)
}

• 插件调度策略

// 持续间隔执行plugin
type PluginScheduler struct {
 Ticker *time.Ticker
 Plugin *Plugin
 Quit chan struct{}
}
// 根据plugin创建新的schedule
func NewPluginScheduler(p *Plugin) *PluginScheduler {
 scheduler := PluginScheduler{Plugin: p}
 scheduler.Ticker = time.NewTicker(time.Duration(p.Cycle) * time.Second)
 scheduler.Quit = make(chan struct{})
 return &scheduler
}
// plugin调度,间隔执行PluginRun,除非收到quit消息
func (this *PluginScheduler) Schedule() {
 go func() {
 for {
 select {
 case <-this.Ticker.C:
 PluginRun(this.Plugin)
 case <-this.Quit:
 this.Ticker.Stop()
 return
 }
 }
 }()
}
// 停止plugin调度
func (this *PluginScheduler) Stop() {
 close(this.Quit)
}
// 执行插件,读取插件运行返回数据并上报transfer
func PluginRun(plugin *Plugin) {
 timeout := plugin.Cycle*1000 - 500
 fpath := filepath.Join(g.Config().Plugin.Dir, plugin.FilePath)
 if !file.IsExist(fpath) {
 log.Println("no such plugin:", fpath)
 return
 }
 debug := g.Config().Debug
 if debug {
 log.Println(fpath, "running...")
 }
 cmd := exec.Command(fpath)
 var stdout bytes.Buffer
 cmd.Stdout = &stdout
 var stderr bytes.Buffer
 cmd.Stderr = &stderr
 cmd.Start()
 err, isTimeout := sys.CmdRunWithTimeout(cmd, time.Duration(timeout)*time.Millisecond)
 errStr := stderr.String()
 if errStr != "" {
 logFile := filepath.Join(g.Config().Plugin.LogDir, plugin.FilePath+".stderr.log")
 if _, err = file.WriteString(logFile, errStr); err != nil {
 log.Printf("[ERROR] write log to %s fail, error: %s\n", logFile, err)
 }
 }
 if isTimeout {
 // has be killed
 if err == nil && debug {
 log.Println("[INFO] timeout and kill process", fpath, "successfully")
 }
 if err != nil {
 log.Println("[ERROR] kill process", fpath, "occur error:", err)
 }
 return
 }
 if err != nil {
 log.Println("[ERROR] exec plugin", fpath, "fail. error:", err)
 return
 }
 // exec successfully
 data := stdout.Bytes()
 if len(data) == 0 {
 if debug {
 log.Println("[DEBUG] stdout of", fpath, "is blank")
 }
 return
 }
 var metrics []*model.MetricValue
 err = json.Unmarshal(data, &metrics)
 if err != nil {
 log.Printf("[ERROR] json.Unmarshal stdout of %s fail. error:%s stdout: \n%s\n", fpath, err, stdout.String())
 return
 }
 g.SendToTransfer(metrics)
}

本文来自:Segmentfault

感谢作者:SkyWay

查看原文:open-falcon-agent源码学习

• 编辑
• 预览