网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

OpenFalcon源码分析(Agent组件)

2020-06-10 11:37:54 admin 1292

### 一 基本说明

> Agent VERSION "5.1.2"

* Agent源代码文件目录说明

文件目录说明

* 调用关系

调用关系

* 组件交互关系

组件交互关系

### 二 源码分析

> Main 入口函数

# 命令行参数解析
        cfg := flag.String("c", "cfg.json", "configuration file")
        version := flag.Bool("v", false, "show version")
        check := flag.Bool("check", false, "check collector")
    
        flag.Parse()
        
        # 命令行版本查看
        if *version {
            fmt.Println(g.VERSION)
            os.Exit(0)
        }
        
        # 命令行收集器检测
        if *check {
            funcs.CheckCollector()      #【参考相关详细分析】
            os.Exit(0)
        }
    
        # Agent配置文件解析，g.Config()可获取完整配置
        g.ParseConfig(*cfg)   #【参考相关详细分析】
    
        if g.Config().Debug {
            g.InitLog("debug")
        } else {
            g.InitLog("info")
        }
        
        # 全局初始化工作（工作目录/本地IP/RPC_Client）
        g.InitRootDir()             #【参考相关详细分析】
        g.InitLocalIp()             #【参考相关详细分析】
        g.InitRpcClients()          #【参考相关详细分析】
    
        # 采集功能与模块构造映射Map
        funcs.BuildMappers()        #【参考相关详细分析】
        
        # 
        go cron.InitDataHistory()     #【参考相关详细分析】
        
        # 上报Agent自身状态
        cron.ReportAgentStatus()      #【参考相关详细分析】
        # 同步模块
        cron.SyncMinePlugins()        #【参考相关详细分析】
        cron.SyncBuiltinMetrics()     #【参考相关详细分析】
        cron.SyncTrustableIps()       #【参考相关详细分析】
        # 采集数据
        cron.Collect()                #【参考相关详细分析】
    
        # API服务器运行
        go http.Start()               #【参考相关详细分析】
    
        # 阻塞主进程
        select {}

> funcs.CheckCollector() 各采集模块功能状态检测

func CheckCollector() {
    
        output := make(map[string]bool)
    
       #  引用外部"github.com/toolkits/nux"功能获取
       #  CurrentProcStat()
       #  ListDiskStats()
       #  ListeningPorts()
       #  AllProcs()
       
        _, procStatErr := nux.CurrentProcStat()
        _, listDiskErr := nux.ListDiskStats()
        ports, listeningPortsErr := nux.ListeningPorts()
        procs, psErr := nux.AllProcs()
    
        # 引用外部"github.com/toolkits/sys"功能执行模块Du系统命令    
        _, duErr := sys.CmdOut("du", "--help")
        
        # 各采集功能模块采集调用与结果判断（各模块采集代码实现可进一步分析）
        output["kernel  "] = len(KernelMetrics()) > 0
        output["df.bytes"] = DeviceMetricsCheck()
        output["net.if  "] = len(CoreNetMetrics([]string{})) > 0
        output["loadavg "] = len(LoadAvgMetrics()) > 0
        output["cpustat "] = procStatErr == nil
        output["disk.io "] = listDiskErr == nil
        output["memory  "] = len(MemMetrics()) > 0
        output["netstat "] = len(NetstatMetrics()) > 0
        output["ss -s   "] = len(SocketStatSummaryMetrics()) > 0
        output["ss -tln "] = listeningPortsErr == nil && len(ports) > 0
        output["ps aux  "] = psErr == nil && len(procs) > 0
        output["du -bs  "] = duErr == nil
       
        # 将MAP output的各值进行结果输出，True为OK,False为fail打印显示
        for k, v := range output {
            status := "fail"
            if v {
                status = "ok"
            }
            fmt.Println(k, "...", status)
        }
    }

> g.ParseConfig(*cfg) Agent配置文件解析

#通过命令指定配置文件，默认为同级目录下cfg.json
    cfg := flag.String("c", "cfg.json", "configuration file")
    
    # ParseConfig()解析cfg配置文件并保存在私有变量config内，可通过
    # Config()公开方法来获取GlobalConfig结构
    # Config().XXX 常用来获取单个配置项
    
    func ParseConfig(cfg string) {
        if cfg == "" {
            log.Fatalln("use -c to specify configuration file")
        }
    
        if !file.IsExist(cfg) {
            log.Fatalln("config file:", cfg, "is not existent. maybe you need `mv cfg.example.json cfg.json`")
        }
    
        ConfigFile = cfg
    
        configContent, err := file.ToTrimString(cfg)
        if err != nil {
            log.Fatalln("read config file:", cfg, "fail:", err)
        }
    
        var c GlobalConfig
        err = json.Unmarshal([]byte(configContent), &c)
        if err != nil {
            log.Fatalln("parse config file:", cfg, "fail:", err)
        }
    
        lock.Lock()
        defer lock.Unlock()
    
        config = &c
    
        log.Println("read config file:", cfg, "successfully")
    }
    
    #公开方法返回config
    func Config() *GlobalConfig {
        lock.RLock()
        defer lock.RUnlock()
        return config
    }
    
    #GlobalConfig结构化配置定义
    type GlobalConfig struct {
        Debug         bool              `json:"debug"`
        Hostname      string            `json:"hostname"`
        IP            string            `json:"ip"`
        Plugin        *PluginConfig     `json:"plugin"`
        Heartbeat     *HeartbeatConfig  `json:"heartbeat"`
        Transfer      *TransferConfig   `json:"transfer"`
        Http          *HttpConfig       `json:"http"`
        Collector     *CollectorConfig  `json:"collector"`
        DefaultTags   map[string]string `json:"default_tags"`
        IgnoreMetrics map[string]bool   `json:"ignore"`
    }

> g 全局化初始化

g.InitRootDir()  // 工作目录
        g.InitLocalIp()  // 本地IP
        g.InitRpcClients() // 实例化RPC_Client
    
    
    # 通过os.Getwd()方式获取当前工作目录
    var Root string
    func InitRootDir() {
        var err error
        Root, err = os.Getwd()
        if err != nil {
            log.Fatalln("getwd fail:", err)
        }
    }
    
    # 通过net.DialTimeout方式获取本地IP地址
    var LocalIp string
    func InitLocalIp() {
        if Config().Heartbeat.Enabled {
            conn, err := net.DialTimeout("tcp", Config().Heartbeat.Addr, time.Second*10)
            if err != nil {
                log.Println("get local addr failed !")
            } else {
                LocalIp = strings.Split(conn.LocalAddr().String(), ":")[0]
                conn.Close()
            }
        } else {
            log.Println("hearbeat is not enabled, can't get localip")
        }
    }
    
    # 构造&SingleConnRpcClient{}对象
    var HbsClient *SingleConnRpcClient
    func InitRpcClients() {
        if Config().Heartbeat.Enabled {
            HbsClient = &SingleConnRpcClient{
                RpcServer: Config().Heartbeat.Addr,
                Timeout:   time.Duration(Config().Heartbeat.Timeout) * time.Millisecond,
            }
        }
    }
    
    
    type SingleConnRpcClient struct {
        sync.Mutex
        rpcClient *rpc.Client  
        RpcServer string       #RPC服务器
        Timeout   time.Duration #超时间隔ms
    }

> funcs.BuildMappers() 构建采集功能映射表，表内每项代表一个采集项  
>  【`重要模块：采集映射表`】

func BuildMappers() {
        interval := g.Config().Transfer.Interval
        Mappers = []FuncsAndInterval{
            {   # 基础项采集 
                Fs: []func() []*model.MetricValue{
                    AgentMetrics,   // Agent状态
                    CpuMetrics,     // CPU
                    NetMetrics,     // NET
                    KernelMetrics,  // Kernel
                    LoadAvgMetrics, // CPU Load Avg
                    MemMetrics,     // Memory
                    DiskIOMetrics,  // Disk IO
                    IOStatsMetrics, // IO stats
                    NetstatMetrics, // NetStat
                    ProcMetrics,    // Proc
                    UdpMetrics,     // Udp
                },
                Interval: interval,
            },
            {  # 设备采集
                Fs: []func() []*model.MetricValue{
                    DeviceMetrics,
                },
                Interval: interval,
            },
            {  # Socket与port采集
                Fs: []func() []*model.MetricValue{
                    PortMetrics,
                    SocketStatSummaryMetrics,
                },
                Interval: interval,
            },
            {   # Du磁盘采集
                Fs: []func() []*model.MetricValue{
                    DuMetrics,
                },
                Interval: interval,
            },
            {    # Url采集
                Fs: []func() []*model.MetricValue{
                    UrlMetrics,
                },
                Interval: interval,
            },
            {   # Gpu采集
                Fs: []func() []*model.MetricValue{
                    GpuMetrics,
                },
                Interval: interval,
            },
        }
    }
    
    # 单个采集功能项
    type FuncsAndInterval struct {
        Fs       []func() []*model.MetricValue  //采集功能函数列表
        Interval int   //采集间隔 
    }
    
    # 完整映射列表
    var Mappers []FuncsAndInterval

> cron.InitDataHistory() 周期性更新CPU/Disk状态，构建历史数据

func InitDataHistory() {
        for {
            funcs.UpdateCpuStat()  
            funcs.UpdateDiskStats()
            time.Sleep(g.COLLECT_INTERVAL)
        }
    }
    
    # funcs->cpustat更新CPU状态，最新状态保存在procStatHistory[0]
    func UpdateCpuStat() error {
        ps, err := nux.CurrentProcStat()
        if err != nil {
            return err
        }
    
        psLock.Lock()
        defer psLock.Unlock()
        for i := historyCount - 1; i > 0; i-- {
            procStatHistory[i] = procStatHistory[i-1]
        }
    
        procStatHistory[0] = ps
        return nil
    }
    
    # funcs->diskstats更新Disk状态
    func UpdateDiskStats() error {
        dsList, err := nux.ListDiskStats()
        if err != nil {
            return err
        }
    
        dsLock.Lock()
        defer dsLock.Unlock()
        for i := 0; i < len(dsList); i++ {
            device := dsList[i].Device
            diskStatsMap[device] = [2]*nux.DiskStats{dsList[i], diskStatsMap[device][0]}   //?
        }
        return nil
    }

> cron.ReportAgentStatus() 周期性向HBS服务器上报状态

func ReportAgentStatus() {
        if g.Config().Heartbeat.Enabled && g.Config().Heartbeat.Addr != "" {
            go reportAgentStatus(time.Duration(g.Config().Heartbeat.Interval) * time.Second)
        }
    }
    
    # 通过RPC方式上报Agent信息
    func reportAgentStatus(interval time.Duration) {
        for {
            hostname, err := g.Hostname()
            if err != nil {
                hostname = fmt.Sprintf("error:%s", err.Error())
            }
    
            req := model.AgentReportRequest{
                Hostname:      hostname,    //主机名
                IP:            g.IP(),      //IP
                AgentVersion:  g.VERSION,   //版本
                PluginVersion: g.GetCurrPluginVersion(), //「插件版本信息」
            }
    
            var resp model.SimpleRpcResponse
    
            //RPC请求类型"Agent.ReportStatus"
            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)
        }
    }
    
    
    //g->tool模块获取当前插件Hash信息
    func GetCurrPluginVersion() string {
        if !Config().Plugin.Enabled {
            return "plugin not enabled"
        }
    
        pluginDir := Config().Plugin.Dir
        if !file.IsExist(pluginDir) {
            return "plugin dir not existent"
        }
      
        //获取当前活跃分支哈希值
        cmd := exec.Command("git", "rev-parse", "HEAD")
        cmd.Dir = pluginDir   //插件路径
    
        var out bytes.Buffer
        cmd.Stdout = &out
        err := cmd.Run()
        if err != nil {
            return fmt.Sprintf("Error:%s", err.Error())
        }
    
        return strings.TrimSpace(out.String())
    }

> cron.SyncMinePlugins() 周期性向HBS服务器同步插件，执行插件采集数据向Transfer上报数据。 【`重要功能模块：插件管理`】

# 公开同步插件方法 func SyncMinePlugins() {
        if !g.Config().Plugin.Enabled {
            return
        }
    
        if !g.Config().Heartbeat.Enabled {
            return
        }
    
        if g.Config().Heartbeat.Addr == "" {
            return
        }
    
        go syncMinePlugins()  //后台线程，同步函数
    }
    
    ## RPC向HBS同步插件
    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调用"Agent.MinePlugins"
            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)
            }
    
            if len(pluginDirs) == 0 {
                plugins.ClearAllPlugins()
            }
    
            desiredAll := make(map[string]*plugins.Plugin)
    
            for _, p := range pluginDirs {
                underOneDir := plugins.ListPlugins(strings.Trim(p, "/"))
                for k, v := range underOneDir {
                    desiredAll[k] = v   //同步到的插件格式化为MAP
                }
            }
    
            plugins.DelNoUsePlugins(desiredAll)  //依据同步信息，「清理插件」
            plugins.AddNewPlugins(desiredAll)    //依据同步信息，「更新插件」
    
        }
    }
    
    
    ### plugins->plugins清理Agent插件（HBS服务器已不存在的插件）
    func DelNoUsePlugins(newPlugins map[string]*Plugin) {
        for currKey, currPlugin := range Plugins {
            newPlugin, ok := newPlugins[currKey]
            if !ok || currPlugin.MTime != newPlugin.MTime {
                deletePlugin(currKey)
            }
        }
    }
    
    ### plugins->plugins更新和添加插件
    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
            sch.Schedule()  // 「 插件调度与执行（Ticker定时器）」
        }
    }
    
    
    
    ### 公有插件变量
    var (
        Plugins              = make(map[string]*Plugin)
        PluginsWithScheduler = make(map[string]*PluginScheduler)
    )
    
    
    #### plugins->scheduler插件调度
    func (this *PluginScheduler) Schedule() {
        go func() {
            for {
                select {
                case <-this.Ticker.C:    
                    PluginRun(this.Plugin) //
                case <-this.Quit:
                    this.Ticker.Stop()
                    return
                }
            }
        }()
    }
    
    ##### plugins->scheduler插件执行
    func PluginRun(plugin *Plugin) {
       
        timeout := plugin.Cycle*1000 - 500  //执行超时设置
        fpath := filepath.Join(g.Config().Plugin.Dir, plugin.FilePath)  //插件执行PATH
    
        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)   // EXEC执行插件命令创建
        var stdout bytes.Buffer
        cmd.Stdout = &stdout
        var stderr bytes.Buffer
        cmd.Stderr = &stderr
        cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
        err := cmd.Start()    // CMD start
        if err != nil {
            log.Printf("[ERROR] plugin start fail, error: %s\n", err)
            return
        }
        if debug {
            log.Println("plugin started:", fpath)
        }
    
        err, isTimeout := sys.CmdRunWithTimeout(cmd, time.Duration(timeout)*time.Millisecond)  // CMD Run
    
        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)    //「 上报数据 」
    }

> g.SendToTransfer() 向transfer上报数据

#g ->var向transfer服务器上报数据
    func SendToTransfer(metrics []*model.MetricValue) {
        if len(metrics) == 0 {
            return
        }
    
        dt := Config().DefaultTags
        if len(dt) > 0 {
            var buf bytes.Buffer
            default_tags_list := []string{}
            for k, v := range dt {
                buf.Reset()
                buf.WriteString(k)
                buf.WriteString("=")
                buf.WriteString(v)
                default_tags_list = append(default_tags_list, buf.String())
            }
            default_tags := strings.Join(default_tags_list, ",")
    
            for i, x := range metrics {
                buf.Reset()
                if x.Tags == "" {
                    metrics[i].Tags = default_tags
                } else {
                    buf.WriteString(metrics[i].Tags)
                    buf.WriteString(",")
                    buf.WriteString(default_tags)
                    metrics[i].Tags = buf.String()
                }
            }
        }
    
        debug := Config().Debug
    
        if debug {
            log.Printf("=> <Total=%d> %v\n", len(metrics), metrics[0])
        }
    
        var resp model.TransferResponse
        SendMetrics(metrics, &resp)  //「send」
        if debug {
            log.Println("<=", &resp)
        }
    }
    
    
    # g->transfer向transfer服务器发送采集数据
    func SendMetrics(metrics []*model.MetricValue, resp *model.TransferResponse) {
        rand.Seed(time.Now().UnixNano())
        for _, i := range rand.Perm(len(Config().Transfer.Addrs)) {
             // 随机获取一台Transfer地址
            addr := Config().Transfer.Addrs[i]
             
            //「获取Transfer RPC Client」，如果获取为失败，「初始化Transfer RPC Client」
            c := getTransferClient(addr) 
            if c == nil {
                c = initTransferClient(addr) 
            }
      
            //「上报数据」
            if updateMetrics(c, metrics, resp) {
                break
            }
        }
    }
    
    ## 获取可用的TransferClient对象
    func getTransferClient(addr string) *SingleConnRpcClient {
        TransferClientsLock.RLock()
        defer TransferClientsLock.RUnlock()
    
        if c, ok := TransferClients[addr]; ok {
            return c
        }
        return nil
    }
    
    ## 构建一个新的TransferClient对象
    func initTransferClient(addr string) *SingleConnRpcClient {
        var c *SingleConnRpcClient = &SingleConnRpcClient{
            RpcServer: addr,
            Timeout:   time.Duration(Config().Transfer.Timeout) * time.Millisecond,
        }
        TransferClientsLock.Lock()
        defer TransferClientsLock.Unlock()
        TransferClients[addr] = c
    
        return c
    }
    
    ## RPC向Transfer服务器上报metrics数据，RPC类型"Transfer.Update"
    func updateMetrics(c *SingleConnRpcClient, metrics []*model.MetricValue, resp *model.TransferResponse) bool {
        err := c.Call("Transfer.Update", metrics, resp)
        if err != nil {
            log.Println("call Transfer.Update fail:", c, err)
            return false
        }
        return true
    }

> cron.SyncBuiltinMetrics() 向HBS服务器同步监控端口、du路径、进程和URL

func SyncBuiltinMetrics() {
        if g.Config().Heartbeat.Enabled && g.Config().Heartbeat.Addr != "" {
            go syncBuiltinMetrics()  //后台线程同步
        }
    }
    
    # 向HBS同步
    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
            
            // RPC调用类型"Agent.BuiltinMetrics"
            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 {
                
                //健康检测的URL列表
                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)
                    }
                }
               
                //监控PORT端口列表
                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
                }
                 
                 // du路径列表
                if metric.Metric == g.DU_BS {
                    arr := strings.Split(metric.Tags, "=")
                    if len(arr) != 2 {
                        continue
                    }
    
                    paths = append(paths, strings.TrimSpace(arr[1]))
                    continue
                }
                
                // 进程列表
                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)     
    
        }
    }

> cron.SyncTrustableIps() 向HBS同步执行脚本的信任IP列表

func SyncTrustableIps() {
        if g.Config().Heartbeat.Enabled && g.Config().Heartbeat.Addr != "" {
            go syncTrustableIps() //后台线程同步信任IP列表
        }
    }
    
    func syncTrustableIps() {
        duration := time.Duration(g.Config().Heartbeat.Interval) * time.Second
    
        for {
            time.Sleep(duration)
            
            var ips string
            // RPC调用类型"Agent.TrustableIps"
            err := g.HbsClient.Call("Agent.TrustableIps", model.NullRpcRequest{}, &ips)
            if err != nil {
                log.Println("ERROR: call Agent.TrustableIps fail", err)
                continue
            }
            
            // 设置列表
            g.SetTrustableIps(ips)
        }
    }

> cron.Collect() 周期性采集数据

func Collect() {
    
        if !g.Config().Transfer.Enabled {
            return
        }
    
        if len(g.Config().Transfer.Addrs) == 0 {
            return
        }
    
        for _, v := range funcs.Mappers {
            go collect(int64(v.Interval), v.Fs) //后台线程采集与上报
        }
    }
    
    ## 采集与向transfer上报数据 【`关键执行收集与上报Codes`】
    func collect(sec int64, fns []func() []*model.MetricValue) {
        t := time.NewTicker(time.Second * time.Duration(sec))
        defer t.Stop()
        for {
            <-t.C
    
            hostname, err := g.Hostname()
            if err != nil {
                continue
            }
             
            mvs := []*model.MetricValue{}
            ignoreMetrics := g.Config().IgnoreMetrics
            
            // 迭代所有采集项，执行采集Funcation，汇总Metrics
            for _, fn := range fns {
                items := fn()   //调用采集模块funcation
                if items == nil {
                    continue
                }
    
                if len(items) == 0 {
                    continue
                }
    
                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
            }
    
            g.SendToTransfer(mvs)  //向transfer上报所收集数据
    
        }
    }

> http.Start() API与Dashboard服务运行

func init() {
        configAdminRoutes()   //管理接口路由
        configCpuRoutes()     //CPU信息路由
        configDfRoutes()      //DF挂载点信息路由
        configHealthRoutes()  //Health路由
        configIoStatRoutes()  //IoStat路由
        configKernelRoutes()  //Kernel路由
        configMemoryRoutes()  //Memory路由
        configPageRoutes()    //Dashboard路由
        configPluginRoutes()  //插件路由
        configPushRoutes()    //Push路由
        configRunRoutes()     //脚本执行路由
        configSystemRoutes()  //System路由
    }
    
    func Start() {
        if !g.Config().Http.Enabled {
            return
        }
    
        addr := g.Config().Http.Listen
        if addr == "" {
            return
        }
       
       //
        s := &http.Server{
            Addr:           addr,
            MaxHeaderBytes: 1 << 30,
        }
    
        log.Println("listening", addr)
        log.Fatalln(s.ListenAndServe())
    }
    
    # 配置http路由（Admin管理）
    func configAdminRoutes() {
        http.HandleFunc("/exit", func(w http.ResponseWriter, r *http.Request) {
            if g.IsTrustable(r.RemoteAddr) {
                w.Write([]byte("exiting..."))
                go func() {
                    time.Sleep(time.Second)
                    os.Exit(0)
                }()
            } else {
                w.Write([]byte("no privilege"))
            }
        })
    
        http.HandleFunc("/config/reload", func(w http.ResponseWriter, r *http.Request) {
            if g.IsTrustable(r.RemoteAddr) {
                g.ParseConfig(g.ConfigFile)
                RenderDataJson(w, g.Config())
            } else {
                w.Write([]byte("no privilege"))
            }
        })
    
        http.HandleFunc("/workdir", func(w http.ResponseWriter, r *http.Request) {
            RenderDataJson(w, file.SelfDir())
        })
    
        http.HandleFunc("/ips", func(w http.ResponseWriter, r *http.Request) {
            RenderDataJson(w, g.TrustableIps())
        })
    }
    
    # 配置http路由（Dashboard静态页）
    func configPageRoutes() {
    
        http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
            if strings.HasSuffix(r.URL.Path, "/") {
                if !file.IsExist(filepath.Join(g.Root, "/public", r.URL.Path, "index.html")) {
                    http.NotFound(w, r)
                    return
                }
            }
            http.FileServer(http.Dir(filepath.Join(g.Root, "/public"))).ServeHTTP(w, r)
        })
    
    }

### 三 默认采集模块分析

1. AgentMetrics -- agent alive状态采集

func AgentMetrics() []*model.MetricValue {
        return []*model.MetricValue{GaugeValue("agent.alive", 1)}
    }

2. CpuMetrics -- CPU相关信息采集

# 在main入口后台线程已周期性收集CPU状态，保存在私有procStatHistory内
    # go cron.InitDataHistory()-> UpdateCpuStat() 
    # 解析procStatHistory历史数据
    # 比如CpuIdle值获取
    func CpuIdle() float64 {
        psLock.RLock()
        defer psLock.RUnlock()
        dt := deltaTotal()
        if dt == 0 {
            return 0.0
        }
        invQuotient := 100.00 / float64(dt)
        return float64(procStatHistory[0].Cpu.Idle-procStatHistory[1].Cpu.Idle) * invQuotient  //解析cpuidle
    }
    
    
    func CpuMetrics() []*model.MetricValue {
        if !CpuPrepared() {
            return []*model.MetricValue{}
        }
    
        # CPU信息: idle, busy, user, nice, system, iowait, irq, softirq, steal, guest, switches
        
        cpuIdleVal := CpuIdle()
        idle := GaugeValue("cpu.idle", cpuIdleVal) //空闲值
        busy := GaugeValue("cpu.busy", 100.0-cpuIdleVal) //繁忙值
        user := GaugeValue("cpu.user", CpuUser()) // 用户使用CPU
        nice := GaugeValue("cpu.nice", CpuNice()) //CPU NICE值
        system := GaugeValue("cpu.system", CpuSystem()) //系统使用CPU
        iowait := GaugeValue("cpu.iowait", CpuIowait()) //IO等待
        irq := GaugeValue("cpu.irq", CpuIrq()) //IO中断请求
        softirq := GaugeValue("cpu.softirq", CpuSoftIrq()) //软中断请求
        steal := GaugeValue("cpu.steal", CpuSteal()) //Steal值
        guest := GaugeValue("cpu.guest", CpuGuest()) //Guest值
        switches := CounterValue("cpu.switches",   CurrentCpuSwitches())  //交换值
        return []*model.MetricValue{idle, busy, user, nice, system, iowait, irq, softirq, steal, guest, switches}
    }

3. NetMetrics 网卡相关信息采集

# 依据配置网卡接口信息采集数据 func NetMetrics() []*model.MetricValue {
        return CoreNetMetrics(g.Config().Collector.IfacePrefix)
    }
    
    #引用"github.com/toolkits/nux"采集网卡信息
    #实现读取与解析"/proc/net/dev"信息
    
    func CoreNetMetrics(ifacePrefix []string) []*model.MetricValue {
    
        netIfs, err := nux.NetIfs(ifacePrefix)
        if err != nil {
            log.Println(err)
            return []*model.MetricValue{}
        }
    
        cnt := len(netIfs)
        ret := make([]*model.MetricValue, cnt*26)
    
        for idx, netIf := range netIfs {
            iface := "iface=" + netIf.Iface
            ret[idx*26+0] = CounterValue("net.if.in.bytes", netIf.InBytes, iface) //入字节数
            ret[idx*26+1] = CounterValue("net.if.in.packets", netIf.InPackages, iface) //入包数
            ret[idx*26+2] = CounterValue("net.if.in.errors", netIf.InErrors, iface) //入错误包数
            ret[idx*26+3] = CounterValue("net.if.in.dropped", netIf.InDropped, iface) //入丢包数
            ret[idx*26+4] = CounterValue("net.if.in.fifo.errs", netIf.InFifoErrs, iface) //FIFO ERROR
            ret[idx*26+5] = CounterValue("net.if.in.frame.errs", netIf.InFrameErrs, iface) //帧ERROR
            ret[idx*26+6] = CounterValue("net.if.in.compressed", netIf.InCompressed, iface) //压缩
            ret[idx*26+7] = CounterValue("net.if.in.multicast", netIf.InMulticast, iface) //多播
            ret[idx*26+8] = CounterValue("net.if.out.bytes", netIf.OutBytes, iface) //出字节数
            ret[idx*26+9] = CounterValue("net.if.out.packets", netIf.OutPackages, iface) //出包数
            ret[idx*26+10] = CounterValue("net.if.out.errors", netIf.OutErrors, iface) //出错误数
            ret[idx*26+11] = CounterValue("net.if.out.dropped", netIf.OutDropped, iface) // 出丢包数
            ret[idx*26+12] = CounterValue("net.if.out.fifo.errs", netIf.OutFifoErrs, iface) //出FIFO ERROR
            ret[idx*26+13] = CounterValue("net.if.out.collisions", netIf.OutCollisions, iface) //出冲突
            ret[idx*26+14] = CounterValue("net.if.out.carrier.errs", netIf.OutCarrierErrs, iface) //载波
            ret[idx*26+15] = CounterValue("net.if.out.compressed", netIf.OutCompressed, iface) //压缩
            ret[idx*26+16] = CounterValue("net.if.total.bytes", netIf.TotalBytes, iface) //总字节
            ret[idx*26+17] = CounterValue("net.if.total.packets", netIf.TotalPackages, iface) //总包数
            ret[idx*26+18] = CounterValue("net.if.total.errors", netIf.TotalErrors, iface) //总错误
            ret[idx*26+19] = CounterValue("net.if.total.dropped", netIf.TotalDropped, iface) //总丢包数
            ret[idx*26+20] = GaugeValue("net.if.speed.bits", netIf.SpeedBits, iface) //速率bs
            ret[idx*26+21] = CounterValue("net.if.in.percent", netIf.InPercent, iface) //进百分比
            ret[idx*26+22] = CounterValue("net.if.out.percent", netIf.OutPercent, iface) //出百分比
            ret[idx*26+23] = CounterValue("net.if.in.bits", netIf.InBytes*8, iface) //进速率bs
            ret[idx*26+24] = CounterValue("net.if.out.bits", netIf.OutBytes*8, iface) //出速率bs
            ret[idx*26+25] = CounterValue("net.if.total.bits", netIf.TotalBytes*8, iface) //总速率bs
        }
        return ret
    }

4. KernelMetrics 系统内核相关信息采集

#引用 "github.com/toolkits/nux"采集内核信息
    #实现读取"/proc/sys/fs/file-max" 
    # "/proc/sys/fs/file-nr"  
    # "/proc/sys/kernel/pid_max" 
    
    func KernelMetrics() (L []*model.MetricValue) {
    
        maxFiles, err := nux.KernelMaxFiles()
        if err != nil {
            log.Println(err)
            return
        }
     
        //最大打开文件数
        L = append(L, GaugeValue("kernel.maxfiles", maxFiles))
    
        maxProc, err := nux.KernelMaxProc()
        if err != nil {
            log.Println(err)
            return
        }
       
        //最大进程数
        L = append(L, GaugeValue("kernel.maxproc", maxProc))
    
        allocateFiles, err := nux.KernelAllocateFiles()
        if err != nil {
            log.Println(err)
            return
        }
       
        //已分配文件数
        L = append(L, GaugeValue("kernel.files.allocated", allocateFiles))
        //已剩余文件数
        L = append(L, GaugeValue("kernel.files.left", maxFiles-allocateFiles))
        return
    }

5. LoadAvgMetrics -- CPU负载信息采集

#引用 "github.com/toolkits/nux"采集CPU负载信息
    #实现读取与解析 "/proc/loadavg"信息 
    #
    
    func LoadAvgMetrics() []*model.MetricValue {
        load, err := nux.LoadAvg()
        if err != nil {
            log.Println(err)
            return nil
        }
    
        return []*model.MetricValue{
            GaugeValue("load.1min", load.Avg1min), //1分钟平均
            GaugeValue("load.5min", load.Avg5min), //5分钟平均
            GaugeValue("load.15min", load.Avg15min), //15分钟平均
        }
    
    }

6. MemMetrics -- 内存信息采集

#引用 "github.com/toolkits/nux"采集内存信息
    #实现读取与解析"/proc/meminfo"信息
    
    func MemMetrics() []*model.MetricValue {
        m, err := nux.MemInfo()
        if err != nil {
            log.Println(err)
            return nil
        }
    
        memFree := m.MemFree + m.Buffers + m.Cached
        memUsed := m.MemTotal - memFree
    
        pmemFree := 0.0
        pmemUsed := 0.0
        if m.MemTotal != 0 {
            pmemFree = float64(memFree) * 100.0 / float64(m.MemTotal)
            pmemUsed = float64(memUsed) * 100.0 / float64(m.MemTotal)
        }
    
        pswapFree := 0.0
        pswapUsed := 0.0
        if m.SwapTotal != 0 {
            pswapFree = float64(m.SwapFree) * 100.0 / float64(m.SwapTotal)
            pswapUsed = float64(m.SwapUsed) * 100.0 / float64(m.SwapTotal)
        }
    
        return []*model.MetricValue{
            GaugeValue("mem.memtotal", m.MemTotal),  //总内存大小
            GaugeValue("mem.memused", memUsed), //已使用内存大小
            GaugeValue("mem.memfree", memFree), //可用内存大小
            GaugeValue("mem.swaptotal", m.SwapTotal), //交换内存总大小
            GaugeValue("mem.swapused", m.SwapUsed), //已使用交换内存大小
            GaugeValue("mem.swapfree", m.SwapFree), //可用交换内存大小
            GaugeValue("mem.memfree.percent", pmemFree), //可用内存占比
            GaugeValue("mem.memused.percent", pmemUsed), //已用内存占比
            GaugeValue("mem.swapfree.percent", pswapFree), //可用交换内存占比
            GaugeValue("mem.swapused.percent", pswapUsed), //已用交换同内存占比
        }
    
    }

7. DiskIOMetrics -- 磁盘IO信息采集

#引用 "github.com/toolkits/nux"采集DISK信息
    #实现读取与解析"/proc/diskstats"信息
    
    func DiskIOMetrics() (L []*model.MetricValue) {
    
        dsList, err := nux.ListDiskStats()
        if err != nil {
            log.Println(err)
            return
        }
       
        //迭代DISK列表
        for _, ds := range dsList {
            if !ShouldHandleDevice(ds.Device) {
                continue
            }
    
            device := "device=" + ds.Device
    
            L = append(L, CounterValue("disk.io.read_requests", ds.ReadRequests, device))//读请求
            L = append(L, CounterValue("disk.io.read_merged", ds.ReadMerged, device)) //相邻的读取请求合并
            L = append(L, CounterValue("disk.io.read_sectors", ds.ReadSectors, device)) //读扇区
            L = append(L, CounterValue("disk.io.msec_read", ds.MsecRead, device)) //disk读花费的时间
            L = append(L, CounterValue("disk.io.write_requests", ds.WriteRequests, device)) //写请求
            L = append(L, CounterValue("disk.io.write_merged", ds.WriteMerged, device)) //相邻的写请求合并
            L = append(L, CounterValue("disk.io.write_sectors", ds.WriteSectors, device)) //写扇区
            L = append(L, CounterValue("disk.io.msec_write", ds.MsecWrite, device)) //disk写花费的时间
            L = append(L, CounterValue("disk.io.ios_in_progress", ds.IosInProgress, device)) //正处理的IO数
            L = append(L, CounterValue("disk.io.msec_total", ds.MsecTotal, device)) //总花费的时间
            L = append(L, CounterValue("disk.io.msec_weighted_total", ds.MsecWeightedTotal, device)) //统计最近的IO完成时间和积压
        }
        return
    }

8. IOStatsMetrics -- 磁盘IO信息采集

# 在main入口后台线程已周期性收集CPU状态，保存在私有procStatHistory内
    # go cron.InitDataHistory()-> UpdateDiskStats()
    # 私有变量diskStatsMap保存周期性数据
    # 解析 diskStatsMap 获取IO状态数据
    
    
    func IOStatsMetrics() (L []*model.MetricValue) {
        dsLock.RLock()
        defer dsLock.RUnlock()
    
        for device := range diskStatsMap {
            if !ShouldHandleDevice(device) {
                continue
            }
    
            tags := "device=" + device
            rio := IODelta(device, IOReadRequests)
            wio := IODelta(device, IOWriteRequests)
            delta_rsec := IODelta(device, IOReadSectors)
            delta_wsec := IODelta(device, IOWriteSectors)
            ruse := IODelta(device, IOMsecRead)
            wuse := IODelta(device, IOMsecWrite)
            use := IODelta(device, IOMsecTotal)
            n_io := rio + wio
            avgrq_sz := 0.0
            await := 0.0
            svctm := 0.0
            if n_io != 0 {
                avgrq_sz = float64(delta_rsec+delta_wsec) / float64(n_io)
                await = float64(ruse+wuse) / float64(n_io)
                svctm = float64(use) / float64(n_io)
            }
    
            duration := IODelta(device, TS)
    
            L = append(L, GaugeValue("disk.io.read_bytes", float64(delta_rsec)*512.0, tags)) // 读字节数
            L = append(L, GaugeValue("disk.io.write_bytes", float64(delta_wsec)*512.0, tags)) // 写字节数
            L = append(L, GaugeValue("disk.io.avgrq_sz", avgrq_sz, tags)) //平均请求扇区的大小
            L = append(L, GaugeValue("disk.io.avgqu-sz",  float64(IODelta(device, IOMsecWeightedTotal))/1000.0, tags)) //是平均请求队列的长度
            L = append(L, GaugeValue("disk.io.await", await, tags)) //每一个IO请求的处理的平均时间
            L = append(L, GaugeValue("disk.io.svctm", svctm, tags))//表示平均每次设备I/O操作的服务时间
            tmp := float64(use) * 100.0 / float64(duration)
            if tmp > 100.0 {
                tmp = 100.0
            }
            L = append(L, GaugeValue("disk.io.util", tmp, tags)) //在统计时间内所有处理IO时间，除以总共统计时间
        }
    
        return
    }

9. NetstatMetrics -- TCP信息采集

#引用 "github.com/toolkits/nux"采集TCP信息
    #实现为读"/proc/net/netstat"信息并解析
    
    func NetstatMetrics() (L []*model.MetricValue) {
        tcpExts, err := nux.Netstat("TcpExt")
    
        if err != nil {
            log.Println(err)
            return
        }
    
        cnt := len(tcpExts)
        if cnt == 0 {
            return
        }
    
        for key, val := range tcpExts {
            if _, ok := USES[key]; !ok {
                continue
            }
            L = append(L, CounterValue("TcpExt."+key, val))
        }
    
        return
    }

10. ProcMetrics -- 采集Proc相关信息

# main入口 syncBuiltinMetrics()从HBS同步SetReportProcs()
    # g.ReportProcs()获取Procs列表map
    # 获取"/proc"目录下"/proc/%pid/cmdline"
    
    func ProcMetrics() (L []*model.MetricValue) {
    
        reportProcs := g.ReportProcs()
        sz := len(reportProcs)
        if sz == 0 {
            return
        }
    
        ps, err := nux.AllProcs()
        if err != nil {
            log.Println(err)
            return
        }
    
        pslen := len(ps)
    
        for tags, m := range reportProcs {
            cnt := 0
            for i := 0; i < pslen; i++ {
                if is_a(ps[i], m) {
                    cnt++
                }
            }
    
            L = append(L, GaugeValue(g.PROC_NUM, cnt, tags))
        }
    
        return
    }

11. UdpMetrics UDP接收和发送数据报采集

# 实现获取 "/proc/net/snmp"信息UDP接收和发送数据报
    
    func UdpMetrics() []*model.MetricValue {
        udp, err := nux.Snmp("Udp")
        if err != nil {
            log.Println("read snmp fail", err)
            return []*model.MetricValue{}
        }
    
        count := len(udp)
        ret := make([]*model.MetricValue, count)
        i := 0
        for key, val := range udp {
            ret[i] = CounterValue("snmp.Udp."+key, val)
            i++
        }
    
        return ret
    }

12. DeviceMetrics -- 挂载设备信息采集

# 实现获取与解析 "/proc/mounts"信息
    
    func DeviceMetrics() (L []*model.MetricValue) {
        mountPoints, err := nux.ListMountPoint()
    
        if err != nil {
            log.Error("collect device metrics fail:", err)
            return
        }
    
        var myMountPoints map[string]bool = make(map[string]bool)
    
        if len(g.Config().Collector.MountPoint) > 0 {
            for _, mp := range g.Config().Collector.MountPoint {
                myMountPoints[mp] = true
            }
        }
    
        var diskTotal uint64 = 0
        var diskUsed uint64 = 0
    
        for idx := range mountPoints {
            fsSpec, fsFile, fsVfstype := mountPoints[idx][0], mountPoints[idx][1], mountPoints[idx][2]
            if len(myMountPoints) > 0 {
                if _, ok := myMountPoints[fsFile]; !ok {
                    log.Debug("mount point not matched with config", fsFile, "ignored.")
                    continue
                }
            }
    
            var du *nux.DeviceUsage
            du, err = nux.BuildDeviceUsage(fsSpec, fsFile, fsVfstype)
            if err != nil {
                log.Error(err)
                continue
            }
    
            if du.BlocksAll == 0 {
                continue
            }
    
            diskTotal += du.BlocksAll
            diskUsed += du.BlocksUsed
    
            tags := fmt.Sprintf("mount=%s,fstype=%s", du.FsFile, du.FsVfstype) 
            L = append(L, GaugeValue("df.bytes.total", du.BlocksAll, tags))  //总大小字节
            L = append(L, GaugeValue("df.bytes.used", du.BlocksUsed, tags))  //已使用小大字节
            L = append(L, GaugeValue("df.bytes.free", du.BlocksFree, tags))  //剩余小大字节
            L = append(L, GaugeValue("df.bytes.used.percent", du.BlocksUsedPercent, tags)) //使用占比 
            L = append(L, GaugeValue("df.bytes.free.percent", du.BlocksFreePercent, tags)) //剩余块占比
    
            if du.InodesAll == 0 {
                continue
            }
    
            L = append(L, GaugeValue("df.inodes.total", du.InodesAll, tags))  //inode总数
            L = append(L, GaugeValue("df.inodes.used", du.InodesUsed, tags))  //已用inode数目
            L = append(L, GaugeValue("df.inodes.free", du.InodesFree, tags))  //可用inode数目
            L = append(L, GaugeValue("df.inodes.used.percent", du.InodesUsedPercent, tags)) // 已用inode占比
            L = append(L, GaugeValue("df.inodes.free.percent", du.InodesFreePercent, tags)) // 可用inode占比
    
        }
    
        if len(L) > 0 && diskTotal > 0 {
            L = append(L, GaugeValue("df.statistics.total", float64(diskTotal))) //统计总数 
            L = append(L, GaugeValue("df.statistics.used", float64(diskUsed))) // 统计使用
            L = append(L, GaugeValue("df.statistics.used.percent", float64(diskUsed)*100.0/float64(diskTotal)))  //统计使用占比
        }
    
        return
    }

13. PortMetrics -- TCP / UDP 端口信息采集

# main入口 syncBuiltinMetrics()-> g.SetReportPorts(ports)从HBS同步检测端口列表
    # g.ReportPorts() 获取列表
    # 实现获取TCP "ss -t -l -n" /获取UDP "ss -u -a -n"监听信息
    
    func PortMetrics() (L []*model.MetricValue) {
    
        reportPorts := g.ReportPorts()
        sz := len(reportPorts)
        if sz == 0 {
            return
        }
       
        allTcpPorts, err := nux.TcpPorts()
        if err != nil {
            log.Println(err)
            return
        }
    
        allUdpPorts, err := nux.UdpPorts()
        if err != nil {
            log.Println(err)
            return
        }
    
        for i := 0; i < sz; i++ {
            tags := fmt.Sprintf("port=%d", reportPorts[i])
            if slice.ContainsInt64(allTcpPorts, reportPorts[i]) || slice.ContainsInt64(allUdpPorts, reportPorts[i]) {
                L = append(L, GaugeValue(g.NET_PORT_LISTEN, 1, tags))
            } else {
                L = append(L, GaugeValue(g.NET_PORT_LISTEN, 0, tags))
            }
        }
    
        return
    }

14. SocketStatSummaryMetrics -- socket统计信息采集

# 实现获取socket统计信息 "ss -s" 
    func SocketStatSummaryMetrics() (L []*model.MetricValue) {
        ssMap, err := nux.SocketStatSummary()
        if err != nil {
            log.Println(err)
            return
        }
    
        for k, v := range ssMap {
            L = append(L, GaugeValue("ss."+k, v))
        }
    
        return
    }

15. DuMetrics

# main入口 syncBuiltinMetrics()从HBS同步g.SetDuPaths(paths)
    # g.DuPaths()获取磁盘分区列表
    
    func DuMetrics() (L []*model.MetricValue) {
        paths := g.DuPaths()
        result := make(chan *model.MetricValue, len(paths))
        var wg sync.WaitGroup
    
        //迭代所有PATH
        for _, path := range paths {
            wg.Add(1)
            go func(filepath string) {
                var err error
                defer func() {
                    if err != nil {
                        log.Println(err)
                        result <- GaugeValue(g.DU_BS, -1, "path="+path)
                    }
                    wg.Done()
                }()
                //tips:osx  does not support -b.
                cmd := exec.Command("du", "-bs", path) //exec执行du
                var stdout bytes.Buffer
                cmd.Stdout = &stdout
                var stderr bytes.Buffer
                cmd.Stderr = &stderr
                err = cmd.Start()
                if err != nil {
                    return
    
                }
                err, isTimeout := sys.CmdRunWithTimeout(cmd, time.Duration(timeout)*time.Second)
                if isTimeout {
                    err = errors.New(fmt.Sprintf("exec cmd : du -bs %s timeout", path))
                    return
                }
    
                errStr := stderr.String()
                if errStr != "" {
                    err = errors.New(errStr)
                    return
                }
    
                if err != nil {
                    err = errors.New(fmt.Sprintf("du -bs %s failed: %s", path, err.Error()))
                    return
                }
    
                arr := strings.Fields(stdout.String())
                if len(arr) < 2 {
                    errors.New(fmt.Sprintf("du -bs %s failed: %s", path, "return fields < 2"))
                    return
                }
    
                size, err := strconv.ParseUint(arr[0], 10, 64)
                if err != nil {
                    err = errors.New(fmt.Sprintf("cannot parse du -bs %s output", path))
                    return
                }
                result <- GaugeValue(g.DU_BS, size, "path="+path)
            }(path)
        }
        wg.Wait()
    
        resultLen := len(result)
        for i := 0; i < resultLen; i++ {
            L = append(L, <-result)
        }
        return
    }

16. UrlMetrics -- URL健康检测采集

func UrlMetrics() (L []*model.MetricValue) {
        //获取周期性向HBS所同步的URL列表
        reportUrls := g.ReportUrls() 
        sz := len(reportUrls)
        if sz == 0 {
            return
        }
        hostname, err := g.Hostname()
        if err != nil {
            hostname = "None"
        }
        // 迭代URL列表，执行URL状态检测
        for furl, timeout := range reportUrls {
            tags := fmt.Sprintf("url=%v,timeout=%v,src=%v", furl, timeout, hostname)
            //「probeUrl检测」
            if ok, _ := probeUrl(furl, timeout); !ok {
                L = append(L, GaugeValue(g.URL_CHECK_HEALTH, 0, tags))
                continue
            }
            L = append(L, GaugeValue(g.URL_CHECK_HEALTH, 1, tags))
        }
        return
    }
    
    //通过调用系统命令执行Curl命令进行Http_code的状态检测
    func probeUrl(furl string, timeout string) (bool, error) {
        bs, err := sys.CmdOutBytes("curl", "--max-filesize", "102400", "-I", "-m", timeout, "-o", "/dev/null", "-s", "-w", "%{http_code}", furl)
        if err != nil {
            log.Printf("probe url [%v] failed.the err is: [%v]\n", furl, err)
            return false, err
        }
        reader := bufio.NewReader(bytes.NewBuffer(bs))
        retcode, err := file.ReadLine(reader)
        if err != nil {
            log.Println("read retcode failed.err is:", err)
            return false, err
        }
        if strings.TrimSpace(string(retcode)) != "200" {
            log.Printf("return code [%v] is not 200.query url is [%v]", string(retcode), furl)
            return false, err
        }
        return true, err
    }

17. GpuMetrics -- GPU信息采集

// 引用外部库"github.com/mindprince/gonvml"
    // 需要load libnvidia-ml.so.1库
    func GpuMetrics() (L []*model.MetricValue) {
    
        if err := gonvml.Initialize(); err != nil {
            debug := g.Config().Debug
            if debug {
                log.Println("Initialize error: ", err)
            }
            return
        }
    
        defer gonvml.Shutdown()
    
        count, err := gonvml.DeviceCount()
        if err != nil {
            log.Println("DeviceCount error: ", err)
            return
        }
    
        if count == 0 {
            return
        }
    
        temperature := uint(0)
        totalMemory := uint64(0)
        usedMemory := uint64(0)
        gpuUtilization := uint(0)
        memoryUtilization := uint(0)
        powerUsage := uint(0)
        allUtilization := uint(0)
        allMemoryUtilization := uint(0)
    
        for i := 0; i < int(count); i++ {
            dev, err := gonvml.DeviceHandleByIndex(uint(i))
            if err != nil {
                log.Println("DeviceHandleByIndex error:", err)
                continue
            }
    
            uuid, err := dev.UUID()
            if err != nil {
                log.Println("dev.UUID error", err)
            }
    
            tag := "uuid=" + uuid
    
            // 不是所有gpu都有风扇
            fanSpeed, err := dev.FanSpeed()
            if err != nil {
                log.Println("dev.FanSpeed error: ", err)
            } else {
                L = append(L, GaugeValue("gpu.fan.speed", fanSpeed, tag))
            }
    
            temperature, err = dev.Temperature()
            if err != nil {
                log.Println("dev.Temperature error: ", err)
                continue
            }
    
            totalMemory, usedMemory, err = dev.MemoryInfo()
            if err != nil {
                log.Println("dev.MemoryInfo error: ", err)
                continue
            }
    
            // 单位换算为兆
            totalBillion := float64(totalMemory / 1024 / 1024)
            usedBillion := float64(usedMemory / 1024 / 1024)
    
            gpuUtilization, memoryUtilization, err = dev.UtilizationRates()
            if err != nil {
                log.Println("dev.UtilizationRates error: ", err)
                continue
            }
    
            allUtilization += gpuUtilization
            allMemoryUtilization += memoryUtilization
    
            powerUsage, err = dev.PowerUsage()
            if err != nil {
                log.Println("dev.PowerUsage error: ", err)
            }
    
            // 单位换算为瓦特
            powerWatt := float64(powerUsage / 1000)
    
            L = append(L, GaugeValue("gpu.temperature", temperature, tag))
            L = append(L, GaugeValue("gpu.memory.total", totalBillion, tag))
            L = append(L, GaugeValue("gpu.memory.used", usedBillion, tag))
            L = append(L, GaugeValue("gpu.memory.util", memoryUtilization, tag))
            L = append(L, GaugeValue("gpu.util", gpuUtilization, tag))
            L = append(L, GaugeValue("gpu.power.usage", powerWatt, tag))
        }
    
        L = append(L, GaugeValue("gpu.count", count))
        L = append(L, GaugeValue("gpu.util.avg", allUtilization/count))
        L = append(L, GaugeValue("gpu.memory.util.avg", allMemoryUtilization/count))
        return L
    }

### 四 思考与查证

1. 请查看源码，查证Agent所关联的组件HBS/Transfer哪个是集群模式？哪个是单点模式？为什么？
  2. 如果需要自定义Agent的采集数据有哪几种方式？从源码角度又应该如何添加相应的采集Code?
  3. 聊一聊从源码角度Agent组件哪些代码技术点或模式点值得你好好学习与借鉴的？

转载文章，原文链接： OpenFalcon源码分析(Agent组件)

关键字词源码 OpenFalcon

分享到：

上一篇：kubernetes 审计日志功能

下一篇：kubernetes 中定时任务的实现

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与

网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

OpenFalcon源码分析(Agent组件)

admin 9个月前

zhou 1年前

zhou 1年前

zhou 1年前

zhou 1年前

网站首页 返回列表 像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

OpenFalcon源码分析(Agent组件)

admin 官方 9个月前

zhou 1年前

zhou 1年前

zhou 1年前

zhou 1年前

网站首页返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

admin 9个月前