Ambari架构源码解析

一、Ambari介绍

Ambari是hadoop分布式集群配置管理工具，是由hortonworks主导的开源项目。它已经成为apache基金会的孵化器项目，已经成为hadoop运维系统中的得力助手。

1.1 组成

Ambari充分利用了一些已有的优秀开源软件，巧妙地把它们结合起来，使其在分布式环境中做到了集群式服务管理能力、监控能力、展示能力，这些优秀的开源软件有：

（1）agent端，采用了puppet管理节点；

（2）在web端，采用ember.js作为前端MVC框架和NodeJS相关工具，用handlebars.js作为页面渲染引擎，在CSS/HTML方面还用了Bootstrap框架。

（3）在Server端，采用了Jetty、Spring、JAX-RS等。

（4）同时利用了Ganglia、Nagios的分布式监控能力。

1.2 架构

Ambari架构采用的是Server/Client的模式，主要由两部分组成：ambari-agent和ambari-server。ambari依赖其它已经成熟的工具，例如其ambari-server 就依赖python，而ambari-agent还同时依赖ruby, puppet，facter等工具，还有它也依赖一些监控工具nagios和ganglia用于监控集群状况。其中：

1. puppet是分布式集群配置管理工具，也是典型的Server/Client模式，能够集中式管理分布式集群的安装配置部署，主要语言是ruby。

2. facter是用python写的一个节点资源采集库，用于采集节点的系统信息，例如OS信息，主机信息等。由于ambari-agent主要是用python写的，因此用facter可以很好地采集到节点信息。

1.2.1 Nagios

Nagios是一款开源的免费网络监视工具，能有效监控Windows、Linux和Unix的主机状态，交换机路由器等网络设备，打印机等。在系统或服务状态异常时发出邮件或短信报警第一时间通知网站运维人员，在状态恢复后发出正常的邮件或短信通知。

1.2.2 ganglia

Ganglia是UC Berkeley发起的一个开源集群监视项目，设计用于测量数以千计的节点。Ganglia的核心包含gmond、gmetad以及一个Web前端。主要是用来监控系统性能，如：cpu 、mem、硬盘利用率， I/O负载、网络流量情况等，通过曲线很容易见到每个节点的工作状态，对合理调整、分配系统资源，提高系统整体性能起到重要作用。

1.3 Technology Stack技术堆栈

1.3.1 Ambari Server

• Server code: Java 1.7 / 1.8
• Agent scripts: Python
• Database: Postgres, Oracle, MySQL
• ORM: EclipseLink
• Security: Spring Security with remote LDAP integration and local database
• REST server: Jersey (JAX-RS)
• Dependency Injection: Guice
• Unit Testing: JUnit
• Mocks: EasyMock
• Configuration management: Python
• Frontend code: JavaScript
• Client-side MVC framework: Ember.js / AngularJS
• Templating: Handlebars.js (integrated with Ember.js)
• DOM manipulation: jQuery
• Look and feel: Bootstrap 2
• CSS preprocessor: LESS
• Unit Testing: Mocha
• Mocks: Sinon.js
• Application assembler/tester: Brunch / Grunt / Gulp

1.3.1 Ambari Web

二、源码部分

2.1 基本概念

1. Resource：Ambari把可以被管理的资源的抽象为一个Resource实例，资源可以包括服务、组件、主机节点等，一个resource实例中包含了一系列该资源的属性；
2. Property：服务组件的指标名称；
3. ResourceProvider和PropertyProvider分别对应Resource和Property的提供方，获取指标需要先获取Resource，然后获取Property对应的metric；
4. Query:Query是Resource的内部对象，代表了对该资源的操作；
5. Request：一个Request代表了对Resource的操作请求，包含http信息及要操作的Resource的实例，Request按照http的请求方式分为四种：GET、PUT、DELETE、POST；
6. Predicate：一个Predicate代表了一系列表达式，如and、or等；
7. org.apache.ambari.server.api.services：对web接口的入口方法，处理/api/v1/*的请求；
8. org.apache.ambari.server.controller：对Ambari中cluster的管理操作，如新增host，更新service、删除Component等；
9. org.apache.ambari.server.controller.internal：主要存放ResourceProvider和PropertyProvider；
10. org.apache.ambari.server.orm.*：数据库操作相关；
11. org.apache.ambari.server.agent.rest：处理与Agent的接口的入口方法；
12. jersy接口接收到请求，创建一个ResourceInstance实例；
13. 解析http请求构造一个Request对象，然后交给reques的process()方法来处理；
14. reques解析url或http_body得到一个Predicate对象；
15. 根据http类型获取handler，GET请求对应ReadHandler；
16. handler向Query对象中添加分页、Render、Predicate等属性后，然后让query.execute()；
17. 根据Resource.Type获得对应的ResourceProvider对象，调用其getResources方法得到Set；
18. 调用对应的PropertyProvider填充Resource；
19. 处理结果，放回json结果

2.2 源代码结构

2.3 获取指标流程：

2.3.1 Ambari-server

ambari-server是一个有状态的，它维护着自己的一个有限状态机FSM。同时这些状态机存储在数据库中，前期数据库主要采用postgres。如下图所示，server端主要维护三类状态：

1. Live Cluster State：集群现有状态，各个节点汇报上来的状态信息会更改该状态;

2. Desired State：用户希望该节点所处状态，是用户在页面进行了一系列的操作，需要更改某些服务的状态，这些状态还没有在节点上产生作用;

3. Action State：操作状态，是状态改变时的请求状态，也可以看作是一种中间状态，这种状态可以辅助Live Cluster State向Desired State状态转变。

Ambari-server的Heartbeat Handler模块用于接收各个agent的心跳请求（心跳请求里面主要包含两类信息：节点状态信息和返回的操作结果），把节点状态信息传递给FSM状态机去维护着该节点的状态，并且把返回的操作结果信息返回给Action Manager去做进一步的处理。

Coordinator模块又可以称为API handler，主要在接收WEB端操作请求后，会检查它是否符合要求，stage planner分解成一组操作，最后提供给Action Manager去完成执行操作。

因此，从上图就可以看出，Ambari-Server的所有状态信息的维护和变更都会记录在数据库中，用户做一些更改服务的操作都会在数据库上做一些相应的记录，同时，agent通过心跳来获得数据库的变更历史。

2.3.2 Ambari-Agent

ambari-agent是一个无状态的。其功能主要分两部分：

- 采集所在节点的信息并且汇总发心跳汇报给ambari-server;

- 处理ambari-server的执行请求。

因此它有两种队列：

- 消息队列MessageQueue，或为ResultQueue。包括节点状态信息（包括注册信息）和执行结果信息，并且汇总后通过心跳发送给ambari-server;

- 操作队列ActionQueue。用于接收ambari-server返回过来的状态操作，然后能过执行器按序调用puppet或python脚本等模块完成任务。

三、实现细节

3.1 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。

Ambari-Server有一个状态机管理模块，所有节点的状态信息更改都最终提供给状态机进行更改操作，因此状态机是一个很忙的组件。在Ambari-Server里面，把每一次更改操作都把它当作是一类事件，采用事件驱动机制完成对应的任务。这种思想有点借鉴已经运用在hadoop 2.x YARN里面的事件驱动机制。事件驱动机制能够一种高效的异步RPC请求方式，直接调用需要执行相应的代码逻辑，而事件驱动只需要产生事件统一提交给事件处理器，因此事件驱动需要一个更复杂的有限状态机结合起来一同使用。

3.2 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。

public class HeartBeatHandler {

...

public HeartBeatResponse handleHeartBeat(HeartBeat heartbeat)

throws AmbariException {

long now = System.currentTimeMillis();

if (heartbeat.getAgentEnv() != null && heartbeat.getAgentEnv().getHostHealth() != null) {

heartbeat.getAgentEnv().getHostHealth().setServerTimeStampAtReporting(now);

}

String hostname = heartbeat.getHostname();

Long currentResponseId = hostResponseIds.get(hostname);

HeartBeatResponse response;

if (currentResponseId == null) {

//Server restarted, or unknown host.

LOG.error("CurrentResponseId unknown for " + hostname + " - send register command");

// 无responseId, 新请求，就进行注册, responseId ＝0

return createRegisterCommand();

}

LOG.debug("Received heartbeat from host"

+ ", hostname=" + hostname

+ ", currentResponseId=" + currentResponseId

+ ", receivedResponseId=" + heartbeat.getResponseId());

if (heartbeat.getResponseId() == currentResponseId - 1) {

LOG.warn("Old responseId received - response was lost - returning cached response");

return hostResponses.get(hostname);

} else if (heartbeat.getResponseId() != currentResponseId) {

LOG.error("Error in responseId sequence - sending agent restart command");

// 心跳是历史记录，那么就要求其重启，重新注册,responseId 不变

return createRestartCommand(currentResponseId);

}

response = new HeartBeatResponse();

//responseId 加 1 , 返回一个新的responseId，下次心跳又要把这个responseId带回来。

response.setResponseId(++currentResponseId);

Host hostObject;

try {

hostObject = clusterFsm.getHost(hostname);

} catch (HostNotFoundException e) {

LOG.error("Host: {} not found. Agent is still heartbeating.", hostname);

if (LOG.isDebugEnabled()) {

LOG.debug("Host associated with the agent heratbeat might have been " +

"deleted", e);

}

// For now return empty response with only response id.

return response;

}

//失去心跳，要求重新注册, responseId=0

if (hostObject.getState().equals(HostState.HEARTBEAT_LOST)) {

// After loosing heartbeat agent should reregister

LOG.warn("Host is in HEARTBEAT_LOST state - sending register command");

return createRegisterCommand();

}

hostResponseIds.put(hostname, currentResponseId);

hostResponses.put(hostname, response);

// If the host is waiting for component status updates, notify it

//如果主机正在等待组件状态更新，请通知它

//节点已经进行了注册，但是该节点还没有汇报相关状态信息,等待服务状态更新

if (heartbeat.componentStatus.size() > 0

&& hostObject.getState().equals(HostState.WAITING_FOR_HOST_STATUS_UPDATES)) {

try {

LOG.debug("Got component status updates");

//更新服务状态机

hostObject.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now));

} catch (InvalidStateTransitionException e) {

LOG.warn("Failed to notify the host about component status updates", e);

}

}

if (heartbeat.getRecoveryReport() != null) {

RecoveryReport rr = heartbeat.getRecoveryReport();

processRecoveryReport(rr, hostname);

}

try {

if (heartbeat.getNodeStatus().getStatus().equals(HostStatus.Status.HEALTHY)) {

//向状态机发送更新事件，更新节点至正常状态

hostObject.handleEvent(new HostHealthyHeartbeatEvent(hostname, now,

heartbeat.getAgentEnv(), heartbeat.getMounts()));

} else { // 把节点列入不健康

hostObject.handleEvent(new HostUnhealthyHeartbeatEvent(hostname, now, null));

}

} catch (InvalidStateTransitionException ex) {

LOG.warn("Asking agent to re-register due to " + ex.getMessage(), ex);

hostObject.setState(HostState.INIT);

return createRegisterCommand();

}

/** * A host can belong to only one cluster. Though getClustersForHost(hostname) * returns a set of clusters, it will have only one entry. *主机只能属于一个集群。 通过getClustersForHost(hostname)返回一组集群，它只有一个条目。 * * TODO: Handle the case when a host is a part of multiple clusters. * 处理 主机是多个集群的一部分时的 情况。 */

Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);

if (clusters.size() > 0) {

String clusterName = clusters.iterator().next().getClusterName();

if (recoveryConfigHelper.isConfigStale(clusterName, hostname, heartbeat.getRecoveryTimestamp())) {

RecoveryConfig rc = recoveryConfigHelper.getRecoveryConfig(clusterName, hostname);

response.setRecoveryConfig(rc);

if (response.getRecoveryConfig() != null) {

LOG.info("Recovery configuration set to {}", response.getRecoveryConfig().toString());

}

}

}

heartbeatProcessor.addHeartbeat(heartbeat);

// Send commands if node is active

if (hostObject.getState().equals(HostState.HEALTHY)) {

sendCommands(hostname, response);

annotateResponse(hostname, response);

}

return response;

}

...

}

3.3 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的状况。

class Controller(threading.Thread):

def __init__(self, config, range=30):

// 在初始化Controller之前，ambari-agent就会在main.py里面进行判断：ambari-server是否正常，正常才会初始化Controller

// 省略初始化代码

def run(self):

try:

// 初始化队列线程

self.actionQueue = ActionQueue(self.config, controller=self)

self.actionQueue.start()

// 初始化注册类

self.register = Register(self.config)

// 初始化心跳类

self.heartbeat = Heartbeat(self.actionQueue, self.config, self.alert_scheduler_handler.collector())

opener = urllib2.build_opener()

urllib2.install_opener(opener)

while True:

self.repeatRegistration = False

//开始注册 并且 定时发心跳

self.registerAndHeartbeat()

if not self.repeatRegistration:

logger.info("Finished heartbeating and registering cycle")

break

except:

logger.exception("Controller thread failed with exception:")

raise

logger.info("Controller thread has successfully finished")

CommandQueue队列主要有3类command:

1. REGISTER_COMMAND：该类命令主要通知agent重新向server发送注册请求；

2. STATUS_COMMAND：该类命令主要告诉agent需要向server发送某组件的状态信息；

3. EXECUTION_COMMAND：要求agent执行puppet或者软件集升级任务；

3.4 Ambari-agent

ambari-agent是一个无状态的，主要功能如下所示：

• 采集所在节点的信息并且汇总发送心跳给ambari-server
• 处理ambari-server的响应请求

因而，它有两种队列：MessageQueue和ActionQueue。

• MessageQueue：包含节点状态信息（注册信息等）和执行结果信息，并且汇总后通过心跳发送给ambari-server
• ActionQueue：用于接收ambari-server返回过来的状态操作，然后能通过执行器按序调用puppet或python脚本等模块完成任务

架构图如下所示：

3.5  Ambari-server

而对于ambari-server来说，其是一个有状态的，它维护着自己的一个有限状态FSM。同时这些状态存储与数据库当中（DB目前可以支持多种，可按序自选），Server端主要维持三类状态：

• Live Cluster State：集群现有状态，各个节点汇报上来的状态信息会更改该状态
• Desired State：使用者希望该节点所处状态，是用户在页面进行了一系列的操作，需要更改某些服务的状态，这些状态还没有在节点商阐述作用
• Action State：操作状态，该状态是一种中间状态，这种状态可以辅助Live Cluster State向Desired State状态的转变

其架构图如下所示：

ambari-server的Heartbeat Handler模块用于接收各个Agent的心跳请求（其中包含节点状态信息和返回的操作结果），把节点状态信息传递给图中的FSM模块去维护该节点的状态，并把响应之后的操作结果信息返回给Action Manager去做更加详细的处理。Coordinator模块可以看作API Handler，主要在接收Web端操作请求后，校验其合法性，Stage Planner分解成一组操作，最后提供给Action 过 Manager去完成执行操作。

因而，从上图中，我们可以看出，ambari-server的所有状态信息的维护和变化都会被记录在数据库当中，使用者做一些更改服务的操作都会在数据库商做对应的记录，同时，Agent通过心跳来获取数据库的变动历史信息。