码迷，mamicode.com
首页 > 其他好文 > 详细

Ambari深入学习（II）－实现细节

时间：2019-05-21 14:42:03      阅读：120      评论：0      收藏：0      [点我收藏+]

标签：告诉   spring框架   加载   功能   书签   产生   manifest   stun   ogg   

在第一节中，我们简单讲了一下Ambari的系统架构。我们这一节主要分析Ambari的源代码，总览Ambari的具体实现方式及其工作细节。

 一、Ambari-Server启动

Ambari-Server是一个WEB Server，提供统一的REST API接口，同时向web和agent开放了两个不同的端口（默认前者是8080, 后者是8440或者8441）。它是由Jetty Server容器构建起来的，通过Spring Framework构建出来的WEB服务器，其中大量采用了google提供的Guice注解完成spring框架所需要的注入功能（想一想，之前spring框架需要加载一个applicationcontext.xml文件来把bean注入进来，现在可以用Guice注解的方式就可以轻松完成）。 REST框架由JAX-RS标准来构建。
 
 Ambari-Server接受来自两处的REST请求，Agent过来的请求处理逻辑由包org.apache.ambari.server.agent处理， 而API所的处理逻辑来自org.apache.ambari.server.api。详见如下代码：
Java代码  技术图片
  1. // API handler加载的REST处理包  
  2.  ServletHolder sh = new ServletHolder(ServletContainer.class);  
  3.     //采用包结构的形式进行加载  
  4.       sh.setInitParameter("com.sun.jersey.config.property.resourceConfigClass",  
  5.           "com.sun.jersey.api.core.PackagesResourceConfig");  
  6.     // 下面是/api/v1/接受的请求由哪些包来处理  
  7.       sh.setInitParameter("com.sun.jersey.config.property.packages",  
  8.           "org.apache.ambari.server.api.rest;" +  
  9.               "org.apache.ambari.server.api.services;" +  
  10.               "org.apache.ambari.eventdb.webservice;" +  
  11.               "org.apache.ambari.server.api");  
  12.       sh.setInitParameter("com.sun.jersey.api.json.POJOMappingFeature",  
  13.           "true");  
  14.       root.addServlet(sh, "/api/v1/*");  
  15.       sh.setInitOrder(2);  
  16.   
  17.       // Agent Handler加载的REST包，主要是org.apache.ambari.server.agent.rest.HeartBeatHandler来接受心跳请求  
  18.       ServletHolder agent = new ServletHolder(ServletContainer.class);  
  19.       agent.setInitParameter("com.sun.jersey.config.property.resourceConfigClass",  
  20.           "com.sun.jersey.api.core.PackagesResourceConfig");  
  21.       agent.setInitParameter("com.sun.jersey.config.property.packages",  
  22.           "org.apache.ambari.server.agent.rest;" + "org.apache.ambari.server.api");  
  23.       agent.setInitParameter("com.sun.jersey.api.json.POJOMappingFeature",  
  24.           "true");  
  25.       agentroot.addServlet(agent, "/agent/v1/*");  
  26.       agent.setInitOrder(3);  
  27.   
  28.       // 对agent发过来的数据包进行证书签名所需要加载的包  
  29.       ServletHolder cert = new ServletHolder(ServletContainer.class);  
  30.       cert.setInitParameter("com.sun.jersey.config.property.resourceConfigClass",  
  31.           "com.sun.jersey.api.core.PackagesResourceConfig");  
  32.       cert.setInitParameter("com.sun.jersey.config.property.packages",  
  33.           "org.apache.ambari.server.security.unsecured.rest;" + "org.apache.ambari.server.api");  
  34.       cert.setInitParameter("com.sun.jersey.api.json.POJOMappingFeature",  
  35.           "true");  
  36.       agentroot.addServlet(cert, "/*");  
  37.       cert.setInitOrder(4);  
  38.      
  39.       // WEB客户端提供的数据包  
  40.       ServletHolder resources = new ServletHolder(ServletContainer.class);  
  41.       resources.setInitParameter("com.sun.jersey.config.property.resourceConfigClass",  
  42.           "com.sun.jersey.api.core.PackagesResourceConfig");  
  43.       resources.setInitParameter("com.sun.jersey.config.property.packages",  
  44.           "org.apache.ambari.server.resources.api.rest;" + "org.apache.ambari.server.api");  
  45.       resources.setInitParameter("com.sun.jersey.api.json.POJOMappingFeature",  
  46.           "true");  
  47.       root.addServlet(resources, "/resources/*");  
  48.       resources.setInitOrder(6)  
 
 正如上一节所述，Ambari-Server有一个状态机管理模块，所有节点的状态信息更改都最终提供给状态机进行更改操作，因此状态机是一个很忙的组件。在Ambari-Server里面，把每一次更改操作都把它当作是一类事件，采用事件驱动机制完成对应的任务。这种思想有点借鉴已经运用在hadoop 2.x YARN里面的事件驱动机制。事件驱动机制能够一种高效的异步RPC请求方式，直接调用需要执行相应的代码逻辑，而事件驱动只需要产生事件统一提交给事件处理器，因此事件驱动需要一个更复杂的有限状态机结合起来一同使用。

二、Ambari-Server处理Ambari-Agent请求

Agent发送过来的心跳请求由org.apache.ambari.server.agent.HeartBeatHandler.handleHeartBeat(HeartBeat)来处理，执行完后，同时会返回org.apache.ambari.server.agent.HeartBeatResponse给agent。 org.apache.ambari.server.agent.HeartBeat里面主要含了两类信息：节点的状态信息nodeStatus和服务状态信息componentStatus。
Java代码  技术图片
  1. public HeartBeatResponse handleHeartBeat(HeartBeat heartbeat)  
  2.      throws AmbariException {  
  3.    String hostname = heartbeat.getHostname();  
  4.    Long currentResponseId = hostResponseIds.get(hostname);  
  5.    HeartBeatResponse response;  
  6.    if (currentResponseId == null) {  
  7.      //Server restarted, or unknown host.  
  8.      LOG.error("CurrentResponseId unknown - send register command");  
  9.      return createRegisterCommand();  // 无responseId, 新请求，就进行注册, responseId ＝0   
  10.    }  
  11.   
  12.    LOG.info("Received heartbeat from host"  
  13.        + ", hostname=" + hostname  
  14.        + ", currentResponseId=" + currentResponseId  
  15.        + ", receivedResponseId=" + heartbeat.getResponseId());  
  16.   
  17.    if (heartbeat.getResponseId() == currentResponseId - 1) {  
  18.      LOG.warn("Old responseId received - response was lost - returning cached response");  
  19.      return hostResponses.get(hostname);  
  20.    } else if (heartbeat.getResponseId() != currentResponseId) {  
  21.      LOG.error("Error in responseId sequence - sending agent restart command");  
  22.      return createRestartCommand(currentResponseId);   //  心跳是历史记录，那么就要求其重启，重新注册，  
  23.                                                        //  responseId 不变  
  24.    }  
  25.   
  26.    response = new HeartBeatResponse();  
  27.    response.setResponseId(++currentResponseId);  // responseId 加 1 , 返回一个新的responseId，下次心跳又要把这个responseId带回来。  
  28.    Host hostObject = clusterFsm.getHost(hostname);  
  29.   
  30.    if (hostObject.getState().equals(HostState.HEARTBEAT_LOST)) {     // 失去心跳  
  31.      // After loosing heartbeat agent should reregister  
  32.      LOG.warn("Host is in HEARTBEAT_LOST state - sending register command");  
  33.      return createRegisterCommand();   //失去连接，要求重新注册, responseId=0  
  34.    }  
  35.   
  36.    hostResponseIds.put(hostname, currentResponseId);  
  37.    hostResponses.put(hostname, response);  
  38.   
  39.    long now = System.currentTimeMillis();  
  40.    HostState hostState = hostObject.getState();  
  41.    // If the host is waiting for component status updates, notify it  
  42.    if (heartbeat.componentStatus.size() > 0  
  43.        && hostObject.getState().equals(HostState.WAITING_FOR_HOST_STATUS_UPDATES)) {  // 节点已经进行了注册，但是该节点还没有汇报相关状态信息,等待服务状态更新  
  44.      try {  
  45.        LOG.debug("Got component status updates");      
  46.        hostObject.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now));  // 更新服务状态机  
  47.      } catch (InvalidStateTransitionException e) {  
  48.        LOG.warn("Failed to notify the host about component status updates", e);  
  49.      }  
  50.    }  
  51.   
  52.    try {  
  53.      if (heartbeat.getNodeStatus().getStatus().equals(HostStatus.Status.HEALTHY)) {  
  54.        hostObject.handleEvent(new HostHealthyHeartbeatEvent(hostname, now,  
  55.            heartbeat.getAgentEnv()));    // 向状态机发送更新事件，更新节点至正常状态  
  56.      } else {  
  57.        hostObject.handleEvent(new HostUnhealthyHeartbeatEvent(hostname, now,  
  58.            null));   // 把节点列入不健康  
  59.      }  
  60.      if (hostState != hostObject.getState()) scanner.updateHBaseMaster(hostObject);  // 更新 hbase master状态，如果该节点上在master节点的话，  
  61.    } catch (InvalidStateTransitionException ex) {  
  62.      LOG.warn("Asking agent to reregister due to " + ex.getMessage(), ex);  
  63.      hostObject.setState(HostState.INIT);   // 出错，重新注册  
  64.      return createRegisterCommand();  
  65.    }  
  66.   
  67.    //Examine heartbeat for command reports  
  68.    processCommandReports(heartbeat, hostname, clusterFsm, now);  // 处理状态更改的汇报信息，看进度到哪一点了。  
  69.   
  70.    // Examine heartbeart for component live status reports  
  71.    processStatusReports(heartbeat, hostname, clusterFsm);  // 处理该节点的服务状态信息（也称作为组件）  
  72.   
  73.    // Send commands if node is active  
  74.    if (hostObject.getState().equals(HostState.HEALTHY)) {  
  75.      sendCommands(hostname, response);  //把该节点的命令AgentCommand组装起来，统一返回给agent  
  76.    }  
  77.    return response;  
  78.  }  
 
下面我们学习一下Ambari-Agent是如何处理heartbeat请求的。agent是由Python代码所写，每个节点上都会有一个python的daemon进程与server进行交互。

三、Ambari-Agent执行流程

安装ambari-agent 服务时会把相应在的python代码置于python执行的环境上下文中，例如其入口代码可能是/usr/lib/python2.6/site-packages/ambari_agent/main.py，并且进行相关初始化工作（例如验证参数，与server建立连接，初始化安全验证证书），最后会产生一个新的控制器Controller子线程来统一管理节点的状态。Controller线程里面有一个动作队列ActionQueue线程，并且开启向Server注册和发心跳服务。可以看出来，ambari-agent主要由两个线程组成，Controller线程向Server发送注册或心跳请求，请求到的Action数据放到ActionQueue线程里面，ActionQueue线程维护着两个队列：commandQueue和resultQueue。ActionQueue线程会监听commandQueue的状况。
Python代码  技术图片
  1. class Controller(threading.Thread):    
  2.   def __init__(self, config, range=30):  // 在初始化Controller之前，ambari-agent就会在main.py里面进行判断：ambari-server是否正常，正常才会初始化Controller  
  3.   // 省略初始化代码  
  4.   def run(self):    
  5.     self.actionQueue = ActionQueue(self.config)  // 初始化队列线程  
  6.     self.actionQueue.start()  
  7.     self.register = Register(self.config)  // 初始化注册类   
  8.     self.heartbeat = Heartbeat(self.actionQueue)  // 初始化心跳类  
  9.   
  10.     opener = urllib2.build_opener()  
  11.     urllib2.install_opener(opener)  
  12.   
  13.     while True:  
  14.       self.repeatRegistration = False  
  15.       self.registerAndHeartbeat()    //开始注册 并且 定时发心跳  
  16.       if not self.repeatRegistration:  
  17.         break  
  18.   
  19.     pass  
 CommandQueue队列主要有3类command: 
  1. REGISTER_COMMAND：该类命令主要通知agent重新向server发送注册请求。
  2. STATUS_COMMAND：该类命令主要告诉agent需要向server发送某组件的状态信息。
  3. EXECUTION_COMMAND：要求agent执行puppet或者软件集升级任务
ActionQueue线程在执行STATUS_COMMAND时，会通过LiveStatus类构建一个StatusCheck检测器，并且通过ps命令来检测该组件是否是活着。
Python代码  技术图片
  1. def getIsLive(self, pidPath):  
  2.     // ....  
  3.     //检测该组件pid文件是否存在...  
  4.     res = self.sh.run([‘ps -p‘, str(pid), ‘-f‘])   //运行shell命令，检测该进程是否存在  
  5.     lines = res[‘output‘].strip().split(os.linesep)  
  6.     try:  
  7.       procInfo = lines[1]  
  8.       isLive = not procInfo == None  
  9.     except IndexError:  
  10.       logger.info(‘Process is dead‘)  
  11.     return isLive  
 ActionQueue线程在执行EXECUTION_COMMAND任务时，通常是用于执行相关Puppet任务，它会在[agent].prefix目录下产生一个puppet文件，然后执行puppet apply命令执行一批puppet module文件完成配置更改和节点管理任务。
Python代码  技术图片
  1. def runCommand(self, command, tmpoutfile, tmperrfile):  
  2.     taskId = 0   
  3.     if command.has_key("taskId"):  
  4.       taskId = command[‘taskId‘]  
  5.     siteppFileName = os.path.join(self.tmpDir, "site-" + str(taskId) + ".pp") // self.tmpdir是ambari-agent.ini里面配置的agent.prefix参数, site-{taskId:int}.pp文件里面主要是一组服务的配置参数  
  6.     generateManifest(command, siteppFileName, self.modulesdir, self.config)  //生成一个puppet配置文件  
  7.     result = self.run_manifest(command, siteppFileName, tmpoutfile, tmperrfile) // 会根据command命令里repo_info参数值，执行相应的puppet命令  
  8.     return result  

Ambari深入学习（II）－实现细节

标签：告诉   spring框架   加载   功能   书签   产生   manifest   stun   ogg   

原文地址：https://www.cnblogs.com/felixzh/p/10899575.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！