Ambari架构源码解析

时间  2019-11-08
标签 ambari 架构 源码 解析 栏目 系统架构 繁體版
原文   原文链接

1. Ambari介绍

　　Apache Ambari是一种基于Web的工具，支持Apache Hadoop集群的供应、管理和监控。Ambari已支持大多数Hadoop组件，包括HDFS、MapReduce、Hive、Pig、 Hbase、Zookeeper、Sqoop和Hcatalog等。前端

1.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等；node

1.2 基本组件

Ambari 能够分为 5个大的组件,分别是是 Ambari-server 、 Ambari-web 、 Ambari-agent 、 Ambari-metrics-collector 和 Ambari-metrics-monitor 。python

1. 在集群的每一台机器上都会部署 Ambari-agent 程序。 Agent 主要负责接收来着 Server 端的命令，这些命令能够是安装、启动、中止 Hadoop 集群上的某一服务。同时, agent 端须要向 Ambari-server 端上 报命令执行的结果,是执行成功仍是失败。 ios

2. Ambari-Server 提供 REST 接口给Agent 和 Web 访问，用户甚至能够不用界面,而是经过 curl 命令来操控集群。 web

3. Ambari-metric-collector和 Ambari-metrics-monitor 是收集群中组件 metrics 的模块。spring

1.3 相关技术

Ambari充分利用了一些已有的优秀开源软件，巧妙地把它们结合起来，使其在分布式环境中作到了集群式服务管理能力、监控能力、展现能力，这些优秀的开源软件有： 
（1）agent端，采用了puppet管理节点。 
（2）在web端，采用ember.js做为前端MVC框架和NodeJS相关工具，用handlebars.js做为页面渲染引擎，在CSS/HTML方面还用了Bootstrap框架。 
（3）在Server端，采用了Jetty、Spring、JAX-RS等。 
（4）同时利用了Ganglia、Nagios的分布式监控能力。sql

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能够很好地采集到节点信息。shell

2. Ambari项目目录结构

2.1 整体目录

ambari-server Ambari的Server程序，主要管理部署在每一个节点上的管理监控程序
ambari-agent 部署在监控节点上运行的管理监控程序
ambari-web Ambari页面UI的代码，做为用户与Ambari server交互的。
ambari-views 用于扩展Ambari Web UI中的框架
ambari-common Ambari-server 和Ambari-agent 共用的代码
ambari-metrics 在Ambari所管理的集群中用来收集、聚合和服务Hadoop和系统计量
contrib 自定义第三方库
docs 文档

   

 

 

 

 

 

 

2.2 ambari-server 目录

目录 描述
org.apache.ambari.server.api.services 对web接口的入口方法，处理/api/v1/* 的请求
org.apache.ambari.server.controller 对Ambari中cluster的管理处理，如新增host，更service、删除component等
org.apache.ambari.server.controller.internal 主要存放ResourceProvider和PropertyProvider；
org.apache.ambari.service.orm.* 对数据库的操做
org.apache.ambari.server.agent.rest 处理与Agent的接口的入口方法
org.apache.ambari.security 使用Spring Security来作权限管理

 每一种Resource都对应一个ResourceProvider,以下表所示：数据库

 

Resource.Type ResourceProvider
Workflow WorkflowResourceProvider
Job JobResourceProvider
TaskAttempt TaskAttemptResourceProvider
View ViewResourceProvider
ViewInstance ViewInstanceResourceProvider
Blueprint BlueprintResourceProvider
Cluster ClusterResourceProvider
Service ServiceResourceProvider
Component ComponentResourceProvider
Host HostResourceProvider
HostComponent HostComponentResourceProvider
Configuration ConfigurationResourceProvider
Action ActionResourceProvider
Request RequestResourceProvider
Task TaskResourceProvider
User UserResourceProvider
Stack StackResourceProvider
StackVersion StackVersionResourceProvider
StackService StackServiceResourceProvider
StackServiceComponent StackServiceComponentResourceProvider
StackConfiguration StackConfigurationResourceProvider
OperatingSystem OperatingSystemResourceProvider
Repository RepositoryResourceProvider
RootService RootServiceResourceProvider
RootServiceComponent RootServiceComponentResourceProvider
RootServiceHostComponent RootServiceHostComponentResourceProvider
ConfigGroup ConfigGroupResourceProvider
RequestSchedule RequestScheduleResourceProvider　　

 

2.3 Ambari-agent目录

3. Ambari-server

3.1 ambari-server架构

　　ambari-server是一个有状态的，它维护着本身的一个有限状态机FSM，同时这些状态机存储在数据库中，默认数据库为postgressql数据库。 apache

1. Ambarii-Server提供ambari web，rest api，ambari shell三大方式操做机群； 
2. ambari将集群的配置、各个服务的配置等信息存在ambari server端的DB中； 
3. ambari server与ambari agent的交流走RPC，即agent向server报告心跳，server将command经过respons发回给agent，agent本地执行命令，好比：agent端执行相应的python脚本； 
4. ambari有本身的一套监控、告警、镜像服务，以可插拔的形式供上层服务调用； 

　

 

　　Ambari-Server是一个WEB Server，提供统一的REST API接口，同时向web和agent开放了两个不一样的端口（默认前者是8080, 后者是8440或者8441）。它是由Jetty Server容器构建起来的，经过Spring Framework构建出来的WEB服务器，其中大量采用了google提供的Guice注解完成spring框架所须要的注入功能，REST服务由JAX-RS标准来实现。

 

以下图所示，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经过心跳来得到数据库的变动历史。

 

 

　　Ambari Server 会读取 Stack 和 Service 的配置文件。当用 Ambari 建立集群的时候，Ambari Server 传送 Stack 和 Service 的配置文件以及 Service 生命周期的控制脚本到 Ambari Agent。Agent 拿到配置文件后，会下载安装公共源里软件包（Redhat，就是使用 yum 服务）。安装完成后，Ambari Server 会通知 Agent 去启动 Service。以后 Ambari Server 会按期发送命令到 Agent 检查 Service 的状态，Agent 上报给 Server，并呈如今 Ambari 的 GUI 上。 
Ambari Server 支持 Rest API，这样能够很容易的扩展和定制化 Ambari。甚至于不用登录 Ambari 的 GUI，只须要在命令行经过 curl 就能够控制 Ambari，以及控制 Hadoop 的 cluster。具体的 API 能够参见 Apache Ambari 的官方网页 API reference。 

4. Ambari-agent

4.1 ambari-agent架构

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

  • 采集所在节点的信息而且汇总发心跳汇报给ambari-server
  • 处理ambari-server的执行请求

所以它有两种队列：

  • 消息队列MessageQueue，或为ResultQueue。包括节点状态信息（包括注册信息）和执行结果信息，而且汇总后经过心跳发送给ambari-server;
  • 操做队列ActionQueue。用于接收ambari-server返回过来的状态操做，而后能过执行器按序调用puppet或python脚本等模块完成任务。 

 

4.2 Ambari-agent引导流程

分别是用SSH和人工手动的非SSH 
步骤： 
1. Ambari Server经过调用bootstrap.py来初始化整个bootstrap进程 
2. Server端经过SSH Keys在Agent上配置Ambari Repo：经过scp 命令将Ambari Server上的ambari.repo文件拷贝到Agent Host上。 
3. 复制Ambari Agent Setup script：利用scp命令将setupAgent.py脚本复制到Agent host上。 
4. 在各个Agent上执行Ambari Agent Setup script：SSH到各个Agent Host上而后执行setupAgent.py。 
5. 在Agent上安装epel-release：用yum工具来安装epel-release包 
6. 在Agent上安装Ambari-agent：用yum工具来安装Ambari-Agent包 
7. 配置Ambari-agent.ini：修改/etc/ambari-agent/conf/ambari-agent.ini，并设置agent host上的hostname 
8. 启动Ambari-agent:启动Ambari-agent进程 
9. 开始Ambari Agent注册：agent开始registration进程

人工手动引导 
具体步骤内容基本同上

 

4.3 Agent注册流程

步骤 
1. 链接握手端口8441：Ambari Agent链接到Ambari Server的握手端口8441。 
2. 下载Server Certification：Ambari Agent下载Server Certification。 
3. 请求签署Agent Certification：Ambari Agent请求Ambari Server来签署Agent证书。 
4. 签署Agent Cert：Ambari Server经过密码签署Agent证书。 
5. 下载Agent Cert并断掉链接：Ambari Agent下载Agent证书，而后断掉以前的链接。 
6. 链接注册端口8440：Ambari Agent链接到Ambari Server的注册端口8441 
7. 用Agent Cert执行2WAY auth：在Agent和Server之间完成2WAY权限认证。 
8. 获取FQDN：Ambari Agent host获取Fully Qualified Domain Name（FQDN） 
9. 注册Host：利用FQDN，host向Ambari Server提出注册。 
10. 完成Host注册：Ambari Server完成host的注册过程，把host加入到Ambari数据库 。
11. Agent心跳程序启动：Ambari Agent向Ambari Server开启心跳程序，确认各类命令的执行 。

5. Ambari-web内部架构

Ambari-web使用了一个流行的前端Embar.js MVC框架实现，Embar.js是一个TodoMVC框架，它涵盖了现今典型的单页面应用（single page application）几乎全部的行为。

使用了nodejs

使用brunch 做为项目的构建管理工具

Brunch ,是一个超快的HTML5构建工具。它有以下功能：

（1）编译你的脚本、模板、样式、连接它们。

（2）将脚本和模板封装进common.js/AMD模块里，连接脚本和样式。

（3）为连接文件生成源地图，复制资源和静态文件。

（4）经过缩减代码和优化图片来收缩输出，看管你的文件更改。

（5）并经过控制台和系统提示通知你错误。

Nodejs 是一个基于Chrome JavaScript运行时创建的一个平台，用来方便的搭建快速的易于扩展的网络应用，NodeJS借助事件驱动，非阻塞I/O模型变得轻量和高效，很是适合运行在分布式设备的数据密集型的实时应用。

6. 源码分析

6.1 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;
  }

  ...
}

6.2 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、获取指标流程：

  1. jersy接口接收到请求，建立一个ResourceInstance实例；
  2. 解析http请求构造一个Request对象，而后交给reques的process()方法来处理；
  3. reques解析url或http_body获得一个Predicate对象；
  4. 根据http类型获取handler，GET请求对应ReadHandler；
  5. handler向Query对象中添加分页、Render、Predicate等属性后，而后让query.execute()；
  6. 根据Resource.Type得到对应的ResourceProvider对象，调用其getResources方法获得Set；
  7. 调用对应的PropertyProvider填充Resource；
  8. 处理结果，放回json结果

Ambari-Server启动

Ambari-Server接受来自两处的REST请求，Agent过来的请求处理逻辑由包org.apache.ambari.server.agent处理， 而API所的处理逻辑来自org.apache.ambari.server.api。详见以下代码：

“`

“`

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

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程