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

Posted 2021-12-02 felixzh

在第一节中，我们简单讲了一下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