关于监控工具:Openfalcon-rpc长连接client

agent通过RCP向transfer发送metrics时,应用rpc长连贯,通过net/rpc保护一个tcp connection。

agent封装了SingleConnRpcClient构造,来实现单连贯rcp client。

1. SingleConnRpcClient如何应用的?

agent对应的后端transfer有多个,只有向1个transfer发送胜利就OK;
发送时,先依据addr找可用的rpcclient,找不到就init一个,而后调用update实现调用操作。

func SendMetrics(metrics []*model.MetricValue, resp *model.TransferResponse) {
    rand.Seed(time.Now().UnixNano())
    for _, i := range rand.Perm(len(Config().Transfer.Addrs)) {    //把数组的程序打乱
        addr := Config().Transfer.Addrs[i]

        c := getTransferClient(addr)        //从map中找
        if c == nil {    
            c = initTransferClient(addr)   //没找到,进行初始化
        }

        if updateMetrics(c, metrics, resp) {    //发送,若胜利则break
            break
        }
    }
}

初始化rpcclient:

func initTransferClient(addr string) *SingleConnRpcClient {
    var c *SingleConnRpcClient = &SingleConnRpcClient{
        RpcServer: addr,
        Timeout:   time.Duration(Config().Transfer.Timeout) * time.Millisecond,
    }
    ......
    TransferClients[addr] = c
    return c
}

发送的逻辑:调用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.Errorln("failed to call Transfer.Update:", c, err)
        return false
    }
    return true
}

2. SingleConnRpcClient如何实现的?

先看一下数据结构:

type SingleConnRpcClient struct {
   sync.Mutex
   rpcClient *rpc.Client
   RpcServer string
   Timeout   time.Duration
}

为防止并发调用,外部封装了sync.Mutext;

重点是Call()办法:

  • 先看一下连贯在不在,不在的话要init一个;
  • 启动一个goroutine来执行rpc办法;
  • 在主流程中用select进行超时管制,只有超时就close连贯;
func (this *SingleConnRpcClient) Call(method string, args interface{}, reply interface{}) error {
   this.Lock()
   defer this.Unlock()

   err := this.serverConn()
   if err != nil {
      return err
   }

   timeout := time.Duration(10 * time.Second)
   done := make(chan error, 1)

   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:        //失常返回err=nil
      if err != nil {
         log.Printf("[ERROR] rpc call with error => %v -- %v", err, this.rpcClient)
         this.close()
         return err
      }
   }
   return nil
}

初始化连贯很简略,就是创立了一个JsonRpCClient:

func (this *SingleConnRpcClient) serverConn() error {
   if this.rpcClient != nil {
      return nil
   }

   var err error
   var retry int = 1

   for {
      if this.rpcClient != nil {
         return nil
      }

      this.rpcClient, err = net.JsonRpcClient("tcp", this.RpcServer, this.Timeout)
      if err != nil {
         log.Printf("dial %s fail: %v", this.RpcServer, err)
         if retry > 3 {
            return err
         }
         time.Sleep(time.Duration(math.Pow(2.0, float64(retry))) * time.Second)
         retry++
         continue
      }
      return err
   }
}

连贯敞开:

func (this *SingleConnRpcClient) close() {
   if this.rpcClient != nil {
      this.rpcClient.Close()
      this.rpcClient = nil
   }
}

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

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

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

您可能还喜欢...

发表回复

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

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