关于监控工具:Openfalcon-hbs源码解读

hbs负责周期性的读取db中内容,缓存到本地cache,而后提供RPC接口以供agent和judge两个组件查问调用。

// modules/hbs/rpc/rpc.go
func Start() {
    server := rpc.NewServer()
    server.Register(new(Agent))
    server.Register(new(Hbs))
    l, e := net.Listen("tcp", addr)
    for {
        conn, err := l.Accept()
        ......
        go server.ServeCodec(jsonrpc.NewServerCodec(conn))
    }
}

hbs对接agent:

  • 解决agent heartbeat申请;
  • 解决agent plugin的查问申请;
  • 解决agent 监控哪些过程、哪些端口的查问申请;

hbs对接judge:

  • 查问以后配置的所有告警策略;
  • 查问以后配置的所有告警表达式;

整体流程图:

1. hbs对接agent

agent查问执行哪些plugin:

// modules/hbs/rpc/agent.go
func (t *Agent) MinePlugins(args model.AgentHeartbeatRequest, reply *model.AgentPluginsResponse) error {
    if args.Hostname == "" {
        return nil
    }

    reply.Plugins = cache.GetPlugins(args.Hostname)
    reply.Timestamp = time.Now().Unix()

    return nil
}

agent查问监控哪些过程、哪些端口:

// modules/hbs/rpc/agent.go
func (t *Agent) BuiltinMetrics(args *model.AgentHeartbeatRequest, reply *model.BuiltinMetricResponse) error {
    if args.Hostname == "" {
        return nil
    }
    metrics, err := cache.GetBuiltinMetrics(args.Hostname)
    if err != nil {
        return nil
    }
    checksum := ""
    if len(metrics) > 0 {
        checksum = DigestBuiltinMetrics(metrics)
    }
    if args.Checksum == checksum {
        reply.Metrics = []*model.BuiltinMetric{}
    } else {
        reply.Metrics = metrics
    }
    reply.Checksum = checksum
    reply.Timestamp = time.Now().Unix()
    return nil
}

能够看到,下面的rpc接口操作的都是cache,hbs会定期的从db中查问数据,而后缓存在本地cache,以供agent查问:

// modules/hbs/cache/cache.go
func LoopInit() {
    for {
        //1min周期
        time.Sleep(time.Minute)
        GroupPlugins.Init()
        ....
    }
}

从db读取数据,而后保留在本地:为了避免并发写data,这里加了锁

// moduels/hbs/cache/plugins.go
var GroupPlugins = &SafeGroupPlugins{M: make(map[int][]string)}

func (this *SafeGroupPlugins) Init() {
    m, err := db.QueryPlugins()
    if err != nil {
        return
    }

    this.Lock()
    defer this.Unlock()
    this.M = m
}

2. hbs对接judge

judge查问告警表达式:

// modules/hbs/rpc/hbs.go
func (t *Hbs) GetExpressions(req model.NullRpcRequest, reply *model.ExpressionResponse) error {
    reply.Expressions = cache.ExpressionCache.Get()
    return nil
}

judge查问告警策略:因为关联多个db table,这里进行了比较复杂的拼装

// modules/hbs/rpc/hbs.go
func (t *Hbs) GetStrategies(req model.NullRpcRequest, reply *model.StrategiesResponse) error {
    reply.HostStrategies = []*model.HostStrategy{}
    hidTids := cache.HostTemplateIds.GetMap()
    hosts := cache.MonitoredHosts.Get()
    tpls := cache.TemplateCache.GetMap()
    strategies := cache.Strategies.GetMap()
    tpl2Strategies := Tpl2Strategies(strategies)
    hostStrategies := make([]*model.HostStrategy, 0, sz)
    for hostId, tplIds := range hidTids {
        ss := CalcInheritStrategies(tpls, tplIds, tpl2Strategies)
        hs := model.HostStrategy{
            Hostname:   h.Name,
            Strategies: ss,
        }
        hostStrategies = append(hostStrategies, &hs)
    }
    reply.HostStrategies = hostStrategies
    return nil
}

同样的,下面的rpc接口操作的都是cache,hbs会定期的从db中查问数据,而后缓存在本地cache,以供judge查问:

// modules/hbs/cache/cache.go
func LoopInit() {
    for {
        time.Sleep(time.Minute)
        ......
        GroupTemplates.Init()
        HostGroupsMap.Init()
        HostMap.Init()
        TemplateCache.Init()
        Strategies.Init(TemplateCache.GetMap())
        HostTemplateIds.Init()
        ExpressionCache.Init()
    }
}

cache的数据保留在map中,而后读取db中的数据(定期),笼罩掉map

// modules/hbs/cache/templates.go
type SafeTemplateCache struct {
    sync.RWMutex
    M map[int]*model.Template
}

var TemplateCache = &SafeTemplateCache{M: make(map[int]*model.Template)}

func (this *SafeTemplateCache) Init() {
    ts, err := db.QueryTemplates()
    if err != nil {
        return
    }
    this.Lock()
    defer this.Unlock()
    this.M = ts
}

【腾讯云】轻量 2核2G4M,首年65元

阿里云限时活动-云数据库 RDS MySQL  1核2G配置 1.88/月 速抢

本文由乐趣区整理发布,转载请注明出处,谢谢。

您可能还喜欢...

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据