32.1 proto通信协议定义

  1. syntax = "proto3";
  3. package lars;
  5. /* Lars系统的消息ID */
  6. enum MessageId {
  7.     ID_UNKNOW                = 0;  //proto3 enum第一个属性必须是0，用来占位
  8.     ID_GetRouteRequest       = 1;  //向DNS请求Route对应的关系的消息ID
  9.     ID_GetRouteResponse      = 2;  //DNS回复的Route信息的消息ID
  10.     ID_ReportStatusRequest   = 3;  //上报host调用状态信息请求消息ID
  11.     ID_GetHostRequest        = 4;  //API 发送请求host信息给 Lb Agent模块 消息ID
  12.     ID_GetHostResponse       = 5;  //agent 回执给 API host信息的 消息ID
  13.     ID_ReportRequest         = 6;  //API report get_host的调用结果给agent的 消息ID
  14. }
  16. //...
  17. //...
  19. // API 上报调用结果给 Agent(UDP)
  20. message ReportRequest {
  21.     int32 modid = 1;
  22.     int32 cmdid = 2;
  23.     HostInfo host = 3;
  24.     int32 retcode = 4;
  25. }

ID_ReportRequest和 message ReportRequest是针对API层与agent的请求互通协议。

32.2 Lars-API:Reporter()方法客户端实现

Lars/api/cpp/lars_api/lars_api.h

  1. #pragma once
  3. #include "lars_reactor.h"
  4. #include <string>
  6. class lars_client
  7. {
  8. public:
  9.     lars_client();
  10.     ~lars_client();
  12.     //lars 系统获取host信息 得到可用host的ip和port
  13.     int get_host(int modid, int cmdid, std::string& ip, int &port);
  15.     //lars 系统上报host调用信息
  16.     void report(int modid, int cmdid, const std::string &ip, int port, int retcode);
  18. private:
  19.     int _sockfd[3]; //3个udp socket fd 对应agent 3个udp server
  20.     uint32_t _seqid; //消息的序列号
  21. };
  1.     新增`report()`方法。

Lars/api/cpp/lars_api/lars_api.cpp

  1. //lars 系统上报host调用信息
  2. void lars_client::report(int modid, int cmdid, const std::string &ip, int port, int retcode)
  3. {
  4.     //1 封装上报消息
  5.     lars::ReportRequest req;    
  7.     req.set_modid(modid);
  8.     req.set_cmdid(cmdid);
  9.     req.set_retcode(retcode);
  11.     //1.1 host信息
  12.     lars::HostInfo *hp = req.mutable_host();
  14.     //ip
  15.     struct in_addr inaddr;
  16.     inet_aton(ip.c_str(), &inaddr);
  17.     int ip_num = inaddr.s_addr;
  18.     hp->set_ip(ip_num);
  20.     //port
  21.     hp->set_port(port);
  23.     //2. send
  24.     char write_buf[4096];
  25.     //消息头
  26.     msg_head head;
  27.     head.msglen = req.ByteSizeLong();
  28.     head.msgid = lars::ID_ReportRequest;
  29.     memcpy(write_buf, &head, MESSAGE_HEAD_LEN);
  30.     req.SerializeToArray(write_buf + MESSAGE_HEAD_LEN, head.msglen);
  32.     int index = (modid+cmdid)%3;
  33.     int ret = sendto(_sockfd[index], write_buf, head.msglen + MESSAGE_HEAD_LEN, 0, NULL, 0);
  34.     if (ret == -1) {
  35.         perror("sendto");
  36.     }
  37. }

Lars/api/cpp/example/example.cpp

  1.      int ret = api.get_host(modid, cmdid, ip, port);
  2.          if (ret == 0) {
  3.          std::cout << "host is " << ip << ":" << port << std::endl;
  5.         //上报调用结果
  6.         api.report(modid, cmdid, ip, port, 0);
  7.      }
  1.     在example的业务应用中，加上调用上报api。在每次调用完`get_host`。

32.3 report业务添加的配置参数信息

Lars/lars_loadbalance_agent/conf/lars_lb_agent.conf

  1. [reporter]
  2. ip = 127.0.0.1
  3. port = 7779
  5. [dnsserver]
  6. ip = 127.0.0.1
  7. port = 7778
  9. [loadbalance]
  11. ;经过若干次获取请求host节点后，试探选择一次overload过载节点
  12. probe_num=10
  14. ;初始化host_info主机信息访问成功的个数，防止刚启动时少量失败就认为过载
  15. init_succ_cnt=180
  17. ;当idle节点失败率高于此值，节点变overload状态
  18. err_rate=0.1
  20. ;当overload节点成功率高于此值，节点变成idle状态
  21. succ_rate=0.95
  23. ;当idle节点连续失败次数超过此值，节点变成overload状态
  24. contin_err_limit=15
  26. ;当overload节点连续成功次数超过此值, 节点变成idle状态
  27. contin_succ_limit=15

配置文件里在[loadbalance]中新增了一些字段。

那么我们需要在启动lb_agent的时候，加载这些配置文件参数.

lars_loadbalance_agent/include/main_server.h

  1. #pragma once
  2. #include "lars_reactor.h"
  3. #include "lars.pb.h"
  4. #include "route_lb.h"
  6. struct load_balance_config
  7. {
  8.     //经过若干次获取请求host节点后，试探选择一次overload过载节点
  9.     int probe_num; 
  11.     //初始化host_info主机信息访问成功的个数，防止刚启动时少量失败就认为过载
  12.     int init_succ_cnt;
  13. //**************************************************
  14.     //当idle节点失败率高于此值，节点变overload状态
  15.     float err_rate;
  17.     //当overload节点成功率高于此值，节点变成idle状态
  18.     float succ_rate;
  20.     //当idle节点连续失败次数超过此值，节点变成overload状态
  21.     int contin_err_limit;
  23.     //当overload节点连续成功次数超过此值, 节点变成idle状态
  24.     int contin_succ_limit;
  26.     //当前agent本地ip地址(用于上报 填充caller字段)
  27.     uint32_t local_ip;
  28. //**************************************************
  29. };

lars_loadbalance_agent/src/main_server.cpp

  1. #include "main_server.h"
  2. #include "lars.pb.h"
  3.  #include <netdb.h>
  5. // ... 
  7. //--------- 全局资源 ----------
  8. static void init_lb_agent()
  9. {
  10.     //1. 加载配置文件
  11.     config_file::setPath("./conf/lars_lb_agent.conf");
  12.     lb_config.probe_num = config_file::instance()->GetNumber("loadbalance", "probe_num", 10);
  13.     lb_config.init_succ_cnt = config_file::instance()->GetNumber("loadbalance", "init_succ_cnt", 180);
  14.     lb_config.err_rate = config_file::instance()->GetFloat("loadbalance", "err_rate", 0.1);
  15.     lb_config.succ_rate = config_file::instance()->GetFloat("loadbalance", "succ_rate", 0.92);
  16.     lb_config.contin_succ_limit = config_file::instance()->GetNumber("loadbalance", "contin_succ_limit", 10);
  17.     lb_config.contin_err_limit = config_file::instance()->GetNumber("loadbalance", "contin_err_limit", 10);
  19.     //2. 初始化3个route_lb模块
  20.     create_route_lb();
  22.     //3. 加载本地ip
  23.     char my_host_name[1024];
  24.     if (gethostname(my_host_name, 1024) == 0) {
  25.         struct hostent *hd = gethostbyname(my_host_name);
  27.         if (hd)
  28.         {
  29.             struct sockaddr_in myaddr;
  30.             myaddr.sin_addr = *(struct in_addr*)hd->h_addr;
  31.             lb_config.local_ip = ntohl(myaddr.sin_addr.s_addr);
  32.         }
  33.     }
  35.     if (!lb_config.local_ip)  {
  36.         struct in_addr inaddr;
  37.         inet_aton("127.0.0.1", &inaddr);
  38.         lb_config.local_ip = ntohl(inaddr.s_addr);
  39.     }
  40. }
  43. // ...

这里的本地ip，是之后在上报的时候，发送消息需要一个caller参数，这个caller参数我们就暂时默认是当前agent的ip为caller。

32.4 Agent UDP Server处理API-Report请求

  1. 接下来我们针对API发送的report的`ID_ReportRequest`进行处理.

lars_loadbalance_agent/src/agent_udp_server.cpp

  1. #include "lars_reactor.h"
  2. #include "main_server.h"
  4. // ...
  6. static void report_cb(const char *data, uint32_t len, int msgid, net_connection *net_conn, void *user_data)
  7. {
  8.     lars::ReportRequest req;
  10.     req.ParseFromArray(data, len);
  12.     route_lb *ptr_route_lb = (route_lb*)user_data;
  13.     ptr_route_lb->report_host(req);
  14. }
  17. void * agent_server_main(void * args)
  18. {
  19.     long index = (long)args;
  21.     short port = index + 8888;
  22.     event_loop loop;
  24.     udp_server server(&loop, "0.0.0.0", port);
  26.     //给server注册消息分发路由业务, 针对ID_GetHostRequest处理 每个udp拥有一个对应的route_lb
  27.     server.add_msg_router(lars::ID_GetHostRequest, get_host_cb,  r_lb[port-8888]);
  29.     //======================================================
  30.     //给server注册消息分发路由业务，针对ID_ReportRequest处理
  31.     server.add_msg_router(lars::ID_ReportRequest, report_cb, r_lb[port-8888]);
  32.     //======================================================
  34.     printf("agent UDP server :port %d is started...\n", port);
  35.     loop.event_process();
  37.     return NULL;
  38. }
  41. void start_UDP_servers(void)
  42. {
  43.     for (long i = 0; i < 3; i ++) {
  44.         pthread_t tid;
  46.         int ret = pthread_create(&tid, NULL, agent_server_main, (void*)i);
  47.         if (ret == -1) {
  48.             perror("pthread_create");
  49.             exit(1);
  50.         }
  52.         pthread_detach(tid);
  53.     }
  54. }

这里主要是通过一个udp server中的route_lb对象来调用的report_host(req)方法。我们来实现这个方法。

lars_loadbalance_agent/src/route_lb.cpp

  1. //agent 上报某主机的获取结果
  2. void route_lb::report_host(lars::ReportRequest req)
  3. {
  4.     int modid = req.modid();
  5.     int cmdid = req.cmdid();
  6.     int retcode = req.retcode();
  7.     int ip = req.host().ip();
  8.     int port = req.host().port();
  10.     uint64_t key = ((uint64_t)modid << 32) + cmdid;
  12.     pthread_mutex_lock(&_mutex);
  13.     if (_route_lb_map.find(key) != _route_lb_map.end()) {
  14.         load_balance *lb = _route_lb_map[key];
  16.         lb->report(ip, port, retcode);
  18.         //上报信息给远程reporter服务器
  19.         lb->commit();
  20.     }
  22.     pthread_mutex_unlock(&_mutex);
  23. }

当然，route_lb最终还是分管每个modid/cmdid对应的load_balance模块,那么选择一个可用的load_balance对象，调用load_balance的report()方法.而通过commit()方法，将report的上报结果提交到远程的report service中去。

接下来我们看一下load_balance的report方法实现.

lars_loadbalance_agent/src/load_balance.cpp

  1. //上报当前host主机调用情况给远端repoter service
  2. void load_balance::report(int ip, int port, int retcode)
  3. {
  4.     uint64_t key = ((uint64_t)ip << 32)  + port;
  6.     if (_host_map.find(key) == _host_map.end()) {
  7.         return;
  8.     }
  10.     //1 计数统计
  12.     host_info *hi = _host_map[key];
  13.     if (retcode == lars::RET_SUCC) { // retcode == 0
  14.         //更新虚拟成功、真实成功次数
  15.         hi->vsucc ++;
  16.         hi->rsucc ++;
  18.         //连续成功增加
  19.         hi->contin_succ ++; 
  20.         //连续失败次数归零
  21.         hi->contin_err = 0;
  22.     }
  23.     else {
  24.         //更新虚拟失败、真实失败次数 
  25.         hi->verr ++;
  26.         hi->rerr ++;
  28.         //连续失败个数增加
  29.         hi->contin_err++;
  30.         //连续成功次数归零
  31.         hi->contin_succ = 0;
  32.     }
  34.     //2.检查节点状态
  36.     //检查idle节点是否满足overload条件
  37.     //或者overload节点是否满足idle条件
  39.     //--> 如果是dile节点,则只有调用失败才有必要判断是否达到overload条件
  40.     if (hi->overload == false && retcode != lars::RET_SUCC) {
  42.         bool overload = false;
  44.         //idle节点，检查是否达到判定为overload的状态条件 
  45.         //(1).计算失败率,如果大于预设值失败率，则为overload
  46.         double err_rate = hi->verr * 1.0 / (hi->vsucc + hi->verr);
  48.         if (err_rate > lb_config.err_rate) {
  49.             overload = true;            
  50.         }
  52.         //(2).连续失败次数达到阈值，判定为overload
  53.         if( overload == false && hi->contin_err >= (uint32_t)lb_config.contin_err_limit) {
  54.             overload = true;
  55.         }
  57.         //判定overload需要做的更改流程
  58.         if (overload) {
  59.             struct in_addr saddr;
  61.             saddr.s_addr = htonl(hi->ip);
  62.             printf("[%d, %d] host %s:%d change overload, succ %u err %u\n", 
  63.                     _modid, _cmdid, inet_ntoa(saddr), hi->port, hi->vsucc, hi->verr);
  65.             //设置hi为overload状态 
  66.             hi->set_overload();
  67.             //移出_idle_list,放入_overload_list
  68.             _idle_list.remove(hi);
  69.             _overload_list.push_back(hi);
  70.             return;
  71.         }
  73.     }
  74.     //--> 如果是overload节点，则只有调用成功才有必要判断是否达到idle条件
  75.     else if (hi->overload == true && retcode == lars::RET_SUCC) {
  76.         bool idle = false;
  78.         //overload节点，检查是否达到回到idle状态的条件
  79.         //(1).计算成功率，如果大于预设值的成功率，则为idle
  80.         double succ_rate = hi->vsucc * 1.0 / (hi->vsucc + hi->verr);
  82.         if (succ_rate > lb_config.succ_rate) {
  83.             idle = true;
  84.         }
  86.         //(2).连续成功次数达到阈值，判定为idle
  87.         if (idle == false && hi->contin_succ >= (uint32_t)lb_config.contin_succ_limit) {
  88.             idle = true;
  89.         }
  91.         //判定为idle需要做的更改流程
  92.         if (idle) {
  93.             struct in_addr saddr;
  94.             saddr.s_addr = htonl(hi->ip);
  95.             printf("[%d, %d] host %s:%d change idle, succ %u err %u\n", 
  96.                     _modid, _cmdid, inet_ntoa(saddr), hi->port, hi->vsucc, hi->verr);
  98.             //设置为idle状态
  99.             hi->set_idle();
  100.             //移出overload_list, 放入_idle_list
  101.             _overload_list.remove(hi);
  102.             _idle_list.push_back(hi);
  103.             return;
  104.         }
  105.     }
  107.     //TODO 窗口检查和超时机制 
  108. }

其中set_idle()与set_overload()方法实现如下：

lars_loadbalance_agent/src/host_info.cpp

  1. #include "host_info.h"
  2. #include "main_server.h"
  4. void host_info::set_idle()
  5. {
  6.     vsucc = lb_config.init_succ_cnt;
  7.     verr = 0;
  8.     rsucc = 0;
  9.     rerr = 0;
  10.     contin_succ = 0;
  11.     contin_err = 0;
  12.     overload = false;
  13. }
  15. void host_info::set_overload()
  16. {
  17.     vsucc = 0;
  18.     verr = lb_config.init_err_cnt;//overload的初试虚拟err错误次数
  19.     rsucc = 0;
  20.     rerr = 0;
  21.     contin_err = 0;
  22.     contin_succ = 0;
  23.     overload = true;
  24. }

load_balance的report()方法实现主要是针对两个链表做节点的交替处理。和成功率失败率的判断。

节点失败率 = 节点verr / (vsucc + verr)
节点成功率 = 节点vsucc / (vsucc + verr)

当idle节点的失败率>预设值（默认10%），将节点判定为overload；
当overload节点的成功率>预设值（默认95%），将节点判定为idle；

而不可以idle/overload节点都只关注成功率or都只关注失败率，这样可能造成节点在idle/overload状态间频繁切换
为idle节点、overload节点设置不同的阈值可以区别对待。

  1. 接下来我们来实现`load_balance`的`commit()`方法。

lars_loadbalance_agent/src/load_balance.cpp

  1. //提交host的调用结果给远程reporter service上报结果
  2. void load_balance::commit()
  3. {
  4.     if (this->empty() == true) {
  5.         return;
  6.     }
  8.     //1. 封装请求消息
  9.     lars::ReportStatusRequest req;
  10.     req.set_modid(_modid);
  11.     req.set_cmdid(_cmdid);
  12.     req.set_ts(time(NULL));
  13.     req.set_caller(lb_config.local_ip);
  15.     //2. 从idle_list取值
  16.     for (host_list_it it = _idle_list.begin(); it != _idle_list.end(); it++) {
  17.         host_info *hi = *it;    
  18.         lars::HostCallResult call_res;
  19.         call_res.set_ip(hi->ip);
  20.         call_res.set_port(hi->port);
  21.         call_res.set_succ(hi->rsucc);
  22.         call_res.set_err(hi->rerr);
  23.         call_res.set_overload(false);
  25.         req.add_results()->CopyFrom(call_res);
  26.     }
  28.     //3. 从over_list取值
  29.     for (host_list_it it = _overload_list.begin(); it != _overload_list.end(); it++) {
  30.         host_info *hi = *it;
  31.         lars::HostCallResult call_res;
  32.         call_res.set_ip(hi->ip);
  33.         call_res.set_port(hi->port);
  34.         call_res.set_succ(hi->rsucc);
  35.         call_res.set_err(hi->rerr);
  36.         call_res.set_overload(true);
  38.         req.add_results()->CopyFrom(call_res);
  39.     }
  41.     //4 发送给report_client 的消息队列
  42.     report_queue->send(req);
  43. }