Hadoop3.2.1 【 YARN 】源码分析 : NodeStatusUpdaterImpl概述

た入场券 2023-01-07 03:53 166阅读 0赞

# 一. 前言 #

NodeStatusUpdater是NodeManager与ResourceManager通信的唯一通道。 当NodeManager启动时， 该组件负责向ResourceManager注册， 并汇报节点上总的可用资源（ 该值在运行过程中不再汇报） ； 之后， 该组件周期性与ResourceManager通信， 汇报各个Container的状态更新， 包括节点上正运行的Container、 已完成的Container等信息， 同时ResourceManager会为之返回待清理Container列表、 待清理应用程序列表、 诊断信息、 各种Token等信息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

# 二. 构造方法 #

NodeStatusUpdaterImpl 是在NodeManager\#serviceInit方法中进行构建

// NodeManager#serviceInit
        // node 状态更细 : org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdaterImpl
        nodeStatusUpdater = createNodeStatusUpdater(context, dispatcher, nodeHealthChecker);

// NodeManager#createNodeStatusUpdater
      protected NodeStatusUpdater createNodeStatusUpdater(Context context,  Dispatcher dispatcher, NodeHealthCheckerService healthChecker) { 
        return new NodeStatusUpdaterImpl(context, dispatcher, healthChecker,metrics);
      }

*  构造方法:  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

# 三. 属性 #

// 心跳锁
      private final Object heartbeatMonitor = new Object();
    
      // 停止监控 锁
      private final Object shutdownMonitor = new Object();
    
      // 环境变量
      private final Context context;
    
      // Dispatcher处理器 :
      private final Dispatcher dispatcher;
    
      // NodeId
      private NodeId nodeId;
    
      // 心跳间隔
      private long nextHeartBeatInterval;
    
      // ResourceTracker 资源监控
      private ResourceTracker resourceTracker;
    
      // 总资源
      private Resource totalResource;
    
      // 物理资源
      private Resource physicalResource;
    
      // http 端口
      private int httpPort;
    
      // nodeManager 版本ID
      private String nodeManagerVersionId;
    
      // ResourceManagerVersion 版本
      private String minimumResourceManagerVersion;
    
      // 是否停止
      private volatile boolean isStopped;
    
      // token 是否启用
      private boolean tokenKeepAliveEnabled;
    
      // token失效移除...
      private long tokenRemovalDelayMs;
    
      // 监控 ApplicationId 心跳时间
      /** Keeps track of when the next keep alive request should be sent for an app*/
      private Map<ApplicationId, Long> appTokenKeepAliveMap =   new HashMap<ApplicationId, Long>();
    
      // 随机 心跳
      private Random keepAliveDelayRandom = new Random();
    
    
    
      // 追踪node manager中最近被停止的container.
      // 避免当AM完成 通过RM停止的时候, 可能在NM 出现no-such-container exception异常的产生.
    
      // It will be used to track recently stopped containers on node manager, this
      // is to avoid the misleading no-such-container exception messages on NM, when
      // the AM finishes it informs the RM to stop the may-be-already-completed
      // containers.
      private final Map<ContainerId, Long> recentlyStoppedContainers;
      // Save the reported completed containers in case of lost heartbeat responses.
      // These completed containers will be sent again till a successful response.
    
      // 保存报告的已完成容器，以防丢失心跳响应。
      // 这些完成的容器将再次发送，直到成功响应。
      private final Map<ContainerId, ContainerStatus> pendingCompletedContainers;
    
      // 跟踪最近停止的容器的持续时间。
      // Duration for which to track recently stopped container.
      private long durationToTrackStoppedContainers;
    
      // 日志聚合是否启用
      private boolean logAggregationEnabled;
    
      // 聚合日志容器
      private final List<LogAggregationReport> logAggregationReportForAppsTempList;
    
      // Node 健康检查
      private final NodeHealthCheckerService healthChecker;
    
      // 度量信息
      private final NodeManagerMetrics metrics;
    
      // 状态检查Runnable
      private Runnable statusUpdaterRunnable;
    
      // 状态检查的 线程 Thread statusUpdater
      private Thread  statusUpdater;
      // 失败重新连接RM连接
      private boolean failedToConnect = false;
    
      // RM 验证
      private long rmIdentifier = ResourceManagerConstants.RM_INVALID_IDENTIFIER;
    
      // 是否注册
      private boolean registeredWithRM = false;
    
      // 待删除的容器
      Set<ContainerId> pendingContainersToRemove = new HashSet<ContainerId>();
    
      // 标签处理
      private NMNodeLabelsHandler nodeLabelsHandler;
    
      // 属性
      private NMNodeAttributesHandler nodeAttributesHandler;
    
      // 标签 Provider
      private NodeLabelsProvider nodeLabelsProvider;
    
      // Node属性 Provider
      private NodeAttributesProvider nodeAttributesProvider;
      
      // timelineService 是否启动...
      private boolean timelineServiceV2Enabled;

# 四. serviceInit #

执行各种初始化操作 , 这里比较重要的是获取配置中的参数: 默认 cpu 4 , vcore 8 , Mem : 8G

> yarn.nm.liveness-monitor.expiry-interval-ms : 600000 —> container 监控过期的时间 10min

// yarn.resource-types.memory-mb.minimum-allocation 如果为-1 这代表没有配置 使用 下面的值:
    // yarn.scheduler.minimum-allocation-mb : 1024
    
    // ----------------------------------------
    
    // yarn.resource-types.memory-mb.maximum-allocation 如果为-1 这代表没有配置 使用 下面的值:
    // yarn.scheduler.maximum-allocation-mb : 8192

*  代码serviceInit

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
    
        // 获取totalResource
        this.totalResource = NodeManagerHardwareUtils.getNodeResources(conf);
    
        long memoryMb = totalResource.getMemorySize();
    
        // yarn.nodemanager.vmem-pmem-ratio : 2.1f
        float vMemToPMem =
            conf.getFloat(
                YarnConfiguration.NM_VMEM_PMEM_RATIO,
                YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
    
        // 虚拟内存 物理内存 * 2.1
        long virtualMemoryMb = (long)Math.ceil(memoryMb * vMemToPMem);
    
        int virtualCores = totalResource.getVirtualCores();
    
        // Update configured resources via plugins.
        updateConfiguredResourcesViaPlugins(totalResource);
    
        LOG.info("Nodemanager resources is set to: " + totalResource);
    
        metrics.addResource(totalResource);
    
        // Get actual node physical resources
        long physicalMemoryMb = memoryMb;
        int physicalCores = virtualCores;
    
    
        ResourceCalculatorPlugin rcp =  ResourceCalculatorPlugin.getNodeResourceMonitorPlugin(conf);
        if (rcp != null) { 
          physicalMemoryMb = rcp.getPhysicalMemorySize() / (1024 * 1024);
          physicalCores = rcp.getNumProcessors();
        }
    
        this.physicalResource =  Resource.newInstance(physicalMemoryMb, physicalCores);
    
        this.tokenKeepAliveEnabled = isTokenKeepAliveEnabled(conf);
    
        // yarn.nm.liveness-monitor.expiry-interval-ms : 600000
        this.tokenRemovalDelayMs =
            conf.getInt(YarnConfiguration.RM_NM_EXPIRY_INTERVAL_MS,
                YarnConfiguration.DEFAULT_RM_NM_EXPIRY_INTERVAL_MS);
    
        // yarn.nodemanager.resourcemanager.minimum.version : NONE
        this.minimumResourceManagerVersion = conf.get(
            YarnConfiguration.NM_RESOURCEMANAGER_MINIMUM_VERSION,
            YarnConfiguration.DEFAULT_NM_RESOURCEMANAGER_MINIMUM_VERSION);
    
        nodeLabelsHandler =   createNMNodeLabelsHandler(nodeLabelsProvider);
    
        nodeAttributesHandler =  createNMNodeAttributesHandler(nodeAttributesProvider);
    
        // nodemanager 监控container 时效 : 10min
        // yarn.nodemanager.duration-to-track-stopped-containers : 10min
    
        // Default duration to track stopped containers on nodemanager is 10Min.
        // This should not be assigned very large value as it will remember all the
        // containers stopped during that time.
        durationToTrackStoppedContainers =
            conf.getLong(YARN_NODEMANAGER_DURATION_TO_TRACK_STOPPED_CONTAINERS,
              600000);
        if (durationToTrackStoppedContainers < 0) { 
          String message = "Invalid configuration for "
            + YARN_NODEMANAGER_DURATION_TO_TRACK_STOPPED_CONTAINERS + " default "
              + "value is 10Min(600000).";
          LOG.error(message);
          throw new YarnException(message);
        }
        if (LOG.isDebugEnabled()) { 
          LOG.debug(YARN_NODEMANAGER_DURATION_TO_TRACK_STOPPED_CONTAINERS + " :"
            + durationToTrackStoppedContainers);
        }
        super.serviceInit(conf);
        LOG.info("Initialized nodemanager with :" +
            " physical-memory=" + memoryMb + " virtual-memory=" + virtualMemoryMb +
            " virtual-cores=" + virtualCores);
    
        // yarn.log-aggregation-enable : false
        this.logAggregationEnabled =
            conf.getBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED,
              YarnConfiguration.DEFAULT_LOG_AGGREGATION_ENABLED);
    
    
        this.timelineServiceV2Enabled = YarnConfiguration.
            timelineServiceV2Enabled(conf);
      }

# 五.getNodeResourceInformation #

getNodeResourceInformation获取Node 资源的信息, 默认是 8core , 8G

配置参数(分配的最大资源和最小资源限制):

*  yarn.resource-types.memory-mb.minimum-allocation -> yarn.scheduler.minimum-allocation-mb
 *  yarn.resource-types.memory-mb.maximum-allocation -> yarn.scheduler.maximum-allocation-mb
 *  yarn.resource-types.vcores.minimum-allocation -> yarn.scheduler.minimum-allocation-vcores
 *  yarn.resource-types.vcores.maximum-allocation -> yarn.scheduler.maximum-allocation-vcores

/** * Function to get the resources for a node. This function will look at the * file {@link YarnConfiguration#NODE_RESOURCES_CONFIGURATION_FILE} to * determine the node resources. * * @param conf configuration file * @return a map to resource name to the ResourceInformation object. The map * is guaranteed to have entries for memory and vcores */
      public static Map<String, ResourceInformation> getNodeResourceInformation(
          Configuration conf) { 
    
    
        if (!initializedNodeResources) { 
          synchronized (ResourceUtils.class) { 
    
            // 第一次加载
            if (!initializedNodeResources) { 
    
              // 获取配置
              Map<String, ResourceInformation> nodeResources = initializeNodeResourceInformation(conf);
    
    
              checkSpecialResources(nodeResources);
    
    
              addMandatoryResources(nodeResources);
    
              // Mandatory Resource 'yarn.resource-types.memory-mb.minimum-allocation' is not configured in resource-types config file. Setting allocation specified using 'yarn.scheduler.minimum-allocation-mb'
              // Mandatory Resource 'yarn.resource-types.memory-mb.maximum-allocation' is not configured in resource-types config file. Setting allocation specified using 'yarn.scheduler.maximum-allocation-mb'
              // Mandatory Resource 'yarn.resource-types.vcores.minimum-allocation' is not configured in resource-types config file. Setting allocation specified using 'yarn.scheduler.minimum-allocation-vcores'
              // Mandatory Resource 'yarn.resource-types.vcores.maximum-allocation' is not configured in resource-types config file. Setting allocation specified using 'yarn.scheduler.maximum-allocation-vcores'
              setAllocationForMandatoryResources(nodeResources, conf);
    
    
              readOnlyNodeResources = Collections.unmodifiableMap(nodeResources);
    
    
              initializedNodeResources = true;
            }
          }
        }
        return readOnlyNodeResources;
      }

# 六.获取配置中的内存资源 #

获取内存资源,单位MB .  
这里需要注意一个参数: yarn.nodemanager.resource.detect-hardware-capabilities (是否自动检测节点的CPU和内存) ,默认值false .  
如果是false的话, 默认从配置文件加载cpu和内存, 默认是8G .  
如果开启的话, 会通过getContainerMemoryMBInternal方法去加载对应的扩展插件,进行内存的动态计算().

*  getContainerMemoryMB

/** * Function to return how much memory we should set aside for YARN containers. * If a number is specified in the configuration file, then that number is * returned. If nothing is specified - 1. If the OS is an "unknown" OS(one for * which we don't have ResourceCalculatorPlugin implemented), return the * default NodeManager physical memory. 2. If the OS has a * ResourceCalculatorPlugin implemented, the calculation is 0.8 * (RAM - 2 * * JVM-memory) i.e. use 80% of the memory after accounting for memory used by * the DataNode and the NodeManager. If the number is less than 1GB, log a * warning message. * * @param conf * - the configuration for the NodeManager * @return the amount of memory that will be used for YARN containers in MB. */
      public static long getContainerMemoryMB(Configuration conf) { 
        // 是否自动检测节点的CPU和内存
        // yarn.nodemanager.resource.detect-hardware-capabilities : false
        if (!isHardwareDetectionEnabled(conf)) { 
          // 获取配置: yarn.nodemanager.resource.memory-mb : default 8g
          return getConfiguredMemoryMB(conf);
        }
        ResourceCalculatorPlugin plugin =  ResourceCalculatorPlugin.getResourceCalculatorPlugin(null, conf);
        if (plugin == null) { 
          return getConfiguredMemoryMB(conf);
        }
        return getContainerMemoryMBInternal(plugin, conf);
      }

*  getConfiguredMemoryMB 获取配置中的内存资源

private static long getConfiguredMemoryMB(Configuration conf) { 
    
        // yarn.nodemanager.resource.memory-mb : 8g
        long memoryMb = conf.getLong(YarnConfiguration.NM_PMEM_MB,
            YarnConfiguration.DEFAULT_NM_PMEM_MB);
        if (memoryMb == -1) { 
          memoryMb = YarnConfiguration.DEFAULT_NM_PMEM_MB;
        }
        return memoryMb;
      }

*  getContainerMemoryMBInternal 通过扩展组件实现获取内存资源

> 如果 配置文件中没有配置yarn.nodemanager.resource.memory-mb, 开启自动配置模式 .  
> **计算方式 : 0.8 \* ( 实际物理机内存 - 2 \* JVM堆内存) - 系统预留内存**

private static long getContainerMemoryMBInternal(ResourceCalculatorPlugin plugin,  Configuration conf) { 
        // 首先验证配置文件中是否已经设置, 如果设置了使用配置中的资源, 如果没有配置, 才会自动计算
        // yarn.nodemanager.resource.memory-mb : -1
        long memoryMb = conf.getInt(YarnConfiguration.NM_PMEM_MB, -1);
    
        if (memoryMb == -1) { 
          // 配置文件中没有配置yarn.nodemanager.resource.memory-mb, 开启自动配置模式
    
          // 计算物理资源 , 这里返回的单位是 bytes , 所以要换成MB
          long physicalMemoryMB = (plugin.getPhysicalMemorySize() / (1024 * 1024));
    
          // 获取jvm 可以使用的最大内存.
          long hadoopHeapSizeMB = (Runtime.getRuntime().maxMemory() / (1024 * 1024));
    
          // 容器可以使用的物理内存 = 0.8 * ( 物理内存 - 2 * JVM内存 )
          long containerPhysicalMemoryMB = (long) (0.8f * (physicalMemoryMB - (2 * hadoopHeapSizeMB)));
    
          // 系统预留的内存 : yarn.nodemanager.resource.system-reserved-memory-mb 默认值 -1
          long reservedMemoryMB = conf.getInt(YarnConfiguration.NM_SYSTEM_RESERVED_PMEM_MB, -1);
    
          if (reservedMemoryMB != -1) { 
            //实际可以使用的内存为 可用的物理内存 - 系统预留内存
            containerPhysicalMemoryMB = physicalMemoryMB - reservedMemoryMB;
          }
    
          if (containerPhysicalMemoryMB <= 0) { 
            LOG.error("Calculated memory for YARN containers is too low."
                + " Node memory is " + physicalMemoryMB
                + " MB, system reserved memory is " + reservedMemoryMB + " MB.");
          }
          // 做了一个参数验证, 内存不能为负数...
          containerPhysicalMemoryMB = Math.max(containerPhysicalMemoryMB, 0);
          memoryMb = containerPhysicalMemoryMB;
        }
        if(memoryMb <= 0) { 
          String message = "Illegal value for " + YarnConfiguration.NM_PMEM_MB
              + ". Value must be greater than 0.";
          throw new IllegalArgumentException(message);
        }
        return memoryMb;
      }

# 七.获取配置中的CPU资源 #

获取内存资源,单位MB .  
这里需要注意一个参数: yarn.nodemanager.resource.detect-hardware-capabilities (是否自动检测节点的CPU和内存) ,默认值false .  
如果是false的话, 默认从配置文件加载cpu和内存, 默认是8 core  
如果开启的话, 会通过getContainerMemoryMBInternal方法去加载对应的扩展插件,进行动态计算.

*  getVCores
 *  getConfiguredVCores

默认加载 配置文件中的参数 : yarn.nodemanager.resource.cpu-vcores , default : 8 core

private static int getConfiguredVCores(Configuration conf) { 
        // yarn.nodemanager.resource.cpu-vcores : 默认 8
        int cores = conf.getInt(YarnConfiguration.NM_VCORES,
            YarnConfiguration.DEFAULT_NM_VCORES);
        if (cores == -1) { 
          cores = YarnConfiguration.DEFAULT_NM_VCORES;
        }
        return cores;
      }

*  getVCoresInternal
 *  通过插件动态获取 cpu , 如果配置文件中指定的话, 使用配置文件中的参数, 如果没有指定则自动获取.

> 计算公司 物理cpu \* 系数(默认1)

// 通过插件动态获取 cpu : 物理cpu * 系数(默认1)
      private static int getVCoresInternal(ResourceCalculatorPlugin plugin,
          Configuration conf) { 
        String message;
    
        // 获取配置文件中的参数 : yarn.nodemanager.resource.cpu-vcores : -1
    
        int cores = conf.getInt(YarnConfiguration.NM_VCORES, -1);
        if (cores == -1) { 
    
          // 如果配置文件中的参数没有设置的话,开始进入自动识别模式.
    
          // 获取通过实现类 获取物理机cpu数量 , 默认使用物理core的数量, 而不是虚拟的core数量.
          float physicalCores =  NodeManagerHardwareUtils.getContainersCPUs(plugin, conf);
    
    
          // 一个物理cpu可以虚拟成几个cpu ? 默认值 1
          // yarn.nodemanager.resource.pcores-vcores-multiplier : default 1
          float multiplier = conf.getFloat(YarnConfiguration.NM_PCORES_VCORES_MULTIPLIER,  YarnConfiguration.DEFAULT_NM_PCORES_VCORES_MULTIPLIER);
    
          if (multiplier > 0) { 
    
            // 物理机cpu * 虚拟cpu的权重
            float tmp = physicalCores * multiplier;
    
            if (tmp > 0 && tmp < 1) { 
              // on a single core machine - tmp can be between 0 and 1
              cores = 1;
            } else { 
              // 处理一下, 四舍五入获取一个 int类型的数值用于标识cpu
              cores = Math.round(tmp);
            }
          } else { 
            message = "Illegal value for "
                + YarnConfiguration.NM_PCORES_VCORES_MULTIPLIER
                + ". Value must be greater than 0.";
            throw new IllegalArgumentException(message);
          }
        }
        if(cores <= 0) { 
          message = "Illegal value for " + YarnConfiguration.NM_VCORES
              + ". Value must be greater than 0.";
          throw new IllegalArgumentException(message);
        }
        // 返回cpu的数据量
        return cores;
      }

# 八.serviceStart #

主要是启动各种服务 , 想RM注册 & 启动StatusUpdaterRunnable 线程

@Override
      protected void serviceStart() throws Exception { 
    
        // NodeManager is the last service to start, so NodeId is available.
        this.nodeId = this.context.getNodeId();
    
        LOG.info("Node ID assigned is : " + this.nodeId);
    
        // httpPort : 8042
        this.httpPort = this.context.getHttpPort();
        // nodeManagerVersionId : 3.2.1
        this.nodeManagerVersionId = YarnVersionInfo.getVersion();
        try { 
    
          // Registration has to be in start so that ContainerManager can get the perNM tokens needed to authenticate ContainerTokens.
          //
          // org.apache.hadoop.yarn.server.api.impl.pb.client.ResourceTrackerPBClientImpl@4f379769
          this.resourceTracker = getRMClient();
          // 向 RM 注册
          registerWithRM();
    
          super.serviceStart();
    
          // 启动 StatusUpdaterRunnable
          startStatusUpdater();
    
    
        } catch (Exception e) { 
          String errorMessage = "Unexpected error starting NodeStatusUpdater";
          LOG.error(errorMessage, e);
          throw new YarnRuntimeException(e);
        }
      }

# 九. registerWithRM #

NodeManger向ResourceManager注册, 注册的信息

*  请求信息  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]
 *  响应信息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

最后输出的日志信息:

> Registered with ResourceManager as bogon:53034 with total resource of <memory:104857600, vCores:16>

@VisibleForTesting
      protected void registerWithRM()
          throws YarnException, IOException { 
        RegisterNodeManagerResponse regNMResponse;
        Set<NodeLabel> nodeLabels = nodeLabelsHandler.getNodeLabelsForRegistration();
        Set<NodeAttribute> nodeAttributes =
            nodeAttributesHandler.getNodeAttributesForRegistration();
    
        // Synchronize NM-RM registration with
        // ContainerManagerImpl#increaseContainersResource and
        // ContainerManagerImpl#startContainers to avoid race condition
        // during RM recovery
        synchronized (this.context) { 
          List<NMContainerStatus> containerReports = getNMContainerStatuses();
    
          // nodeId： 该NodeManager所在的host和对外的RPC端口号。 ==> boyi-pro.lan:61671
          // httpPort : 该NodeManager对外提供的HTTP端口号， ResourceManager会在界面上提供一个可直接访问NodeManager Web界面的超链接。 ==> 8042
          // totalResource：
          // 该NodeManager所在节点总的可分配资源， 当前支持（ 物理） 内存和虚拟CPU两种资源，
          // 管理员可通过参数yarn.nodemanager.resource.cpu-vcores（ 默认是8） 和 yarn.nodemanager.resource.memory-mb （ 单位为MB， 默认是8192 )
          //
          // 还可通过参数yarn.nodemanager.vmem-pmem-ratio设置物理内存和虚拟内存使用比率， 默认是2.1， 即每使用1MB物理内存， 最多可使用2.1MB虚拟内存 配置。
          //
          //
          //
          // ==> <memory:8192, vCores:8>
          // 构建注册请求
          RegisterNodeManagerRequest request =
              RegisterNodeManagerRequest.newInstance(nodeId, httpPort, totalResource,
                  nodeManagerVersionId, containerReports, getRunningApplications(),
                  nodeLabels, physicalResource, nodeAttributes);
    
          if (containerReports != null) { 
            LOG.info("Registering with RM using containers :" + containerReports);
          }
          if (logAggregationEnabled) { 
            // pull log aggregation status for application running in this NM
            List<LogAggregationReport> logAggregationReports =
                context.getNMLogAggregationStatusTracker()
                    .pullCachedLogAggregationReports();
            if (LOG.isDebugEnabled()) { 
              LOG.debug("The cache log aggregation status size:"
                  + logAggregationReports.size());
            }
            if (logAggregationReports != null
                && !logAggregationReports.isEmpty()) { 
              request.setLogAggregationReportsForApps(logAggregationReports);
            }
          } // 发送注册请求.
          regNMResponse =  resourceTracker.registerNodeManager(request);
          // Make sure rmIdentifier is set before we release the lock
          // rmIdentifier： ResourceManager的标示符（ 用于ResourceManager 重启恢复或者HA场景） ， NodeManager通过该标识符判断ApplicationMaster发送的Container来自原始的还是新启动的ResourceManager。
          this.rmIdentifier = regNMResponse.getRMIdentifier();
        }
    
        // if the Resource Manager instructs NM to shutdown.
        if (NodeAction.SHUTDOWN.equals(regNMResponse.getNodeAction())) { 
          String message =
              "Message from ResourceManager: "
                  + regNMResponse.getDiagnosticsMessage();
          throw new YarnRuntimeException(
            "Received SHUTDOWN signal from Resourcemanager, Registration of NodeManager failed, "
                + message);
        }
    
        // if ResourceManager version is too old then shutdown
        if (!minimumResourceManagerVersion.equals("NONE")){ 
          if (minimumResourceManagerVersion.equals("EqualToNM")){ 
            minimumResourceManagerVersion = nodeManagerVersionId;
          }
          String rmVersion = regNMResponse.getRMVersion();
          if (rmVersion == null) { 
            String message = "The Resource Manager's did not return a version. "
                + "Valid version cannot be checked.";
            throw new YarnRuntimeException("Shutting down the Node Manager. "
                + message);
          }
          if (VersionUtil.compareVersions(rmVersion,minimumResourceManagerVersion) < 0) { 
            String message = "The Resource Manager's version ("
                + rmVersion +") is less than the minimum "
                + "allowed version " + minimumResourceManagerVersion;
            throw new YarnRuntimeException("Shutting down the Node Manager on RM "
                + "version error, " + message);
          }
        }
        this.registeredWithRM = true;
    
        // 新生成的Container Token和Node Token的Master Key。
        MasterKey masterKey = regNMResponse.getContainerTokenMasterKey();
        // do this now so that its set before we start heartbeating to RM
        // It is expected that status updater is started by this point and
        // RM gives the shared secret in registration during
        // StatusUpdater#start().
        if (masterKey != null) { 
          this.context.getContainerTokenSecretManager().setMasterKey(masterKey);
        }
    
        masterKey = regNMResponse.getNMTokenMasterKey();
        if (masterKey != null) { 
          this.context.getNMTokenSecretManager().setMasterKey(masterKey);
        }
    
        StringBuilder successfullRegistrationMsg = new StringBuilder();
        successfullRegistrationMsg.append("Registered with ResourceManager as ")
            .append(this.nodeId);
    
        Resource newResource = regNMResponse.getResource();
        if (newResource != null) { 
          updateNMResource(newResource);
          successfullRegistrationMsg.append(" with updated total resource of ")
              .append(this.totalResource);
        } else { 
          successfullRegistrationMsg.append(" with total resource of ")
              .append(this.totalResource);
        }
    
        successfullRegistrationMsg.append(nodeLabelsHandler
            .verifyRMRegistrationResponseForNodeLabels(regNMResponse));
        successfullRegistrationMsg.append(nodeAttributesHandler
            .verifyRMRegistrationResponseForNodeAttributes(regNMResponse));
    
        LOG.info(successfullRegistrationMsg.toString());
      }

# 十.startStatusUpdater #

startStatusUpdater方法就简单了, 直接用一个线程启动statusUpdaterRunnable 就可以了.

protected void startStatusUpdater() { 
        statusUpdaterRunnable = new StatusUpdaterRunnable();
        statusUpdater =
            new Thread(statusUpdaterRunnable, "Node Status Updater");
        statusUpdater.start();
      }

# 十一. statusUpdaterRunnable #

statusUpdaterRunnable 主要的作用是向RM \[ ResourceTrackerService 8031端口\]发送心跳数据. 并根据心跳的反馈处理任务.

*  请求信息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

// 构建心跳请求
              NodeHeartbeatRequest request =
                  NodeHeartbeatRequest.newInstance(nodeStatus,
                      NodeStatusUpdaterImpl.this.context
                          .getContainerTokenSecretManager().getCurrentKey(),
                      NodeStatusUpdaterImpl.this.context
                          .getNMTokenSecretManager().getCurrentKey(),
                      nodeLabelsForHeartbeat,
                      nodeAttributesForHeartbeat,
                      NodeStatusUpdaterImpl.this.context
                          .getRegisteringCollectors());

*  响应信息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]

*  向RM发送心跳数据

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 7]

*  代码

@Override
        @SuppressWarnings("unchecked")
        public void run() { 
          int lastHeartbeatID = 0;
          while (!isStopped) { 
            // Send heartbeat
            try { 
              NodeHeartbeatResponse response = null;
    
              // 标签信息
              Set<NodeLabel> nodeLabelsForHeartbeat =  nodeLabelsHandler.getNodeLabelsForHeartbeat();
    
              // 属性信息
              Set<NodeAttribute> nodeAttributesForHeartbeat =  nodeAttributesHandler.getNodeAttributesForHeartbeat();
    
              // 节点的状态信息
              NodeStatus nodeStatus = getNodeStatus(lastHeartbeatID);
    
              // 构建心跳请求
              NodeHeartbeatRequest request =
                  NodeHeartbeatRequest.newInstance(nodeStatus,
                      NodeStatusUpdaterImpl.this.context
                          .getContainerTokenSecretManager().getCurrentKey(),
                      NodeStatusUpdaterImpl.this.context
                          .getNMTokenSecretManager().getCurrentKey(),
                      nodeLabelsForHeartbeat,
                      nodeAttributesForHeartbeat,
                      NodeStatusUpdaterImpl.this.context
                          .getRegisteringCollectors());
    
              // 是否开启日志聚合
              if (logAggregationEnabled) { 
                // pull log aggregation status for application running in this NM
                List<LogAggregationReport> logAggregationReports = getLogAggregationReportsForApps(context .getLogAggregationStatusForApps());
    
                // 发送日志信息
                if (logAggregationReports != null
                    && !logAggregationReports.isEmpty()) { 
                  request.setLogAggregationReportsForApps(logAggregationReports);
                }
              }
    
              // 获取响应信息
              response = resourceTracker.nodeHeartbeat(request);
    
    
              //get next heartbeat interval from response
    
              // 更新心跳间隔
              nextHeartBeatInterval = response.getNextHeartBeatInterval();
    
              // 更新token
              updateMasterKeys(response);
    
              // 处理响应数据
              if (!handleShutdownOrResyncCommand(response)) { 
    
                // label 验证 心跳请求
                nodeLabelsHandler.verifyRMHeartbeatResponseForNodeLabels( response);
    
                // node 属性 处理心跳请求.
                nodeAttributesHandler.verifyRMHeartbeatResponseForNodeAttributes(response);
    
                // Explicitly put this method after checking the resyncresponse.
                // We don't want to remove the completed containers before resync because these completed containers will be reported back to RM when NM re-registers with RM.
                // Only remove the cleanedup containers that are acked
    
    
                // 在检测resyncresponse之后显示的调用此方法
                // 在同步信息之前,不会移除已经完成的containers , 因为已经完成的containers信息需要在重新注册的时候重新连接.
                // 仅仅会移除被请求 移除的 containers
                removeOrTrackCompletedContainersFromContext(response.getContainersToBeRemovedFromNM());
    
                // 清理日志聚合数据
                logAggregationReportForAppsTempList.clear();
    
                lastHeartbeatID = response.getResponseId();
    
                // 获取需要 清理的容器
                List<ContainerId> containersToCleanup = response.getContainersToCleanup();
                if (!containersToCleanup.isEmpty()) { 
    
                  // 执行清理操作
                  dispatcher.getEventHandler().handle(
                      new CMgrCompletedContainersEvent(containersToCleanup,
                          CMgrCompletedContainersEvent.Reason
                              .BY_RESOURCEMANAGER));
                }
    
                // 获取需要 清理的 apps
                List<ApplicationId> appsToCleanup = response.getApplicationsToCleanup();
                //Only start tracking for keepAlive on FINISH_APP
                trackAppsForKeepAlive(appsToCleanup);
                if (!appsToCleanup.isEmpty()) { 
    
                  // 执行app 的清理操作
                  dispatcher.getEventHandler().handle(
                      new CMgrCompletedAppsEvent(appsToCleanup,
                          CMgrCompletedAppsEvent.Reason.BY_RESOURCEMANAGER));
                }
    
                // 更新凭证信息
                Map<ApplicationId, ByteBuffer> systemCredentials =  response.getSystemCredentialsForApps();
                if (systemCredentials != null && !systemCredentials.isEmpty()) { 
                  ((NMContext) context).setSystemCrendentialsForApps(
                      parseCredentials(systemCredentials));
                  context.getContainerManager().handleCredentialUpdate();
                }
    
    
                // 处理需要更新的容器信息
                List<org.apache.hadoop.yarn.api.records.Container> containersToUpdate = response.getContainersToUpdate();
                if (!containersToUpdate.isEmpty()) { 
                  dispatcher.getEventHandler().handle(
                      new CMgrUpdateContainersEvent(containersToUpdate));
                }
    
                // 处理容器中的指令
                // SignalContainer request originally comes from end users via ClientRMProtocol's SignalContainer.
                // Forward the request to ContainerManager which will dispatch the event to ContainerLauncher.
                List<SignalContainerRequest> containersToSignal = response.getContainersToSignalList();
                if (!containersToSignal.isEmpty()) { 
                  dispatcher.getEventHandler().handle(
                      new CMgrSignalContainersEvent(containersToSignal));
                }
    
                // 如果ContainerManager支持排队，则更新QueuingLimits
                // Update QueuingLimits if ContainerManager supports queuing
                ContainerQueuingLimit queuingLimit =  response.getContainerQueuingLimit();
    
                if (queuingLimit != null) { 
                  context.getContainerManager().updateQueuingLimit(queuingLimit);
                }
              }
    
              // 处理资源的更新
              // Handling node resource update case.
              Resource newResource = response.getResource();
              if (newResource != null) { 
    
                // 更新NM的资源
                updateNMResource(newResource);
                if (LOG.isDebugEnabled()) { 
                  LOG.debug("Node's resource is updated to " +
                      newResource.toString());
                }
              }
              if (timelineServiceV2Enabled) { 
                // 更新 timeline
                updateTimelineCollectorData(response);
              }
    
            } catch (ConnectException e) { 
              //catch and throw the exception if tried MAX wait time to connect RM
              dispatcher.getEventHandler().handle(
                  new NodeManagerEvent(NodeManagerEventType.SHUTDOWN));
              // failed to connect to RM.
              failedToConnect = true;
              throw new YarnRuntimeException(e);
            } catch (Exception e) { 
    
              // TODO Better error handling. Thread can die with the rest of the
              // NM still running.
              LOG.error("Caught exception in status-updater", e);
            } finally { 
              synchronized (heartbeatMonitor) { 
                // 处理心跳
                nextHeartBeatInterval = nextHeartBeatInterval <= 0 ?
                    YarnConfiguration.DEFAULT_RM_NM_HEARTBEAT_INTERVAL_MS :
                    nextHeartBeatInterval;
                try { 
                  heartbeatMonitor.wait(nextHeartBeatInterval);
                } catch (InterruptedException e) { 
                  // Do Nothing
                }
              }
            }
          }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221119/4c993b91d9664b7eab21eef5a60d775b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221119/3c77fe11d6304621b9e5c96d98814248.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210117193952233.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210117194035475.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210117201540850.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210117201718916.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210117201736780.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20210117230618212.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70