Yarn之ResourceManager详细分析笔记（一）待续

一、概述
本文将介绍ResourceManager在Yarn中的功能作用，从更细的粒度分析RM内部组成的各个组件功能和他们相互的交互方式。
二、ResourceManager的交互协议与基本职能
1、ResourceManager交互协议
在整个Yarn框架中主要涉及到7个协议，分别是ApplicationClientProtocol、MRClientProtocol、ContainerManagementProtocol、ApplicationMasterProtocol、ResourceTracker、LocalizationProtocol、TaskUmbilicalProtocol，这些协议封装了各个组件交互的信息。ResourceManager现实功能需要和NodeManager以及ApplicationMaster进行信息交互，其中涉及到的RPC协议有ResourceTrackerProtocol、ApplicationMasterProtocol和ResourceTrackerProtocol。

ResourceTracker

NodeManager通过该协议向ResourceManager中注册、汇报节点健康情况以及Container的运行状态，并且领取ResourceManager下达的重新初始化、清理Container等命令。NodeManager和ResourceManager这种RPC通信采用了和MRv1类似的“pull模型”（ResourceManager充当RPC server角色，NodeManager充当RPC client角色），NodeManager周期性主动地向ResourceManager发起请求，并且领取下达给自己的命令。
ApplicationMasterProtocol

应用程序的ApplicationMaster同过该协议向ResourceManager注册、申请和释放资源。该协议和上面协议同样也是采用了“pull模型”，其中在RPC机制中，ApplicationMaster充当RPC client角色，ResourceManager充当RPC server角色。

ApplicationClientProtocol

客户端通过该协议向ResourceManager提交应用程序、控制应用程序（如杀死job）以及查询应用程序的运行状态等。在该RPC 协议中应用程序客户端充当RPC client角色，ResourceManager充当RPC server角色。

整理一下ResourceManager与NodeManager、ApplicationMaster和客户端RPC协议交互的信息：



上图中的ResourceTrackeServer、ApplicationMasterService 、ClientRMServer是ResourceManager中处理上述功能的组件。
1、ResourceManager基本职能
ResourceManager基本职能概括起来就以下几方面：
与客户端进行交互，处理来自于客户端的请求，如查询应用的运行情况等。

启动和管理各个应用的ApplicationMaster，并且为ApplicationMaster申请第一个Container用于启动和在它运行失败时将它重新启动。

管理NodeManager，接收来自NodeManager的资源和节点健康情况汇报，并向NodeManager下达管理资源命令，例如kill掉某个container。

资源管理和调度，接收来自ApplicationMaster的资源申请，并且为其进行分配。这个是它的最重要的职能。

三、ResourceManager内部组成架构分析
ResourceManager在底层代码实现上将各个功能模块分的比较细，各个模块功能具有很强的独立性。下图所示的是ResourceManager中的大概的功能模块组成：



1、用户交互模块
用户交互模块即上图显示的User Service管理模块。在这里边还可以看到根据不同的用户类型启用了不同的服务进行处理，AdminService处理管理员相关请求，ClientRMService处理普通客户相关请求，这样使得管理员不会因为普通客户请求太多而造成堵塞。下面看看这2个服务的具体实现代码：

ClientRMService

public class ClientRMService extends AbstractService implements
ApplicationClientProtocol {
private static final ArrayList<ApplicationReport> EMPTY_APPS_REPORT = new ArrayList<ApplicationReport>();

private static final Log LOG = LogFactory.getLog(ClientRMService.class);

final private AtomicInteger applicationCounter = new AtomicInteger(0);
final private YarnScheduler scheduler;//调度器
final private RMContext rmContext;//RM上下文对象，其包含了RM大部分运行时信息，如节点列表、队列列表、应用程序列表等
private final RMAppManager rmAppManager;//app管理对象

private Server server;//一个RPC Server
protected RMDelegationTokenSecretManager rmDTSecretManager;

private final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
InetSocketAddress clientBindAddress;
//访问控制对象，例如，一些应用程序在提交时设置了查看权限的话，其他普通用户就无法查看。
private final ApplicationACLsManager applicationsACLsManager;
private final QueueACLsManager queueACLsManager;
......
@Override
protected void serviceStart() throws Exception {
Configuration conf = getConfig();
YarnRPC rpc = YarnRPC.create(conf);
this.server =   //实现RPC协议ApplicationClientProtocol
rpc.getServer(ApplicationClientProtocol.class, this,
clientBindAddress,
conf, this.rmDTSecretManager,
conf.getInt(YarnConfiguration.RM_CLIENT_THREAD_COUNT,
YarnConfiguration.DEFAULT_RM_CLIENT_THREAD_COUNT));

// Enable service authorization?
if (conf.getBoolean(
CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
false)) {
refreshServiceAcls(conf, new RMPolicyProvider());
}

this.server.start();
......
}

从上面ClientRMService的基本代码架构我们可以看出：
（1）ClientRMService是一个RPC Server，主要为来自于普通客户端的各种RPC请求。从代码实现的角度看，它是ApplicationClientProtocol协议的一个实现。
（2）之前我们已经说了，普通用户可以通过该服务来获得正在运行应用程序的相关信息，如进度情况、应用程序列表等。上面代码中都将ResourceManager运行信息封装在RMContxt接口中了，下面来看看这个接口的一个实现对象RMContextImpl:

public class RMContextImpl implements RMContext {
//中央异步调度器。RM中的各个服务和组件以及它们处理和输出的事件类型都是通过中央异步调度器组织在一起的，这样可以有效提高系统的吞吐量。
private final Dispatcher rmDispatcher;

private final ConcurrentMap<ApplicationId, RMApp> applications//应用程序列表
= new ConcurrentHashMap<ApplicationId, RMApp>();

private final ConcurrentMap<NodeId, RMNode> nodes//节点列表
= new ConcurrentHashMap<NodeId, RMNode>();

private final ConcurrentMap<String, RMNode> inactiveNodes//非活跃节点列表
= new ConcurrentHashMap<String, RMNode>();
//正在运行中的AP心跳监控对象
private AMLivelinessMonitor amLivelinessMonitor;//正在运行中的AP心跳监控对象
//运行完毕后的AM心跳监控对象
private AMLivelinessMonitor amFinishingMonitor;
//用于存储ResourceManager运行状态
private RMStateStore stateStore = null;
//用于Container的超时监控，应用程序必须在一定时间内（默认10Min）使用分配到的Container去运行task，否则会被回收
private ContainerAllocationExpirer containerAllocationExpirer;
//下面变量都是与安全管理相关的对象
private final DelegationTokenRenewer delegationTokenRenewer;
private final AMRMTokenSecretManager amRMTokenSecretManager;
private final RMContainerTokenSecretManager containerTokenSecretManager;
private final NMTokenSecretManagerInRM nmTokenSecretManager;
private final ClientToAMTokenSecretManagerInRM clientToAMTokenSecretManager;
private ClientRMService clientRMService;
private RMDelegationTokenSecretManager rmDelegationTokenSecretManager;
......
}

AdminService

AdminService和ClientRMService一样都是作为RPC的服务端，它针对的处理管理员RPC请求，负责访问权限的控制，中Yarn中管理员权限的设定可以在yarn-site.xml中yarn.admi.acl项进行设置，该项的默认值是*，也就是说如果不进行设置的话就当所有的用户都是管理员。从代码上看，它是ResourceManagerAdministrationProtocol协议的一个实现：

public class AdminService extends AbstractService implements ResourceManagerAdministrationProtocol {

private static final Log LOG = LogFactory.getLog(AdminService.class);

private final Configuration conf;
private final ResourceScheduler scheduler;
private final RMContext rmContext;
private final NodesListManager nodesListManager;

private final ClientRMService clientRMService;
private final ApplicationMasterService applicationMasterService;
private final ResourceTrackerService resourceTrackerService;

private Server server;
private InetSocketAddress masterServiceAddress;
private AccessControlList adminAcl;

private final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null);
.....
}

AdminService代码和ClientRMService比较相似，它各类功能对象也差不多。
2、NodeManager管理
NodeManager主要是通过NMLivelinessMonitor、ResourceTrackerService和NodeListManager这3大组件来对NodeManager的生命周期、心跳处理以及黑名单处理。
（1）ResourceTrackerService
ResourceTrackerService是RPC协议ResourceTracker的一个实现，它作为一个RPC Server端接收NodeManager的RPC请求，请求主要包含2种信息，注册NodeManager和处理心跳信息。NodeManger启动时第一件事就是像ResourceManager注册，注册时NodeManager发给ResourceTrackerService的RPC包主要包含NodeManager所在节点的可用资源总量、对外开放的htpp端口、节点的host和port等信息，具体代码看ResourceTrackerService#registerNodeManager方法：

@SuppressWarnings("unchecked")
@Override
public RegisterNodeManagerResponse registerNodeManager(
RegisterNodeManagerRequest request) throws YarnException,
IOException {

NodeId nodeId = request.getNodeId();//从NodeManager带来的NodeID
String host = nodeId.getHost();//NodeManager所在节点的host
int cmPort = nodeId.getPort(); //NodeManager所在节点的port
int httpPort = request.getHttpPort();//对外开放的http端口
Resource capability = request.getResource();//获得NodeManager所在节点的资源上限

RegisterNodeManagerResponse response = recordFactory
.newRecordInstance(RegisterNodeManagerResponse.class);

// Check if this node is a 'valid' node
//检测节点host名称的的合法性
if (!this.nodesListManager.isValidNode(host)) {
String message =
"Disallowed NodeManager from  " + host
+ ", Sending SHUTDOWN signal to the NodeManager.";
LOG.info(message);
response.setDiagnosticsMessage(message);
response.setNodeAction(NodeAction.SHUTDOWN);
return response;
}
.....
}

ResourceTrackerService另外一种功能就是处理心跳信息了，当NodeManager启动后，它会周期性地调用RPC函数ResourceTracker#nodeHeartbeat汇报心跳，心跳信息主要包含该节点的各个Container的运行状态、正在运行的Application列表、节点的健康状况等，随后ResourceManager为该NodeManager返回需要释放的Container列表、Application列表等信息。其中心跳信息处理的流程：首先，从NodeManager发来的心跳包中获得节点的状态状态信息，然后检测该节点是否已经注册过，然后检测该节点的host名称是否合法，例如是否在excluded列表中，然后再检测该次心跳是不是第一次心跳信息，这点非常重要，因为关系到心跳的重复发送与应答的相关问题。其实ResourceTrackerService和NodeManager的心跳处理机制和之前Hadoop1.x中的JobTracker与TaskTacker之间的心跳处理很想象，具体请看我之前写的一篇blog:http://zengzhaozheng.blog.51cto.com/8219051/1359887 ，再然后，为NodeManager返回心跳应答信息，最后，想RMNode发送该NodeManager的状态信息并且保存最近一次心跳应答信息。再具体看看ResourceTracker#nodeHeart方法：

public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)
throws YarnException, IOException {
//从RPC Clinet中获得nodeManager所在节点的健康状况
NodeStatus remoteNodeStatus = request.getNodeStatus();
/**
* Here is the node heartbeat sequence...
* 1. Check if it's a registered node
* 2. Check if it's a valid (i.e. not excluded) node
* 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
* 4. Send healthStatus to RMNode
*/

NodeId nodeId = remoteNodeStatus.getNodeId();

// 1. Check if it's a registered node
RMNode rmNode = this.rmContext.getRMNodes().get(nodeId);
if (rmNode == null) {
/* node does not exist */
String message = "Node not found resyncing " + remoteNodeStatus.getNodeId();
LOG.info(message);
resync.setDiagnosticsMessage(message);
return resync;
}

// Send ping
this.nmLivelinessMonitor.receivedPing(nodeId);

// 2. Check if it's a valid (i.e. not excluded) node
if (!this.nodesListManager.isValidNode(rmNode.getHostName())) {
String message =
"Disallowed NodeManager nodeId: " + nodeId + " hostname: "
+ rmNode.getNodeAddress();
LOG.info(message);
shutDown.setDiagnosticsMessage(message);
this.rmContext.getDispatcher().getEventHandler().handle(
new RMNodeEvent(nodeId, RMNodeEventType.DECOMMISSION));
return shutDown;
}

// 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
NodeHeartbeatResponse lastNodeHeartbeatResponse = rmNode.getLastNodeHeartBeatResponse();
if (remoteNodeStatus.getResponseId() + 1 == lastNodeHeartbeatResponse
.getResponseId()) {
LOG.info("Received duplicate heartbeat from node "
+ rmNode.getNodeAddress());
return lastNodeHeartbeatResponse;
} else if (remoteNodeStatus.getResponseId() + 1 < lastNodeHeartbeatResponse
.getResponseId()) {
String message =
"Too far behind rm response id:"
+ lastNodeHeartbeatResponse.getResponseId() + " nm response id:"
+ remoteNodeStatus.getResponseId();
LOG.info(message);
resync.setDiagnosticsMessage(message);
// TODO: Just sending reboot is not enough. Think more.
this.rmContext.getDispatcher().getEventHandler().handle(
new RMNodeEvent(nodeId, RMNodeEventType.REBOOTING));
return resync;
}

// Heartbeat response
NodeHeartbeatResponse nodeHeartBeatResponse = YarnServerBuilderUtils
.newNodeHeartbeatResponse(lastNodeHeartbeatResponse.
getResponseId() + 1, NodeAction.NORMAL, null, null, null, null,
nextHeartBeatInterval);
rmNode.updateNodeHeartbeatResponseForCleanup(nodeHeartBeatResponse);

populateKeys(request, nodeHeartBeatResponse);

// 4. Send status to RMNode, saving the latest response.
this.rmContext.getDispatcher().getEventHandler().handle(
new RMNodeStatusEvent(nodeId, remoteNodeStatus.getNodeHealthStatus(),
remoteNodeStatus.getContainersStatuses(),
remoteNodeStatus.getKeepAliveApplications(), nodeHeartBeatResponse));

return nodeHeartBeatResponse;
}

（2）NodeListManager
NodeListManager主要分管黑名单（include列表）和白名单（exlude列表）管理功能，分别有yarnresouecemanager.nodes.include-path和yarnresourcemanager.nodes.exclude-path指定。黑名单列表中的nodes不能够和RM直接通信（直接抛出RPC异常），管理员可以对这两个列表进行编辑，然后使用$HADOOP_HOME/bin/yarn rmadmin-refreshNodes动态加载修改后的列表，使之生效。

（3）NMLivelinessMonitor
NMLivelinessMonitor主要是分管心跳异常请求。该服务会周期性地遍历集群中的所有NodeManager，如果某个NodeManager在一定时间内（默认10min，可以有参数yarn.nm.liveness-monitor.expiry-interval-ms配置）没有进行心跳汇报，那么则认为它已经死掉，同时在该节点上运行的Container也会被置为运行失败释放资源。那么这些被置为失败的Container是不会直接被RM分配执行的，RM只是负责将这些被置为失败的Container信息告诉它们所对应的ApplicationMaster，需不需要重新运行它说的算，如果需要从新运行的话，该ApplicationMaster要从新向RM申请资源，然后由ApplicationMaster与对应的NodeManager通信以从新运行之前失败的Container。
2、ApplicationMaster管理模块
后面部分请看：http://zengzhaozheng.blog.51cto.com/8219051/1542067