1. 云图网首页
  2. 技术专区
  3. 大数据

YARN源码分析(二)—–ResourceManager中的NM节点管理详解大数据

• 大数据

前言

继上一篇文章讲述完ApplicationMaster的相关用法,核心主题都是围绕着2个字”应用”,当然在RM中还有另外一项比较重要的服务也很重要,他就是节点管理服务,在RM中是如何维系管理多个节点,对于应用管理的话,在RM中已经有了ApplicationMasterService这个服务对象了,那么对应于节点NodeManager来说,难道叫做NodeManagerService吗,听起来非常顺,其实他叫做?ResourceTrackerService,当然名称叫什么都无所谓啦,他扮演的功能就是类似于节点NodeManager大管家的角色了.OK,在这里我们就以NodeManager管理为核心线索,逐步分析RM在此方面的设计思想.

相关涉及类

在分析之前,还是需要了解一下相关类,在阅读本篇文章之前,可以建议大家阅读我的上一篇文章ApplicationMaster文章的分析,因为NM和AM管理许多思想共同,也有共同的父类,比如AbstractService这样的抽象服务类.下面是我归纳出的几个类.

1.NodeManager.java–节点管理类,这个类是yarn-resourcemanager包中的类,不是yarn-nodemanager中的同名类,这个类是本篇文章的核心角色类,

2.NodesListManager–节点列表管理类，这个类中管理了类似黑名单，白名单的节点列表形式。

3.NMLivelinessMonitor–节点存活状态监控线程类，与之前的AMLivelinessMonitor线程的原理类似，最简单的心跳更新检查。

4.ResourceTrackerService–节点服务管理对象，负责与各个NodeManager通信。包括NM在此服务上的注册请求处理，心跳更新操作等等。

下面是一张结构简图帮助大家宏观上理解RM中的NM管理：

YARN源码分析(二)-----ResourceManager中的NM节点管理详解大数据

NodeManager节点注册

我们从一个比较初始的状态出发，比如说节点注册开始，一步步的贯穿的去分析整个流程。节点注册操作，在NodeManager类自身中。这个类中定义的基本信息如下

[java] 
view plain
copy
print
?

  1. //ResourceManager下资源管理器类  
  2. public class NodeManager implements ContainerManagementProtocol {  
  3.   private static final Log LOG = LogFactory.getLog(NodeManager.class);  
  4.   private static final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);  
  5.     
  6.   final private String containerManagerAddress;  
  7.   //节点通信地址  
  8.   final private String nodeHttpAddress;  
  9.   //所在机架名称  
  10.   final private String rackName;  
  11.   //节点ID  
  12.   final private NodeId nodeId;  
  13.   final private Resource capability;  
  14.   Resource available = recordFactory.newRecordInstance(Resource.class);  
  15.   Resource used = recordFactory.newRecordInstance(Resource.class);  

注册操作并没有独立出方法来，而是包含在了构造函数中，也就是说，当你构造新的NodeManager的时候，你已经在注册节点到ResourceTrackerService。

[java] 
view plain
copy
print
?

  1. public NodeManager(String hostName, int containerManagerPort, int httpPort,  
  2.       String rackName, Resource capability,  
  3.       ResourceTrackerService resourceTrackerService, RMContext rmContext)  
  4.       throws IOException, YarnException {  
  5.     this.containerManagerAddress = hostName + “:” + containerManagerPort;  
  6.     this.nodeHttpAddress = hostName + “:” + httpPort;  
  7.     this.rackName = rackName;  
  8.     this.resourceTrackerService = resourceTrackerService;  
  9.     this.capability = capability;  
  10.     Resources.addTo(available, capability);  
  11.   
  12.     this.nodeId = NodeId.newInstance(hostName, containerManagerPort);  
  13.     //新建nodemanager注册请求  
  14.     RegisterNodeManagerRequest request = recordFactory  
  15.         .newRecordInstance(RegisterNodeManagerRequest.class);  
  16.     //往请求内写入状态信息  
  17.     request.setHttpPort(httpPort);  
  18.     request.setNodeId(this.nodeId);  
  19.     request.setResource(capability);  
  20.     request.setNodeId(this.nodeId);  
  21.     //调用resourceTrackerService服务对象进行节点注册操作  
  22.     resourceTrackerService.registerNodeManager(request);  
  23.     this.schedulerNode = new FiCaSchedulerNode(rmContext.getRMNodes().get(  
  24.         this.nodeId), false);  
  25.    …..  
  26.   }  

顺着这行代码，来看一下服务端处理注册请求的方法。服务端的类对象是ResourceTrackerService。

[java] 
view plain
copy
print
?

  1. //节点资源跟踪服务,与各个节点的NodeManager通信服务  
  2. public class ResourceTrackerService extends AbstractService implements  
  3.     ResourceTracker {  
  4.   
  5.   private static final Log LOG = LogFactory.getLog(ResourceTrackerService.class);  
  6.   
  7.   private static final RecordFactory recordFactory =   
  8.     RecordFactoryProvider.getRecordFactory(null);  
  9.   //资源管理器上下文  
  10.   private final RMContext rmContext;  
  11.   //节点列表管理器  
  12.   private final NodesListManager nodesListManager;  
  13.   //节点存活状态监控  
  14.   private final NMLivelinessMonitor nmLivelinessMonitor;  
  15.   //节点安全认证相关  
  16.   private final RMContainerTokenSecretManager containerTokenSecretManager;  
  17.   private final NMTokenSecretManagerInRM nmTokenSecretManager;  
  18.     
  19.   //心跳间隔  
  20.   private long nextHeartBeatInterval;  
  21.   //远程RPC服务  
  22.   private Server server;  
  23.   private InetSocketAddress resourceTrackerAddress;  
  24.   
  25.   private static final NodeHeartbeatResponse resync = recordFactory  
  26.       .newRecordInstance(NodeHeartbeatResponse.class);  
  27.   private static final NodeHeartbeatResponse shutDown = recordFactory  
  28.   .newRecordInstance(NodeHeartbeatResponse.class);  
  29.     
  30.   //最小分配的内存的大小  
  31.   private int minAllocMb;  
  32.   //最小分配的核数大小  
  33.   private int minAllocVcores;  

也是继承了抽象服务类，这里面包含的内容就多了许多，重点关注，节点列表管理器对象NodesListManager和NMLivelinessMonitor，这2者与本文叙述所相关。然后跳到节点注册请求处理操作。在节点注册请求进来的时候，首先会做一些请求的过滤条件的验证，过滤不符合要求的节点。

[java] 
view plain
copy
print
?

  1. //响应NodeManager的节点注册请求方法  
  2.   @SuppressWarnings(“unchecked”)  
  3.   @Override  
  4.   public RegisterNodeManagerResponse registerNodeManager(  
  5.       RegisterNodeManagerRequest request) throws YarnException,  
  6.       IOException {  
  7.   
  8.     NodeId nodeId = request.getNodeId();  
  9.     String host = nodeId.getHost();  
  10.     …..  
  11.   
  12.     // Check if this node is a ‘valid’ node  
  13.     //如果此节点是在exclude名单中,注册请求将会被拒绝,调用的是节点列表管理器的isValidNode方法  
  14.     if (!this.nodesListManager.isValidNode(host)) {  
  15.       String message =  
  16.           “Disallowed NodeManager from  “ + host  
  17.               + “, Sending SHUTDOWN signal to the NodeManager.”;  
  18.       LOG.info(message);  
  19.       response.setDiagnosticsMessage(message);  
  20.       response.setNodeAction(NodeAction.SHUTDOWN);  
  21.       return response;  
  22.     }  
  23.   
  24.     // Check if this node has minimum allocations  
  25.     //判断节点资源是否满足最小内存和核数的限制,如果没有同样拒绝注册  
  26.     if (capability.getMemory() < minAllocMb  
  27.         || capability.getVirtualCores() < minAllocVcores) {  
  28.       String message =  
  29.           “NodeManager from  “ + host  
  30.               + ” doesn’t satisfy minimum allocations, Sending SHUTDOWN”  
  31.               + ” signal to the NodeManager.”;  
  32.       LOG.info(message);  
  33.       response.setDiagnosticsMessage(message);  
  34.       response.setNodeAction(NodeAction.SHUTDOWN);  
  35.       return response;  
  36.     }  
  37.     …..  

2个条件，节点是否有效，有效的规则就是在NodeListManager中定义的，这个后面会提到，第二个是节点所剩资源是否足够启动NodeManager。如果这2个请求都过了的话，则表明可以进行注册，此节点将会被注册进行存活监控线程中。

[java] 
view plain
copy
print
?

  1. …..  
  2.     // On every node manager register we will be clearing NMToken keys if  
  3.     // present for any running application.  
  4.     this.nmTokenSecretManager.removeNodeKey(nodeId);  
  5.     //同时将节点注册到节点存活监控线程中  
  6.     this.nmLivelinessMonitor.register(nodeId);  
  7.   
  8.     String message =  
  9.         “NodeManager from node “ + host + “(cmPort: “ + cmPort + ” httpPort: “  
  10.             + httpPort + “) “ + “registered with capability: “ + capability  
  11.             + “, assigned nodeId “ + nodeId;  
  12.     LOG.info(message);  
  13.     response.setNodeAction(NodeAction.NORMAL);  
  14.     response.setRMIdentifier(ResourceManager.clusterTimeStamp);  
  15.     return response;  
  16.   }  

注册操作在上篇文章都详细讲述过了，在基础监控类中声明了，如下：

[java] 
view plain
copy
print
?

  1. //进程存活状态监控类  
  2. public abstract class AbstractLivelinessMonitor<O> extends AbstractService {  
  3.   ……  
  4.   
  5.   private final Clock clock;  
  6.     
  7.   //保存了心跳检验的结果记录  
  8.   private Map<O, Long> running = new HashMap<O, Long>();  
  9.     
  10.   //更新心跳监控检测最新时间  
  11.   public synchronized void receivedPing(O ob) {  
  12.     //only put for the registered objects  
  13.     if (running.containsKey(ob)) {  
  14.       running.put(ob, clock.getTime());  
  15.     }  
  16.   }  

具体细节请求点击
YARN源码分析(一)
。

OK，回到之前没有说清楚的NodeListManager节点列表管理器类，这个类提供了节点有效性检查的方法

[java] 
view plain
copy
print
?

  1. …..  
  2.   
  3.     // Check if this node is a ‘valid’ node  
  4.     //如果此节点是在exclude名单中,注册请求将会被拒绝,调用的是节点列表管理器的isValidNode方法  
  5.     if (!this.nodesListManager.isValidNode(host)) {  
  6.       String message =  
  7.           “Disallowed NodeManager from  “ + host  
  8.               + “, Sending SHUTDOWN signal to the NodeManager.”;  
  9.       LOG.info(message);  
  10.       response.setDiagnosticsMessage(message);  
  11.       response.setNodeAction(NodeAction.SHUTDOWN);  
  12.       return response;  
  13.     }  
  14.     …..  

传入的是主机名，可以联系之前Decommision文章中提到的include，exclude名单列表的内容。

[java] 
view plain
copy
print
?

  1. //节点列表管理器,主要是根据include白名单和exclude黑名单属性进行判断,也是一个服务  
  2. public class NodesListManager extends AbstractService implements  
  3.     EventHandler<NodesListManagerEvent> {  
  4.   
  5.   private static final Log LOG = LogFactory.getLog(NodesListManager.class);  
  6.   //节点列表读取器  
  7.   private HostsFileReader hostsReader;  
  8.   private Configuration conf;  
  9.   //不允许使用的节点列表  
  10.   private Set<RMNode> unusableRMNodesConcurrentSet = Collections  
  11.       .newSetFromMap(new ConcurrentHashMap<RMNode,Boolean>());  
  12.   //资源管理上下文  
  13.   private final RMContext rmContext;  
  14.   …..  

在这个类中写明了unsableNodes无法使用的节点列表名单，但是有效性检查的方法并没有使用到此变量。下面是真正的valid检测方法

[java] 
view plain
copy
print
?

  1. //输入主机名,判断是否是有效的节点,  
  2.   public boolean isValidNode(String hostName) {  
  3.     synchronized (hostsReader) {  
  4.       //获取可接入和不可接入主机名列表  
  5.       Set<String> hostsList = hostsReader.getHosts();  
  6.       Set<String> excludeList = hostsReader.getExcludedHosts();  
  7.       String ip = NetUtils.normalizeHostName(hostName);  
  8.       //判断是否在相应的列表中以此判断节点是否有效  
  9.       return (hostsList.isEmpty() || hostsList.contains(hostName) || hostsList  
  10.           .contains(ip))  
  11.           && !(excludeList.contains(hostName) || excludeList.contains(ip));  
  12.     }  
  13.   }  

也是通过hostReader对象读取配置文件中的include，exclude主机名列表做判断。初始主机从这里读取出来

[java] 
view plain
copy
print
?

  1. @Override  
  2.   protected void serviceInit(Configuration conf) throws Exception {  
  3.   
  4.     this.conf = conf;  
  5.   
  6.     // Read the hosts/exclude files to restrict access to the RM  
  7.     //在服务初始化的时候读取include和exclude文件信息,exclude的节点列表名单将会被RM拒绝接入  
  8.     try {  
  9.       this.hostsReader =   
  10.         new HostsFileReader(  
  11.             conf.get(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH,   
  12.                 YarnConfiguration.DEFAULT_RM_NODES_INCLUDE_FILE_PATH),  
  13.             conf.get(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH,   
  14.                 YarnConfiguration.DEFAULT_RM_NODES_EXCLUDE_FILE_PATH)  
  15.                 );  
  16.       //输出节点信息  
  17.       printConfiguredHosts();  
  18.       …..  

OK，节点注册操作分析完毕。

节点HeartBeat心跳

心跳方法在NodeManager中有直接定义

[java] 
view plain
copy
print
?

  1. public class NodeManager implements ContainerManagementProtocol {  
  2.    ….  
  3.   //周期心跳方法  
  4.   public void heartbeat() throws IOException, YarnException {  
  5.     NodeStatus nodeStatus =   
  6.       org.apache.hadoop.yarn.server.resourcemanager.NodeManager.createNodeStatus(  
  7.           nodeId, getContainerStatuses(containers));  
  8.     nodeStatus.setResponseId(responseID);  
  9.     NodeHeartbeatRequest request = recordFactory  
  10.         .newRecordInstance(NodeHeartbeatRequest.class);  
  11.     request.setNodeStatus(nodeStatus);  
  12.     //调用resourceTrackerService发送心跳包，并获取响应回复  
  13.     NodeHeartbeatResponse response = resourceTrackerService  
  14.         .nodeHeartbeat(request);  
  15.     responseID = response.getResponseId();  
  16.   }  

也是远程调用ResourceTrackerService方法

[java] 
view plain
copy
print
?

  1. //节点心跳相应方法  
  2.   @SuppressWarnings(“unchecked”)  
  3.   @Override  
  4.   public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)  
  5.       throws YarnException, IOException {  
  6.       
  7.     //从心跳中获取远程节点状态信息  
  8.     NodeStatus remoteNodeStatus = request.getNodeStatus();  
  9.     /** 
  10.      * Here is the node heartbeat sequence… 
  11.      * 1. Check if it’s a registered node 
  12.      * 2. Check if it’s a valid (i.e. not excluded) node  
  13.      * 3. Check if it’s a ‘fresh’ heartbeat i.e. not duplicate heartbeat  
  14.      * 4. Send healthStatus to RMNode 
  15.      */  
  16.   
  17.     NodeId nodeId = remoteNodeStatus.getNodeId();  
  18.   
  19.     // 1. Check if it’s a registered node  
  20.     RMNode rmNode = this.rmContext.getRMNodes().get(nodeId);  
  21.     if (rmNode == null) {  
  22.       /* node does not exist */  
  23.       String message = “Node not found resyncing “ + remoteNodeStatus.getNodeId();  
  24.       LOG.info(message);  
  25.       resync.setDiagnosticsMessage(message);  
  26.       return resync;  
  27.     }  
  28.   
  29.     // Send ping  
  30.     //更新心跳响应最新时间  
  31.     this.nmLivelinessMonitor.receivedPing(nodeId);  
  32.   
  33.     // 2. Check if it’s a valid (i.e. not excluded) node  
  34.     //每次心跳检测都会检查节点是否被拉入exclude名单  
  35.     if (!this.nodesListManager.isValidNode(rmNode.getHostName())) {  
  36.       String message =  
  37.           “Disallowed NodeManager nodeId: “ + nodeId + ” hostname: “  
  38.               + rmNode.getNodeAddress();  
  39.       LOG.info(message);  
  40.       shutDown.setDiagnosticsMessage(message);  
  41.       //如果是被拉入,则触发节点撤销事件  
  42.       this.rmContext.getDispatcher().getEventHandler().handle(  
  43.           new RMNodeEvent(nodeId, RMNodeEventType.DECOMMISSION));  
  44.       return shutDown;  
  45.     }  
  46.       
  47.     …..  
  48.   
  49.     // Heartbeat response  
  50.     //设置心跳回复  
  51.     NodeHeartbeatResponse nodeHeartBeatResponse = YarnServerBuilderUtils  
  52.         .newNodeHeartbeatResponse(lastNodeHeartbeatResponse.  
  53.             getResponseId() + 1, NodeAction.NORMAL, null, null, null, null,  
  54.             nextHeartBeatInterval);  
  55.     rmNode.updateNodeHeartbeatResponseForCleanup(nodeHeartBeatResponse);  

心跳方法也不是特别的复杂。

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/9188.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论