Hadoop3.2.1 【 YARN 】源码分析 : ResourceManager # NodeManager 浅析

短命女 2022-12-28 01:43 168阅读 0赞

# 一. 前言 #

NodeManager管理部分主要由三个服务构成， 分别是NMLivelinessMonitor、 NodesListManager和ResourceTrackerService， 它们共同管理NodeManager的生存周期 .  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

# 二. NMLivelinessMonitor #

该服务周期性遍历集群中所有NodeManager， 如果一个NodeManager在一定时间（可通过参数yarn.nm.liveness-monitor.expiry-interval-ms配置， 默认为10min） 内未汇报心跳信息， 则认为它死掉了， 它上面所有正在运行的Container将被置为运行失败。 需要注意的  
是， RM不会重新执行这些Container， 它只会通过心跳机制告诉对应的AM， 由AM决定是否重新执行。 如果需要， 则AM重新向RM申请资源， 然后由AM与对应的NodeManager通信以重新运行失败的Container .

## 2.1. AbstractLivelinessMonitor ##

一个简单的活跃度监视器，客户端可以用它注册，信任组件来监视活跃度，在过期时得到一个回调，然后最终取消注册。  
AMLivelinessMonitor 的父类是AbstractLivelinessMonitor, AbstractLivelinessMonitor是一个抽象类,它继承于AbstractService .

### 2.1.1. 属性 ###

在属性里面,需要重点关注的几个参数

1.  Thread checkerThread : 定时检测心跳的线程
2.  DEFAULT\_EXPIRE : 默认的过期时间为5min
3.  monitorInterval : 监控间隔, 为过期时间的 三分之一
4.  running : 监控程序的清单

// 定时检测心跳的线程
      private Thread checkerThread;
    
      // 标识是否停止.
      private volatile boolean stopped;
    
      // 默认过期时间 5min [ AMLivelinessMonitor 的过期时间为 10 min] 
      public static final int DEFAULT_EXPIRE = 5*60*1000;//5 mins
    
      // 过期时间
      private long expireInterval = DEFAULT_EXPIRE;
    
      // 监控间隔, 过期时间的 / 3
      private long monitorInterval = expireInterval / 3;
      
      // 在启动时 是否 重置Timer
      private volatile boolean resetTimerOnStart = true;
      
      // 时钟对象
      private final Clock clock;
    
      // 正在监控程序的清单   对象 -> 上一次心跳时间
      private Map<O, Long> running = new HashMap<O, Long>();

### 2.1.2. serviceStart 启动服务 ###

这里看到 使用PingChecker对象,构建了一个线程,用于监控任务.

@Override
      protected void serviceStart() throws Exception { 
        assert !stopped : "starting when already stopped";
        resetTimer();
        checkerThread = new Thread(new PingChecker());
        checkerThread.setName("Ping Checker");
        checkerThread.start();
        super.serviceStart();
      }

### 2.1.3. receivedPing ###

注册监控方法

public synchronized void receivedPing(O ob) {
        //only put for the registered objects
        if (running.containsKey(ob)) {
          running.put(ob, clock.getTime());
        }
      }

### 2.1.4. PingChecker 内部类 ###

PingChecker是AbstractLivelinessMonitor 的内部类.  
实现了Runnable方法, 主要是迭代running中记录的数据, 验证是否过期. 如果过期,触发回调函数expire.

@Override
        public void run() { 
          while (!stopped && !Thread.currentThread().isInterrupted()) { 
            synchronized (AbstractLivelinessMonitor.this) { 
    
              // 获取 running 中要监控对象的结合
              Iterator<Map.Entry<O, Long>> iterator = running.entrySet().iterator();
    
              // avoid calculating current time everytime in loop
              // 获取当前时间
              long currentTime = clock.getTime();
    
              while (iterator.hasNext()) { 
                // 获取监控对象
                Map.Entry<O, Long> entry = iterator.next();
                O key = entry.getKey();
                // 获取过期间隔
                long interval = getExpireInterval(key);
                // 判断是否过期,  当前时间 > 心跳时间+过期时间 ?
                if (currentTime > entry.getValue() + interval) { 
                  iterator.remove();
                  // 触发回调函数
                  expire(key);
                  LOG.info("Expired:" + entry.getKey().toString()
                      + " Timed out after " + interval / 1000 + " secs");
                }
              }
            }
            try { 
              Thread.sleep(monitorInterval);
            } catch (InterruptedException e) { 
              LOG.info(getName() + " thread interrupted");
              break;
            }
          }
        }

## 2.2.构造方法 ##

NMLivelinessMonitor 需要两个属性 Dispatcher d 和 Clock clock .

public NMLivelinessMonitor(Dispatcher d) { 
        super("NMLivelinessMonitor");
        this.dispatcher = d.getEventHandler();
      }

## 2.3. 初始化方法 ##

初始化是交由serviceInit方法完成的, 主要是设置默认过期时间10min和监控间隔, 10min的三分之一, 就是说大概三分30秒.监控一次.

public void serviceInit(Configuration conf) throws Exception { 
        // yarn.nm.liveness-monitor.expiry-interval-ms : 600000
        int expireIntvl = conf.getInt(YarnConfiguration.RM_NM_EXPIRY_INTERVAL_MS,
                YarnConfiguration.DEFAULT_RM_NM_EXPIRY_INTERVAL_MS);
        
        // 设置过期时效为10min
        setExpireInterval(expireIntvl);
        
        // 监控间隔 expireIntvl/3  = 3min 30s
        setMonitorInterval(expireIntvl/3);
        super.serviceInit(conf);
      }

## 2.4. 超时回调 ##

超时回调函数为expire . 入参为ApplicationAttemptId  
使用dispatcher进行处理.

@Override
      protected void expire(NodeId id) { 
        dispatcher.handle(
            new RMNodeEvent(id, RMNodeEventType.EXPIRE)); 
      }

# 三. NodesListManager #

NodesListManager管理exlude（类似于黑名单） 和inlude（类似于白名单） 节点列表，这两个列表所在的文件分别可通过yarn.resourcemanager.nodes.include-path和yarn.resourcemanager.nodes.exclude-path配置（每个节点的host或者IP占一行） ， 其中，exlude节点列表可认为是黑名单， 它们不允许直接与RM通信（直接在RPC层抛出异常，导致NM退出） ， 而inlude节点列表可认为是白名单（通常不将这两个名单直接称为黑名单和白名单， 这两个概念被用到了YARN其他地方） 。 默认情况下， 这两个列表均为空，表示任何节点均被允许接入RM。 需要注意的是， 管理员可通过命令"bin/yarn rmadmin -refreshNodes"动态加载这两个文件。

## 3.1. serviceInit 初始化 ##

serviceInit构建了一个Timer ,定时监控黑白名单的变化 \[超时时间为10分钟, 默认监控间隔5分钟, \],根据黑白名单更新NODE信息.

\*\*设置参数时最大监控为10分钟 : 计算公式–> (Math.min(超时时间/2,600000)); \*\*

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
    
        this.conf = conf;
        //  节点IP缓存的过期间隔。-1  禁用
        //  yarn.resourcemanager.node-ip-cache.expiry-interval-secs :  -1
        int nodeIpCacheTimeout = conf.getInt(
            YarnConfiguration.RM_NODE_IP_CACHE_EXPIRY_INTERVAL_SECS,
            YarnConfiguration.DEFAULT_RM_NODE_IP_CACHE_EXPIRY_INTERVAL_SECS);
        if (nodeIpCacheTimeout <= 0) { 
          resolver = new DirectResolver();
        } else { 
          resolver =
              new CachedResolver(SystemClock.getInstance(), nodeIpCacheTimeout);
          addIfService(resolver);
        }
    
        // Read the hosts/exclude files to restrict access to the RM
        try { 
          // 白名单   yarn.resourcemanager.nodes.include-path: ""
          this.includesFile = conf.get(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH,
              YarnConfiguration.DEFAULT_RM_NODES_INCLUDE_FILE_PATH);
    
          // 黑名单 yarn.resourcemanager.nodes.exclude-path: ""
          this.excludesFile = conf.get(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH,
              YarnConfiguration.DEFAULT_RM_NODES_EXCLUDE_FILE_PATH);
    
          // 构建HostsFileReader 读取 黑/白名单
          this.hostsReader =
              createHostsFileReader(this.includesFile, this.excludesFile);
    
          setDecomissionedNMs();
    
          printConfiguredHosts();
        } catch (YarnException ex) { 
          disableHostsFileReader(ex);
        } catch (IOException ioe) { 
          disableHostsFileReader(ioe);
        }
    
        // yarn.resourcemanager.node-removal-untracked.timeout-ms : 60000 =  10min
        final int nodeRemovalTimeout =
            conf.getInt(
                YarnConfiguration.RM_NODEMANAGER_UNTRACKED_REMOVAL_TIMEOUT_MSEC,
                YarnConfiguration.
                    DEFAULT_RM_NODEMANAGER_UNTRACKED_REMOVAL_TIMEOUT_MSEC);
    
        // 处理间隔   5min , 最大10min
        nodeRemovalCheckInterval = (Math.min(nodeRemovalTimeout/2,
            600000));
    
        // 构建节点监控计时器
        removalTimer = new Timer("Node Removal Timer");
    
        // 启动服务  每10分钟一次, 间隔5min
        removalTimer.schedule(new TimerTask() { 
          @Override
          public void run() { 
            // 获取当前时间戳
            long now = Time.monotonicNow();
    
            // 遍历所有的node
            for (Map.Entry<NodeId, RMNode> entry : rmContext.getInactiveRMNodes().entrySet()) { 
    
              NodeId nodeId = entry.getKey();
              RMNode rmNode = entry.getValue();
              //是否处于黑白名单之中
              if (isUntrackedNode(rmNode.getHostName())) { 
                //未在黑白名单之中
                if (rmNode.getUntrackedTimeStamp() == 0) { 
                  // 设置 监控 时间戳
                  rmNode.setUntrackedTimeStamp(now);
                } else
    
                  if (now - rmNode.getUntrackedTimeStamp() >  nodeRemovalTimeout) { 
                    // 超时, 移除node
                    RMNode result = rmContext.getInactiveRMNodes().remove(nodeId);
                    if (result != null) { 
                      // 更新集群缓存信息
                      decrInactiveNMMetrics(rmNode);
                      LOG.info("Removed " +result.getState().toString() + " node "
                          + result.getHostName() + " from inactive nodes list");
                    }
                  }
              } else { 
                // 设置未监控时间戳为0
                rmNode.setUntrackedTimeStamp(0);
              }
            }
          }
        }, nodeRemovalCheckInterval, nodeRemovalCheckInterval);
    
        super.serviceInit(conf);
      }

## 3.2. refreshNodes 刷新节点信息 ##

// Handle excluded nodes based on following rules:
      // Recommission DECOMMISSIONED or DECOMMISSIONING nodes no longer excluded;
      // Gracefully decommission excluded nodes that are not already
      // DECOMMISSIONED nor DECOMMISSIONING; Take no action for excluded nodes
      // that are already DECOMMISSIONED or DECOMMISSIONING.
      private void handleExcludeNodeList(boolean graceful, int timeout) { 
        // DECOMMISSIONED/DECOMMISSIONING nodes need to be re-commissioned.
        List<RMNode> nodesToRecom = new ArrayList<RMNode>();
    
        // Nodes need to be decommissioned (graceful or forceful);
        List<RMNode> nodesToDecom = new ArrayList<RMNode>();
    
        HostDetails hostDetails = hostsReader.getHostDetails();
        Set<String> includes = hostDetails.getIncludedHosts();
        Map<String, Integer> excludes = hostDetails.getExcludedMap();
    
        for (RMNode n : this.rmContext.getRMNodes().values()) { 
          NodeState s = n.getState();
          // An invalid node (either due to explicit exclude or not include)
          // should be excluded.
          boolean isExcluded = !isValidNode(
              n.getHostName(), includes, excludes.keySet());
          String nodeStr = "node " + n.getNodeID() + " with state " + s;
          if (!isExcluded) { 
            // Note that no action is needed for DECOMMISSIONED node.
            if (s == NodeState.DECOMMISSIONING) { 
              LOG.info("Recommission " + nodeStr);
              nodesToRecom.add(n);
            }
            // Otherwise no-action needed.
          } else { 
            // exclude is true.
            if (graceful) { 
              // Use per node timeout if exist otherwise the request timeout.
              Integer timeoutToUse = (excludes.get(n.getHostName()) != null)?
                  excludes.get(n.getHostName()) : timeout;
              if (s != NodeState.DECOMMISSIONED &&
                  s != NodeState.DECOMMISSIONING) { 
                LOG.info("Gracefully decommission " + nodeStr);
                nodesToDecom.add(n);
              } else if (s == NodeState.DECOMMISSIONING &&
                         !Objects.equals(n.getDecommissioningTimeout(),
                             timeoutToUse)) { 
                LOG.info("Update " + nodeStr + " timeout to be " + timeoutToUse);
                nodesToDecom.add(n);
              } else { 
                LOG.info("No action for " + nodeStr);
              }
            } else { 
              if (s != NodeState.DECOMMISSIONED) { 
                LOG.info("Forcefully decommission " + nodeStr);
                nodesToDecom.add(n);
              }
            }
          }
        }
    
        for (RMNode n : nodesToRecom) { 
          RMNodeEvent e = new RMNodeEvent(
              n.getNodeID(), RMNodeEventType.RECOMMISSION);
          this.rmContext.getDispatcher().getEventHandler().handle(e);
        }
    
        for (RMNode n : nodesToDecom) { 
          RMNodeEvent e;
          if (graceful) { 
            Integer timeoutToUse = (excludes.get(n.getHostName()) != null)?
                excludes.get(n.getHostName()) : timeout;
            e = new RMNodeDecommissioningEvent(n.getNodeID(), timeoutToUse);
          } else { 
            RMNodeEventType eventType = isUntrackedNode(n.getHostName())?
                RMNodeEventType.SHUTDOWN : RMNodeEventType.DECOMMISSION;
            e = new RMNodeEvent(n.getNodeID(), eventType);
          }
          this.rmContext.getDispatcher().getEventHandler().handle(e);
        }
    
        updateInactiveNodes();
      }

# 四. ResourceTrackerService #

ResourceTrackerService实现了RPC协议ResourceTracker， 负责处理来自各个NodeManager的请求， 请求主要包括注册和心跳两种，  
**注册**是NodeManager启动时发生的行为， 请求包中包含节点ID， 可用的资源上限等信息；  
**心跳**是周期性行为， 包含各个Container运行状态， 运行的Application列表、 节点健康状况 ，  
而ResourceTrackerService则为NM返回待释放的Container列表、Application列表等。

## 4.1.registerNodeManager 注册NodeManager ##

NM启动时， 它所做的第一件事是向RM注册， 这是通过RPC函数ResourceTracker\#registerNodeManager实现的， 注册信息包括节点可用资源总量、 对外开放的HTTP端口号等。

@SuppressWarnings("unchecked")
      @Override
      public RegisterNodeManagerResponse registerNodeManager(
          RegisterNodeManagerRequest request) throws YarnException,
          IOException { 
    
        //获取NodeId
        NodeId nodeId = request.getNodeId();
        // 获取host
        String host = nodeId.getHost();
        // 获取cm端口
        int cmPort = nodeId.getPort();
        // 获取http端口
        int httpPort = request.getHttpPort();
        // 获取资源容量
        Resource capability = request.getResource();
    
        // 获取NM版本
        String nodeManagerVersion = request.getNMVersion();
        // 获取物理资源
        Resource physicalResource = request.getPhysicalResource();
    
    
        RegisterNodeManagerResponse response = recordFactory
            .newRecordInstance(RegisterNodeManagerResponse.class);
    
        // 更新NM的版本信息
        if (!minimumNodeManagerVersion.equals("NONE")) { 
          if (minimumNodeManagerVersion.equals("EqualToRM")) { 
            minimumNodeManagerVersion = YarnVersionInfo.getVersion();
          }
    
          if ((nodeManagerVersion == null) || (VersionUtil.compareVersions(nodeManagerVersion,minimumNodeManagerVersion)) < 0) { 
            String message =
                "Disallowed NodeManager Version " + nodeManagerVersion
                    + ", is less than the minimum version "
                    + minimumNodeManagerVersion + " sending SHUTDOWN signal to "
                    + "NodeManager.";
            LOG.info(message);
            response.setDiagnosticsMessage(message);
            response.setNodeAction(NodeAction.SHUTDOWN);
            return response;
          }
        }
        // 验证该node是有效
        // Check if this node is a 'valid' node
        if (!this.nodesListManager.isValidNode(host) &&
            !isNodeInDecommissioning(nodeId)) { 
          String message =
              "Disallowed NodeManager from " + host
                  + ", Sending SHUTDOWN signal to the NodeManager.";
          LOG.info(message);
          response.setDiagnosticsMessage(message);
          response.setNodeAction(NodeAction.SHUTDOWN);
          return response;
        }
        // 检查节点的容量是否从  dynamic-resources.xml 加载
        // check if node's capacity is load from dynamic-resources.xml String nid = nodeId.toString(); // 获取哦动态 资源容量 Resource dynamicLoadCapability = loadNodeResourceFromDRConfiguration(nid); if (dynamicLoadCapability != null) { if (LOG.isDebugEnabled()) { LOG.debug("Resource for node: " + nid + " is adjusted from: " + capability + " to: " + dynamicLoadCapability + " due to settings in dynamic-resources.xml."); } capability = dynamicLoadCapability; // sync back with new resource. response.setResource(capability); } // 检测该node是否有最小资源限制 // Check if this node has minimum allocations if (capability.getMemorySize() < minAllocMb || capability.getVirtualCores() < minAllocVcores) { String message = "NodeManager from " + host + " doesn't satisfy minimum allocations, Sending SHUTDOWN" + " signal to the NodeManager. Node capabilities are " + capability + "; minimums are " + minAllocMb + "mb and " + minAllocVcores + " vcores"; LOG.info(message); response.setDiagnosticsMessage(message); response.setNodeAction(NodeAction.SHUTDOWN); return response; } response.setContainerTokenMasterKey(containerTokenSecretManager .getCurrentKey()); response.setNMTokenMasterKey(nmTokenSecretManager .getCurrentKey()); // 构建RMNode对象 RMNode rmNode = new RMNodeImpl(nodeId, rmContext, host, cmPort, httpPort, resolve(host), capability, nodeManagerVersion, physicalResource); RMNode oldNode = this.rmContext.getRMNodes().putIfAbsent(nodeId, rmNode); if (oldNode == null) { // 新构建 RMnode RMNodeStartedEvent startEvent = new RMNodeStartedEvent(nodeId, request.getNMContainerStatuses(), request.getRunningApplications()); if (request.getLogAggregationReportsForApps() != null && !request.getLogAggregationReportsForApps().isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("Found the number of previous cached log aggregation " + "status from nodemanager:" + nodeId + " is :" + request.getLogAggregationReportsForApps().size()); } startEvent.setLogAggregationReportsForApps(request .getLogAggregationReportsForApps()); } this.rmContext.getDispatcher().getEventHandler().handle( startEvent); } else { // 重新连接 LOG.info("Reconnect from the node at: " + host); this.nmLivelinessMonitor.unregister(nodeId); if (CollectionUtils.isEmpty(request.getRunningApplications()) && rmNode.getState() != NodeState.DECOMMISSIONING && rmNode.getHttpPort() != oldNode.getHttpPort()) { // Reconnected node differs, so replace old node and start new node switch (rmNode.getState()) { case RUNNING: ClusterMetrics.getMetrics().decrNumActiveNodes(); break; case UNHEALTHY: ClusterMetrics.getMetrics().decrNumUnhealthyNMs(); break; default: LOG.debug("Unexpected Rmnode state"); } this.rmContext.getDispatcher().getEventHandler() .handle(new NodeRemovedSchedulerEvent(rmNode)); this.rmContext.getRMNodes().put(nodeId, rmNode); this.rmContext.getDispatcher().getEventHandler() .handle(new RMNodeStartedEvent(nodeId, null, null)); } else { // Reset heartbeat ID since node just restarted. oldNode.resetLastNodeHeartBeatResponse(); this.rmContext.getDispatcher().getEventHandler() .handle(new RMNodeReconnectEvent(nodeId, rmNode, request.getRunningApplications(), request.getNMContainerStatuses())); } } // On every node manager register we will be clearing NMToken keys if // present for any running application. this.nmTokenSecretManager.removeNodeKey(nodeId); this.nmLivelinessMonitor.register(nodeId); // Handle received container status, this should be processed after new // RMNode inserted if (!rmContext.isWorkPreservingRecoveryEnabled()) { if (!request.getNMContainerStatuses().isEmpty()) { LOG.info("received container statuses on node manager register :" + request.getNMContainerStatuses()); for (NMContainerStatus status : request.getNMContainerStatuses()) { handleNMContainerStatus(status, nodeId); } } } // 更新node 标签 // Update node's labels to RM's NodeLabelManager. Set<String> nodeLabels = NodeLabelsUtils.convertToStringSet( request.getNodeLabels()); if (isDistributedNodeLabelsConf && nodeLabels != null) { try { updateNodeLabelsFromNMReport(nodeLabels, nodeId); response.setAreNodeLabelsAcceptedByRM(true); } catch (IOException ex) { // Ensure the exception is captured in the response response.setDiagnosticsMessage(ex.getMessage()); response.setAreNodeLabelsAcceptedByRM(false); } } else if (isDelegatedCentralizedNodeLabelsConf) { this.rmContext.getRMDelegatedNodeLabelsUpdater().updateNodeLabels(nodeId); } // Update node's attributes to RM's NodeAttributesManager. if (request.getNodeAttributes() != null) { try { // update node attributes if necessary then update heartbeat response updateNodeAttributesIfNecessary(nodeId, request.getNodeAttributes()); response.setAreNodeAttributesAcceptedByRM(true); } catch (IOException ex) { //ensure the error message is captured and sent across in response String errorMsg = response.getDiagnosticsMessage() == null ? ex.getMessage() : response.getDiagnosticsMessage() + "\n" + ex.getMessage(); response.setDiagnosticsMessage(errorMsg); response.setAreNodeAttributesAcceptedByRM(false); } } StringBuilder message = new StringBuilder(); message.append("NodeManager from node ").append(host).append("(cmPort: ") .append(cmPort).append(" httpPort: "); message.append(httpPort).append(") ") .append("registered with capability: ").append(capability); message.append(", assigned nodeId ").append(nodeId); if (response.getAreNodeLabelsAcceptedByRM()) { message.append(", node labels {  ").append( StringUtils.join(",", nodeLabels) + " } "); } if (response.getAreNodeAttributesAcceptedByRM()) { message.append(", node attributes {  ") .append(request.getNodeAttributes() + " } ");
        }
    
        LOG.info(message.toString());
        response.setNodeAction(NodeAction.NORMAL);
        response.setRMIdentifier(ResourceManager.getClusterTimeStamp());
        response.setRMVersion(YarnVersionInfo.getVersion());
        return response;
      }

## 4.2. nodeHeartbeat 心跳 ##

NM启动后， 它会周期性地通过RPC函数ResourceTracker\#nodeHeartbeat汇报心跳， 心跳信息包含各个Container运行状态、 运行的Application列表、 节点健康状况等信息， 而RM则为之返回需要释放的Container列表、 Application列表等。  
步骤清单:

1.  检查node是否有效, 检查是否退役中.
2.  检查node 是否被注册,发送心跳数据给nmLivelinessMonitor监控
3.  检查心跳是否是新的,而不是重复的.
4.  发送响应给RMNode. 保存最后一次请求
5.  更新node的标签信息
6.  检查节点的容量是否是从dynamic-resources.xml 加载, 如果是的话,发送更新资源信息
7.  发送Container的数量限制给node, 如果超出node中队列限制,则进行截取操作…
8.  获取node 属性, 并进行更新操作.

@SuppressWarnings("unchecked")
      @Override
      public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)
          throws YarnException, IOException { 
    
        NodeStatus remoteNodeStatus = request.getNodeStatus();
        /**
         * 这里是node心跳顺序
         * 1. 检查是否是有效node
         * 2. 检查是否被注册, 跟新心跳
         * 3. 检查是否是新的心跳,而不是重复的.
         * 4. 发送心跳状态给 RMNode
         * 5. 如果启用了分布式节点标签配置，则更新节点的标签
         *
         * Here is the node heartbeat sequence...
         * 1. Check if it's a valid (i.e. not excluded) node * 2. Check if it's a registered node
         * 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat * 4. Send healthStatus to RMNode * 5. Update node's labels if distributed Node Labels configuration is enabled
         */
        NodeId nodeId = remoteNodeStatus.getNodeId();
        // 检查node是否有效, 检查是否退役中.
        // 1. Check if it's a valid (i.e. not excluded) node, if not, see if it is // in decommissioning. if (!this.nodesListManager.isValidNode(nodeId.getHost()) && !isNodeInDecommissioning(nodeId)) { String message = "Disallowed NodeManager nodeId: " + nodeId + " hostname: " + nodeId.getHost(); LOG.info(message); return YarnServerBuilderUtils.newNodeHeartbeatResponse( NodeAction.SHUTDOWN, message); } // 检查node 是否被注册 // 2. 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);
          return YarnServerBuilderUtils.newNodeHeartbeatResponse(NodeAction.RESYNC,
              message);
        }
        // 发送心跳数据给监控
        // Send ping
        this.nmLivelinessMonitor.receivedPing(nodeId);
        this.decommissioningWatcher.update(rmNode, remoteNodeStatus);
        // 检查心跳是否是新的,而不是重复的.
        // 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat NodeHeartbeatResponse lastNodeHeartbeatResponse = rmNode.getLastNodeHeartBeatResponse(); if (getNextResponseId( remoteNodeStatus.getResponseId()) == lastNodeHeartbeatResponse .getResponseId()) { LOG.info("Received duplicate heartbeat from node " + rmNode.getNodeAddress()+ " responseId=" + remoteNodeStatus.getResponseId()); return lastNodeHeartbeatResponse; } else if (remoteNodeStatus.getResponseId() != lastNodeHeartbeatResponse .getResponseId()) { String message = "Too far behind rm response id:" + lastNodeHeartbeatResponse.getResponseId() + " nm response id:" + remoteNodeStatus.getResponseId(); LOG.info(message); // 直接给RMnode 发送重启命令 // TODO: Just sending reboot is not enough. Think more. this.rmContext.getDispatcher().getEventHandler().handle( new RMNodeEvent(nodeId, RMNodeEventType.REBOOTING)); return YarnServerBuilderUtils.newNodeHeartbeatResponse(NodeAction.RESYNC, message); } // 检查是否是退役中或者已经退役. // Evaluate whether a DECOMMISSIONING node is ready to be DECOMMISSIONED. if (rmNode.getState() == NodeState.DECOMMISSIONING && decommissioningWatcher.checkReadyToBeDecommissioned( rmNode.getNodeID())) { String message = "DECOMMISSIONING " + nodeId + " is ready to be decommissioned"; LOG.info(message); this.rmContext.getDispatcher().getEventHandler().handle( new RMNodeEvent(nodeId, RMNodeEventType.DECOMMISSION)); this.nmLivelinessMonitor.unregister(nodeId); return YarnServerBuilderUtils.newNodeHeartbeatResponse( NodeAction.SHUTDOWN, message); } if (timelineServiceV2Enabled) { // Check & update collectors info from request. updateAppCollectorsMap(request); } // 构建心跳响应... // Heartbeat response NodeHeartbeatResponse nodeHeartBeatResponse = YarnServerBuilderUtils.newNodeHeartbeatResponse( getNextResponseId(lastNodeHeartbeatResponse.getResponseId()), NodeAction.NORMAL, null, null, null, null, nextHeartBeatInterval); rmNode.setAndUpdateNodeHeartbeatResponse(nodeHeartBeatResponse); populateKeys(request, nodeHeartBeatResponse); ConcurrentMap<ApplicationId, ByteBuffer> systemCredentials = rmContext.getSystemCredentialsForApps(); if (!systemCredentials.isEmpty()) { nodeHeartBeatResponse.setSystemCredentialsForApps(systemCredentials); } if (timelineServiceV2Enabled) { // Return collectors' map that NM needs to know
          setAppCollectorsMapToResponse(rmNode.getRunningApps(),
              nodeHeartBeatResponse);
        }
        // 4. 发送响应给RMNode. 保存最后一次请求
        // 4. Send status to RMNode, saving the latest response.
        RMNodeStatusEvent nodeStatusEvent =
            new RMNodeStatusEvent(nodeId, remoteNodeStatus);
        if (request.getLogAggregationReportsForApps() != null
            && !request.getLogAggregationReportsForApps().isEmpty()) { 
          nodeStatusEvent.setLogAggregationReportsForApps(request
            .getLogAggregationReportsForApps());
        }
        this.rmContext.getDispatcher().getEventHandler().handle(nodeStatusEvent);
        // 5. 更新node的标签信息
        // 5. Update node's labels to RM's NodeLabelManager.
        if (isDistributedNodeLabelsConf && request.getNodeLabels() != null) { 
          try { 
            updateNodeLabelsFromNMReport(
                NodeLabelsUtils.convertToStringSet(request.getNodeLabels()),
                nodeId);
            nodeHeartBeatResponse.setAreNodeLabelsAcceptedByRM(true);
          } catch (IOException ex) { 
            //ensure the error message is captured and sent across in response
            nodeHeartBeatResponse.setDiagnosticsMessage(ex.getMessage());
            nodeHeartBeatResponse.setAreNodeLabelsAcceptedByRM(false);
          }
        }
        // 6. 检查节点的容量是否是从dynamic-resources.xml 加载, 如果是的话,发送更新资源信息
        // 6. check if node's capacity is load from dynamic-resources.xml // if so, send updated resource back to NM. String nid = nodeId.toString(); Resource capability = loadNodeResourceFromDRConfiguration(nid); // sync back with new resource if not null. if (capability != null) { nodeHeartBeatResponse.setResource(capability); } // 7. 发送Container的数量限制给node, 如果超出node中队列限制,则进行截取操作.. // 7. Send Container Queuing Limits back to the Node. This will be used by // the node to truncate the number of Containers queued for execution. if (this.rmContext.getNodeManagerQueueLimitCalculator() != null) { nodeHeartBeatResponse.setContainerQueuingLimit( this.rmContext.getNodeManagerQueueLimitCalculator() .createContainerQueuingLimit()); } // 8. 获取node 属性, 并进行更新操作. // 8. Get node's attributes and update node-to-attributes mapping
        // in RMNodeAttributeManager.
        if (request.getNodeAttributes() != null) { 
          try { 
            // update node attributes if necessary then update heartbeat response
            updateNodeAttributesIfNecessary(nodeId, request.getNodeAttributes());
            nodeHeartBeatResponse.setAreNodeAttributesAcceptedByRM(true);
          } catch (IOException ex) { 
            //ensure the error message is captured and sent across in response
            String errorMsg =
                nodeHeartBeatResponse.getDiagnosticsMessage() == null ?
                    ex.getMessage() :
                    nodeHeartBeatResponse.getDiagnosticsMessage() + "\n" + ex
                        .getMessage();
            nodeHeartBeatResponse.setDiagnosticsMessage(errorMsg);
            nodeHeartBeatResponse.setAreNodeAttributesAcceptedByRM(false);
          }
        }
    
        return nodeHeartBeatResponse;
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221120/7afa4ce97bb34c329224440a6a80420b.png