devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Hadoop-Yarn-NodeManager是如何监控容器的

作者：我家小花儿 | 2024-03-02 18:57:32

赞

踩

Hadoop-Yarn-NodeManager是如何监控容器的

一、源码下载

下面是hadoop官方源码下载地址，我下载的是hadoop-3.2.4，那就一起来看下吧

Index of /dist/hadoop/core

二、上下文

在我的博客<Hadoop-Yarn-NodeManager是如何启动容器的>中的ContainerLaunch  prepareForLaunch()会触发ContainerEventType.CONTAINER_LAUNCHED事件，ContainerImpl会处理该事件，监控该容器的资源使用以及处理后续操作，下面让我们把源码捋起来吧。

三、开始捋源码

1、ContainerImpl

  1. public class ContainerImpl implements Container {
  2.   private static StateMachineFactory
  3.            <ContainerImpl, ContainerState, ContainerEventType, ContainerEvent>
  4.         stateMachineFactory =
  5.       new StateMachineFactory<ContainerImpl, ContainerState, ContainerEventType,             ContainerEvent>(ContainerState.NEW).
  6.     //......省略其他事件处理......
  7.     addTransition(ContainerState.SCHEDULED, ContainerState.RUNNING,
  8.         ContainerEventType.CONTAINER_LAUNCHED, new LaunchTransition())
  9.     //......省略其他事件处理......
  10.     .installTopology();
  11.  
  12.   static class LaunchTransition extends ContainerTransition {
  13.     @SuppressWarnings("unchecked")
  14.     @Override
  15.     public void transition(ContainerImpl container, ContainerEvent event) {
  16.       //发送容器监控事件，去监控容器的使用
  17.       container.sendContainerMonitorStartEvent();
  18.       container.metrics.runningContainer();
  19.       container.wasLaunched  = true;
  20.  
  21.       if (container.isReInitializing()) {
  22.         NMAuditLogger.logSuccess(container.user,
  23.             AuditConstants.FINISH_CONTAINER_REINIT, "ContainerImpl",
  24.             container.containerId.getApplicationAttemptId().getApplicationId(),
  25.             container.containerId);
  26.       }
  27.       container.setIsReInitializing(false);
  28.       // Check if this launch was due to a re-initialization.
  29.       // If autocommit == true, then wipe the re-init context. This ensures
  30.       // that any subsequent failures do not trigger a rollback.
  31.       if (container.reInitContext != null
  32.           && !container.reInitContext.canRollback()) {
  33.         container.reInitContext = null;
  34.       }
  35.  
  36.       if (container.recoveredAsKilled) {
  37.         LOG.info("Killing " + container.containerId
  38.             + " due to recovered as killed");
  39.         container.addDiagnostics("Container recovered as killed.\n");
  40.         container.dispatcher.getEventHandler().handle(
  41.             new ContainersLauncherEvent(container,
  42.                 ContainersLauncherEventType.CLEANUP_CONTAINER));
  43.       }
  44.     }
  45.   }
  46.  
  47.   private void sendContainerMonitorStartEvent() {
  48.     long launchDuration = clock.getTime() - containerLaunchStartTime;
  49.     metrics.addContainerLaunchDuration(launchDuration);
  50.  
  51.     long pmemBytes = getResource().getMemorySize() * 1024 * 1024L;
  52.     float pmemRatio = daemonConf.getFloat(
  53.         YarnConfiguration.NM_VMEM_PMEM_RATIO,
  54.         YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
  55.     long vmemBytes = (long) (pmemRatio * pmemBytes);
  56.     int cpuVcores = getResource().getVirtualCores();
  57.     long localizationDuration = containerLaunchStartTime -
  58.         containerLocalizationStartTime;
  59.     //这里会触发 ContainersMonitorEventType.START_MONITORING_CONTAINER
  60.     //该事件由ContainersMonitorImpl处理
  61.     dispatcher.getEventHandler().handle(
  62.         new ContainerStartMonitoringEvent(containerId,
  63.         vmemBytes, pmemBytes, cpuVcores, launchDuration,
  64.         localizationDuration));
  65.   }
  66.  
  67. }

2、ContainersMonitorImpl

监视收集资源使用情况的容器，并在容器超出限制时抢占容器

  1. public class ContainersMonitorImpl extends AbstractService implements
  2.     ContainersMonitor {
  3.  
  4.   private final static Logger LOG =
  5.        LoggerFactory.getLogger(ContainersMonitorImpl.class);
  6.   private final static Logger AUDITLOG =
  7.        LoggerFactory.getLogger(ContainersMonitorImpl.class.getName()+".audit");
  8.  
  9.   private long monitoringInterval;
  10.   private MonitoringThread monitoringThread;
  11.   private int logCheckInterval;
  12.   private LogMonitorThread logMonitorThread;
  13.   private long logDirSizeLimit;
  14.   private long logTotalSizeLimit;
  15.   private CGroupElasticMemoryController oomListenerThread;
  16.   private boolean containerMetricsEnabled;
  17.   private long containerMetricsPeriodMs;
  18.   private long containerMetricsUnregisterDelayMs;
  19.  
  20.   @VisibleForTesting
  21.   final Map<ContainerId, ProcessTreeInfo> trackingContainers =
  22.       new ConcurrentHashMap<>();
  23.  
  24.   private final ContainerExecutor containerExecutor;
  25.   private final Dispatcher eventDispatcher;
  26.   private final Context context;
  27.   private ResourceCalculatorPlugin resourceCalculatorPlugin;
  28.   private Configuration conf;
  29.   private static float vmemRatio;
  30.   //用于获取进程资源使用情况的接口类
  31.   //注意：此类不应由外部用户使用，而只能由外部开发人员使用，以扩展和包括他们自己的流程树实现，尤其是对于Linux和Windows以外的平台。
  32.   private Class<? extends ResourceCalculatorProcessTree> processTreeClass;
  33.  
  34.   private long maxVmemAllottedForContainers = UNKNOWN_MEMORY_LIMIT;
  35.   private long maxPmemAllottedForContainers = UNKNOWN_MEMORY_LIMIT;
  36.  
  37.   private boolean pmemCheckEnabled;
  38.   private boolean vmemCheckEnabled;
  39.   private boolean elasticMemoryEnforcement;
  40.   private boolean strictMemoryEnforcement;
  41.   private boolean containersMonitorEnabled;
  42.   private boolean logMonitorEnabled;
  43.  
  44.   private long maxVCoresAllottedForContainers;
  45.  
  46.   private static final long UNKNOWN_MEMORY_LIMIT = -1L;
  47.   private int nodeCpuPercentageForYARN;
  48.  
  49.   /**
  50.    * 容器度量的类型
  51.    */
  52.   @Private
  53.   public enum ContainerMetric {
  54.     CPU, MEMORY
  55.   }
  56.  
  57.   //ResourceUtilization对集群中一组计算机资源的利用率进行建模
  58.   private ResourceUtilization containersUtilization;
  59.  
  60.   private volatile boolean stopped = false;
  61.  
  62.   public ContainersMonitorImpl(ContainerExecutor exec,
  63.       AsyncDispatcher dispatcher, Context context) {
  64.     super("containers-monitor");
  65.  
  66.     this.containerExecutor = exec;
  67.     this.eventDispatcher = dispatcher;
  68.     this.context = context;
  69.  
  70.     this.monitoringThread = new MonitoringThread();
  71.  
  72.     this.logMonitorThread = new LogMonitorThread();
  73.  
  74. 	//ResourceUtilization.newInstance(物理内存, 虚拟内存, cpu利用率)
  75.     this.containersUtilization = ResourceUtilization.newInstance(0, 0, 0.0f);
  76.   }
  77.  
  78.   @Override
  79.   protected void serviceInit(Configuration myConf) throws Exception {
  80.     this.conf = myConf;
  81. 	//监视容器的频率
  82. 	//获取 yarn.nodemanager.container-monitor.interval-ms 的值 
  83. 	//如果未设置，则将使用yarn.nodemanager.resource-monitor.interval-ms的值。如果为0或为负数，则禁用容器监视。
  84. 	//监视节点和容器的频率
  85. 	//获取 yarn.nodemanager.resource-monitor.interval-ms 的值 默认值 3000ms 即 3s 如果为0或为负数，则禁用监视
  86.     this.monitoringInterval =
  87.         this.conf.getLong(YarnConfiguration.NM_CONTAINER_MON_INTERVAL_MS,
  88.             this.conf.getLong(YarnConfiguration.NM_RESOURCE_MON_INTERVAL_MS,
  89.                 YarnConfiguration.DEFAULT_NM_RESOURCE_MON_INTERVAL_MS));
  90. 	//检查容器日志目录使用情况的频率（以毫秒为单位）
  91. 	//获取 yarn.nodemanager.container-log-monitor.interval-ms 的值 默认值 60000ms 即 1min
  92.     this.logCheckInterval =
  93.         conf.getInt(YarnConfiguration.NM_CONTAINER_LOG_MON_INTERVAL_MS,
  94.             YarnConfiguration.DEFAULT_NM_CONTAINER_LOG_MON_INTERVAL_MS);
  95. 	//单个容器日志目录的磁盘空间限制（以字节为单位）1GB = 1024MB = 1024*1024KB = 1024*1024*1024B B就是字节
  96. 	//获取 yarn.nodemanager.container-log-monitor.dir-size-limit-bytes 的值 默认值 1000000000L 约等于 1G 
  97.     this.logDirSizeLimit =
  98.         conf.getLong(YarnConfiguration.NM_CONTAINER_LOG_DIR_SIZE_LIMIT_BYTES,
  99.             YarnConfiguration.DEFAULT_NM_CONTAINER_LOG_DIR_SIZE_LIMIT_BYTES);
  100. 	//容器所有日志的磁盘空间限制（以字节为单位）
  101. 	//获取 yarn.nodemanager.container-log-monitor.total-size-limit-bytes 的值 默认值 10000000000L 即 10G
  102.     this.logTotalSizeLimit =
  103.         conf.getLong(YarnConfiguration.NM_CONTAINER_LOG_TOTAL_SIZE_LIMIT_BYTES,
  104.             YarnConfiguration.DEFAULT_NM_CONTAINER_LOG_TOTAL_SIZE_LIMIT_BYTES);
  105.  
  106. 	//用于计算系统上的资源信息的插件，如果未配置插件，此方法将尝试返回可用于此系统的内存计算器插件。
  107. 	//先获取 yarn.nodemanager.container-monitor.resource-calculator.class (计算当前资源利用率的类) 的值 默认空
  108. 	//再获取 yarn.nodemanager.resource-calculator.class (计算当前资源利用率的类) 的值 默认空
  109. 	//如果都为空会判断操作系统，LINUX 返回 SysInfoLinux WINDOWS 返回 SysInfoWindows
  110.     this.resourceCalculatorPlugin =
  111.         ResourceCalculatorPlugin.getContainersMonitorPlugin(this.conf);
  112. 		
  113.     LOG.info(" Using ResourceCalculatorPlugin : "
  114.         + this.resourceCalculatorPlugin);
  115. 	//获取 yarn.nodemanager.container-monitor.process-tree.class (用于计算进程树资源利用率) 的值 默认为空
  116.     processTreeClass = this.conf.getClass(
  117.             YarnConfiguration.NM_CONTAINER_MON_PROCESS_TREE, null,
  118.             ResourceCalculatorProcessTree.class);
  119.     LOG.info(" Using ResourceCalculatorProcessTree : "
  120.         + this.processTreeClass);
  121.  
  122. 	//启用容器度量的标志
  123. 	//获取 yarn.nodemanager.container-metrics.enable 的值 默认 true 
  124.     this.containerMetricsEnabled =
  125.         this.conf.getBoolean(YarnConfiguration.NM_CONTAINER_METRICS_ENABLE,
  126.             YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_ENABLE);
  127. 	//容器度量刷新周期（毫秒）。设置为-1表示完成时刷新
  128. 	//获取 yarn.nodemanager.container-metrics.period-ms 的值 默认为-1
  129.     this.containerMetricsPeriodMs =
  130.         this.conf.getLong(YarnConfiguration.NM_CONTAINER_METRICS_PERIOD_MS,
  131.             YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_PERIOD_MS);
  132. 	//完成后注销容器度量的延迟时间ms
  133. 	//获取 yarn.nodemanager.container-metrics.unregister-delay-ms 的值 默认 10000ms 即 10s
  134.     this.containerMetricsUnregisterDelayMs = this.conf.getLong(
  135.         YarnConfiguration.NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS,
  136.         YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS);
  137.  
  138. 	//NodeManagerHardwareUtils：用于确定与硬件相关的特性，例如节点上的处理器数量和内存量
  139. 	//函数返回应该为YARN容器留出多少内存。如果在配置文件中指定了一个数字，则会返回该数字。如果未指定任何内容，则为-1。
  140. 	//如果操作系统是“未知”操作系统（我们没有为其实现ResourceCalculatorPlugin），则返回默认的NodeManager物理内存。
  141. 	//如果操作系统实现了ResourceCalculatorPlugin，则计算为0.8*（RAM-2*JVM内存），即在考虑了DataNode和NodeManager使用的内存后，使用80%的内存。
  142. 	//如果数字小于1GB，请记录一条警告消息
  143. 	//获取 yarn.nodemanager.resource.detect-hardware-capabilities (启用节点功能的自动检测，如内存和CPU) 的值 默认 false
  144. 	//如果为 false ，即默认会 获取配置文件中的数字 yarn.nodemanager.resource.memory-mb (可分配给容器的内存量（MB）) 
  145. 	//这里 源码 和 官方文档 有出入 ，官方文档默认值为-1 源码默认值为 8 * 1024 MB 即 8G ，如果设置为 -1 源码还是会更改为 8G ，可以设置其他值
  146. 	//返回的值是 8*1024 这里又 * 1024 * 1024L 即为 转换为 8G 对应的字节 B 
  147.     long configuredPMemForContainers =
  148.         NodeManagerHardwareUtils.getContainerMemoryMB(
  149.             this.resourceCalculatorPlugin, this.conf) * 1024 * 1024L;
  150.  
  151. 	//函数返回系统上可用于YARN容器的vcore数。如果在配置文件中指定了一个数字，则会返回该数字。如果未指定任何内容，则为-1。
  152. 	//如果操作系统是“未知”操作系统（我们没有为其实现ResourceCalculatorPlugin），则返回默认的NodeManager内核。
  153. 	//2.如果配置变量yarn.nodemanager.cpu.use_logical_processers设置为true，则返回逻辑处理器计数（将超线程计数为核心），否则返回物理核心计数。
  154. 	//获取 yarn.nodemanager.resource.cpu-vcores (可分配给容器的虚拟CPU内核数) 的值 
  155. 	//可以分配给容器的vcore数。这是RM调度程序在为容器分配资源时使用的。这并不用于限制YARN容器使用的CPU数量。如果它设置为-1，
  156. 	//并且yarn.nodemanager.resource.detect-hardware-cability为true，则在Windows和Linux的情况下，它将自动从硬件中确定。
  157. 	//在其他情况下，默认情况下vcore的数量为8。
  158.     long configuredVCoresForContainers =
  159.         NodeManagerHardwareUtils.getVCores(this.resourceCalculatorPlugin,
  160.             this.conf);
  161.  
  162.     //无论是否启用检查，都要设置这些。UI中必需
  163.     // / 物理内存配置 //
  164. 	//maxPmemAllottedForContainers = 8G 
  165. 	//maxVCoresAllottedForContainers = 8个虚拟核
  166. 	//这样看来 默认的容器能申请到的最多的资源为 8vc 8G
  167.     this.maxPmemAllottedForContainers = configuredPMemForContainers;
  168.     this.maxVCoresAllottedForContainers = configuredVCoresForContainers;
  169.  
  170.     // / 虚拟内存配置 //
  171. 	//获取 yarn.nodemanager.vmem-pmem-ratio 的值 默认 2.1
  172. 	//为容器设置内存限制时，虚拟内存与物理内存之间的比率。容器分配是以物理内存的形式表示的，虚拟内存的使用率可以超过此分配比例。
  173.     vmemRatio = this.conf.getFloat(YarnConfiguration.NM_VMEM_PMEM_RATIO,
  174.         YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
  175. 	//校验 为容器设置的内存限制比率，必须大于 0.99
  176.     Preconditions.checkArgument(vmemRatio > 0.99f,
  177.         YarnConfiguration.NM_VMEM_PMEM_RATIO + " should be at least 1.0");
  178. 	//容器可分配的最大虚拟默认为 ： 2.1 * 8 = 16.8 G
  179.     this.maxVmemAllottedForContainers =
  180.         (long) (vmemRatio * configuredPMemForContainers);
  181.  
  182. 	//是否将对容器强制执行物理内存限制
  183. 	//获取 yarn.nodemanager.pmem-check-enabled 的值 默认 true 
  184.     pmemCheckEnabled = this.conf.getBoolean(
  185.         YarnConfiguration.NM_PMEM_CHECK_ENABLED,
  186.         YarnConfiguration.DEFAULT_NM_PMEM_CHECK_ENABLED);
  187. 	//是否将对容器强制执行虚拟内存限制
  188. 	//获取 yarn.nodemanager.vmem-check-enabled 的值 默认 true 
  189.     vmemCheckEnabled = this.conf.getBoolean(
  190.         YarnConfiguration.NM_VMEM_CHECK_ENABLED,
  191.         YarnConfiguration.DEFAULT_NM_VMEM_CHECK_ENABLED);
  192. 	//启用弹性内存控制。这是Linux独有的功能。启用后，如果所有容器都超过了限制，则节点管理器会添加一个侦听器来接收事件。
  193. 	//限制由yarn.nodemanager.resource.memory-mb指定。如果未设置此项，则会根据功能设置限制。
  194. 	//有关详细信息，请参阅yarn.nodemanager.resource.detect-hardware-cability。该限制适用于物理或虚拟（rss+交换）内存，
  195. 	//具体取决于是否设置了yarn.nodemanager.pmem-check-enabled或yarn.node manager.vmem-check-enabled。
  196. 	//获取 yarn.nodemanager.elastic-memory-control.enabled 的值 默认 false 
  197.     elasticMemoryEnforcement = this.conf.getBoolean(
  198.         YarnConfiguration.NM_ELASTIC_MEMORY_CONTROL_ENABLED,
  199.         YarnConfiguration.DEFAULT_NM_ELASTIC_MEMORY_CONTROL_ENABLED);
  200. 	//是否启用YARN CGroups严格内存强制，顾名思义就是资源一旦超过设置的限制就会里面kill掉
  201. 	//获取 yarn.nodemanager.resource.memory.enforced 的值 默认 true
  202.     strictMemoryEnforcement = conf.getBoolean(
  203.         YarnConfiguration.NM_MEMORY_RESOURCE_ENFORCED,
  204.         YarnConfiguration.DEFAULT_NM_MEMORY_RESOURCE_ENFORCED);
  205.     LOG.info("Physical memory check enabled: " + pmemCheckEnabled);
  206.     LOG.info("Virtual memory check enabled: " + vmemCheckEnabled);
  207.     LOG.info("Elastic memory control enabled: " + elasticMemoryEnforcement);
  208.     LOG.info("Strict memory control enabled: " + strictMemoryEnforcement);
  209.  
  210. 	//默认不开启弹性内存控制，这段逻辑不走
  211.     if (elasticMemoryEnforcement) {
  212.       if (!CGroupElasticMemoryController.isAvailable()) {
  213.         // Test for availability outside the constructor
  214.         // to be able to write non-Linux unit tests for
  215.         // CGroupElasticMemoryController
  216.         throw new YarnException(
  217.             "CGroup Elastic Memory controller enabled but " +
  218.             "it is not available. Exiting.");
  219.       } else {
  220.         this.oomListenerThread = new CGroupElasticMemoryController(
  221.             conf,
  222.             context,
  223.             ResourceHandlerModule.getCGroupsHandler(),
  224.             pmemCheckEnabled,
  225.             vmemCheckEnabled,
  226.             pmemCheckEnabled ?
  227.                 maxPmemAllottedForContainers : maxVmemAllottedForContainers
  228.         );
  229.       }
  230.     }
  231.  
  232. 	//isContainerMonitorEnabled() 默认为 true 
  233. 	//monitoringInterval 默认 3000ms 即 3s
  234. 	//因此 containersMonitorEnabled 默认为 true 容器监视默认是开启的
  235.     containersMonitorEnabled =
  236.         isContainerMonitorEnabled() && monitoringInterval > 0;
  237.     LOG.info("ContainersMonitor enabled: " + containersMonitorEnabled);
  238.  
  239. 	//用于启用容器日志监视器的标志，该监视器强制执行容器日志目录大小限制
  240. 	//获取 yarn.nodemanager.container-log-monitor.enable 的值 默认 false
  241.     logMonitorEnabled =
  242.             conf.getBoolean(YarnConfiguration.NM_CONTAINER_LOG_MONITOR_ENABLED,
  243.                     YarnConfiguration.DEFAULT_NM_CONTAINER_LOG_MONITOR_ENABLED);
  244.     LOG.info("Container Log Monitor Enabled: "+ logMonitorEnabled);
  245.  
  246. 	//获取为YARN容器配置的物理CPU的百分比。返回值是 0 ~ 100
  247. 	//可以分配给容器的CPU百分比。此设置允许用户限制YARN容器使用的CPU数量。目前仅在使用cgroups的Linux上运行。默认情况是使用100%的CPU。
  248. 	//获取 yarn.nodemanager.resource.percentage-physical-cpu-limit 的值 默认值 100
  249. 	//nodeCpuPercentageForYARN 默认为 100
  250.     nodeCpuPercentageForYARN =
  251.         NodeManagerHardwareUtils.getNodeCpuPercentage(this.conf);
  252.  
  253. 	//默认为 true 对容器强制执行物理内存限制
  254.     if (pmemCheckEnabled) {
  255.       //如果无法确定实际设备，则记录下
  256.       long totalPhysicalMemoryOnNM = UNKNOWN_MEMORY_LIMIT;
  257. 	  //默认操作系统是LINUX resourceCalculatorPlugin = SysInfoLinux
  258.       if (this.resourceCalculatorPlugin != null) {
  259. 		//SysInfoLinux 只读取/proc/meminfo、解析和计算一次内存信息。给  ramSize、hardwareCorruptSize、hugePagesTotal、hugePageSize赋值
  260. 		//totalPhysicalMemoryOnNM =  (ramSize - hardwareCorruptSize - (hugePagesTotal * hugePageSize)) * 1024
  261. 		//totalPhysicalMemoryOnNM =  (ram磁盘空间 - ram已损坏空间 - (保留的标准大页 * 每个标准大页的大小)) * 1024
  262. 		//可以参考我的这篇 <Hadoop-Yarn-NodeManager如何计算Linux系统上的资源信息> 博客中了解
  263. 		//ramSize : ram 磁盘空间
  264. 		//hardwareCorruptSize : RAM已损坏且不可用大小
  265. 		//hugePagesTotal : 保留的标准大页
  266. 		//hugePageSize : 每个标准大页的大小
  267.         totalPhysicalMemoryOnNM = this.resourceCalculatorPlugin
  268.             .getPhysicalMemorySize();
  269.         if (totalPhysicalMemoryOnNM <= 0) {
  270.           LOG.warn("NodeManager's totalPmem could not be calculated. "
  271.               + "Setting it to " + UNKNOWN_MEMORY_LIMIT);
  272.           totalPhysicalMemoryOnNM = UNKNOWN_MEMORY_LIMIT;
  273.         }
  274.       }
  275.  
  276. 	  //分配给容器的物理内存，占可用物理内存总量的80%以上可能会发生Thrashing
  277.       if (totalPhysicalMemoryOnNM != UNKNOWN_MEMORY_LIMIT &&
  278.           this.maxPmemAllottedForContainers > totalPhysicalMemoryOnNM * 0.80f) {
  279.         LOG.warn("NodeManager configured with "
  280.             + TraditionalBinaryPrefix.long2String(maxPmemAllottedForContainers,
  281.                 "", 1)
  282.             + " physical memory allocated to containers, which is more than "
  283.             + "80% of the total physical memory available ("
  284.             + TraditionalBinaryPrefix.long2String(totalPhysicalMemoryOnNM, "",
  285.                 1) + "). Thrashing might happen.");
  286.       }
  287.     }
  288.     super.serviceInit(this.conf);
  289.   }
  290.  
  291.   //是否启用容器监视器
  292.   //获取 yarn.nodemanager.container-monitor.enabled 的值 默认 true
  293.   private boolean isContainerMonitorEnabled() {
  294.     return conf.getBoolean(YarnConfiguration.NM_CONTAINER_MONITOR_ENABLED,
  295.         YarnConfiguration.DEFAULT_NM_CONTAINER_MONITOR_ENABLED);
  296.   }
  297.  
  298.   /**
  299.    * 获取最佳进程树计算器
  300.    * @param pId container process id
  301.    * @return process tree calculator
  302.    */
  303.   private ResourceCalculatorProcessTree
  304.       getResourceCalculatorProcessTree(String pId) {
  305.     return ResourceCalculatorProcessTree.
  306.         getResourceCalculatorProcessTree(
  307.             pId, processTreeClass, conf);
  308.   }
  309.  
  310.   private boolean isResourceCalculatorAvailable() {
  311.     if (resourceCalculatorPlugin == null) {
  312.       LOG.info("ResourceCalculatorPlugin is unavailable on this system. " + this
  313.           .getClass().getName() + " is disabled.");
  314.       return false;
  315.     }
  316.     if (getResourceCalculatorProcessTree("0") == null) {
  317.       LOG.info("ResourceCalculatorProcessTree is unavailable on this system. "
  318.           + this.getClass().getName() + " is disabled.");
  319.       return false;
  320.     }
  321.     return true;
  322.   }
  323.  
  324.   @Override
  325.   protected void serviceStart() throws Exception {
  326. 	//containersMonitorEnabled 默认为 true 容器监视默认是开启的
  327.     if (containersMonitorEnabled) {
  328. 	  //起一个线程对容器进行监视
  329.       this.monitoringThread.start();
  330.     }
  331. 	//默认不开启弹性内存控制
  332.     if (oomListenerThread != null) {
  333. 	  //如果开启基于cgroups的一种弹性内存控制，允许某些container可以使用超过设定值的资源，只要不超过整体的阈值。
  334. 	  //因此会启动这个线程oomListenerThread监控是否超过了整体的阈值
  335.       oomListenerThread.start();
  336.     }
  337. 	//容器日志监视器默认关闭
  338.     if (logMonitorEnabled) {
  339.       this.logMonitorThread.start();
  340.     }
  341.     super.serviceStart();
  342.   }
  343.  
  344.   
  345.  
  346.   private class MonitoringThread extends Thread {
  347.     MonitoringThread() {
  348.       super("Container Monitor");
  349.     }
  350.  
  351.     @Override
  352.     public void run() {
  353.  
  354.       while (!stopped && !Thread.currentThread().isInterrupted()) {
  355.         // 打印processTrees以进行调试
  356.         if (LOG.isDebugEnabled()) {
  357.           StringBuilder tmp = new StringBuilder("[ ");
  358.           for (ProcessTreeInfo p : trackingContainers.values()) {
  359.             tmp.append(p.getPID());
  360.             tmp.append(" ");
  361.           }
  362.           LOG.debug("Current ProcessTree list : "
  363.               + tmp.substring(0, tmp.length()) + "]");
  364.         }
  365.  
  366.         //用于计算容器的总资源利用率的临时结构
  367.         ResourceUtilization trackedContainersUtilization  =
  368.             ResourceUtilization.newInstance(0, 0, 0.0f);
  369.  
  370.         //现在对trackingContainers进行监视，检查内存使用情况并杀死任何溢出的容器
  371. 		//每个容器在启动时都会将本容器信息放入trackingContainers中，详细看onStartMonitoringContainer()
  372.         long vmemUsageByAllContainers = 0;
  373.         long pmemByAllContainers = 0;
  374.         long cpuUsagePercentPerCoreByAllContainers = 0;
  375.         for (Entry<ContainerId, ProcessTreeInfo> entry : trackingContainers
  376.             .entrySet()) {
  377.           ContainerId containerId = entry.getKey();
  378.           ProcessTreeInfo ptInfo = entry.getValue();
  379.           try {
  380.             //初始化未初始化的进程树
  381.             initializeProcessTrees(entry);
  382.  
  383.             String pId = ptInfo.getPID();
  384.             if (pId == null || !isResourceCalculatorAvailable()) {
  385.               continue; //无法跟踪该 processTree
  386.             }
  387.             if (LOG.isDebugEnabled()) {
  388.               LOG.debug("Constructing ProcessTree for : PID = " + pId
  389.                   + " ContainerId = " + containerId);
  390.             }
  391.             ResourceCalculatorProcessTree pTree = ptInfo.getProcessTree();
  392.             pTree.updateProcessTree();    // 更新 process-tree
  393. 			//获取进程树中所有进程使用的虚拟内存。
  394.             long currentVmemUsage = pTree.getVirtualMemorySize();
  395. 			//获取进程树中所有进程使用的常驻集大小（rss）内存
  396. 			//rss 是 Resident Set Size 的缩写 表示驻留内存大小，是进程当前实际使用物理内存大小（包含共享库占用的内存）
  397.             long currentPmemUsage = pTree.getRssMemorySize();
  398.             if (currentVmemUsage < 0 || currentPmemUsage < 0) {
  399.               // YARN-6862/YARN-5021 If the container just exited or for
  400.               // another reason the physical/virtual memory is UNAVAILABLE (-1)
  401.               // the values shouldn't be aggregated.
  402.               LOG.info("Skipping monitoring container {} because "
  403.                   + "memory usage is not available.", containerId);
  404.               continue;
  405.             }
  406.  
  407.             // if machine has 6 cores and 3 are used,
  408.             // cpuUsagePercentPerCore should be 300%
  409. 			//基于样本之间的平均值，获取进程树中所有进程的CPU使用率，作为与顶部相似的总CPU周期的比率。因此，如果使用四分之二的核心，则返回200.0。
  410. 			//注意：在CPU使用率不可用的情况下，将返回UNAVAILABLE。不建议返回任何其他错误代码。
  411.             float cpuUsagePercentPerCore = pTree.getCpuUsagePercent();
  412.             if (cpuUsagePercentPerCore < 0) {
  413.               // CPU usage is not available likely because the container just
  414.               // started. Let us skip this turn and consider this container
  415.               // in the next iteration.
  416.               LOG.info("Skipping monitoring container " + containerId
  417.                   + " since CPU usage is not yet available.");
  418.               continue;
  419.             }
  420. 			
  421. 			//记录使用情况指标
  422.             recordUsage(containerId, pId, pTree, ptInfo, currentVmemUsage,
  423.                     currentPmemUsage, trackedContainersUtilization);
  424. 			//检查资源限制，如果超出限制，请采取措施
  425.             checkLimit(containerId, pId, pTree, ptInfo,
  426.                     currentVmemUsage, currentPmemUsage);
  427.  
  428.             //计算所有容器的总内存使用情况
  429.             vmemUsageByAllContainers += currentVmemUsage;
  430.             pmemByAllContainers += currentPmemUsage;
  431.             //计算所有容器的总cpu使用量
  432.             cpuUsagePercentPerCoreByAllContainers += cpuUsagePercentPerCore;
  433.  
  434. 			//向时间线服务报告使用情况指标
  435.             reportResourceUsage(containerId, currentPmemUsage,
  436.                     cpuUsagePercentPerCore);
  437.           } catch (Exception e) {
  438.             // Log the exception and proceed to the next container.
  439.             LOG.warn("Uncaught exception in ContainersMonitorImpl "
  440.                 + "while monitoring resource of {}", containerId, e);
  441.           }
  442.         }
  443.         if (LOG.isDebugEnabled()) {
  444.           LOG.debug("Total Resource Usage stats in NM by all containers : "
  445.               + "Virtual Memory= " + vmemUsageByAllContainers
  446.               + ", Physical Memory= " + pmemByAllContainers
  447.               + ", Total CPU usage(% per core)= "
  448.               + cpuUsagePercentPerCoreByAllContainers);
  449.         }
  450.  
  451.         //保存容器的聚合利用率
  452.         setContainersUtilization(trackedContainersUtilization);
  453.  
  454.         //将容器利用率度量发布到节点管理器度量系统
  455.         NodeManagerMetrics nmMetrics = context.getNodeManagerMetrics();
  456.         if (nmMetrics != null) {
  457.           nmMetrics.setContainerUsedMemGB(
  458.               trackedContainersUtilization.getPhysicalMemory());
  459.           nmMetrics.setContainerUsedVMemGB(
  460.               trackedContainersUtilization.getVirtualMemory());
  461.           nmMetrics.setContainerCpuUtilization(
  462.               trackedContainersUtilization.getCPU());
  463.         }
  464.  
  465.         try {
  466. 		  //监视容器的频率 默认3s
  467.           Thread.sleep(monitoringInterval);
  468.         } catch (InterruptedException e) {
  469.           LOG.warn(ContainersMonitorImpl.class.getName()
  470.               + " is interrupted. Exiting.");
  471.           break;
  472.         }
  473.       }
  474.     }
  475.  
  476.    
  477.     private void recordUsage(ContainerId containerId, String pId,
  478.                              ResourceCalculatorProcessTree pTree,
  479.                              ProcessTreeInfo ptInfo,
  480.                              long currentVmemUsage, long currentPmemUsage,
  481.                              ResourceUtilization trackedContainersUtilization) {
  482.       // if machine has 6 cores and 3 are used,
  483.       // cpuUsagePercentPerCore should be 300% and
  484.       // cpuUsageTotalCoresPercentage should be 50%
  485.       float cpuUsagePercentPerCore = pTree.getCpuUsagePercent();
  486.       float cpuUsageTotalCoresPercentage = cpuUsagePercentPerCore /
  487.               resourceCalculatorPlugin.getNumProcessors();
  488.  
  489.       //乘以1000以避免在转换为int时丢失数据
  490. 	  //cpu 核数利用率 * 1000 * 8 / 100 
  491. 	  //比如 0.5 * 1000 * 8 / 100 = 40
  492.       int milliVcoresUsed = (int) (cpuUsageTotalCoresPercentage * 1000
  493.               * maxVCoresAllottedForContainers /nodeCpuPercentageForYARN);
  494. 	  //进程树的虚拟内存限制（字节）
  495.       long vmemLimit = ptInfo.getVmemLimit();
  496. 	  //进程树的物理内存限制（字节）
  497.       long pmemLimit = ptInfo.getPmemLimit();
  498.       if (AUDITLOG.isDebugEnabled()) {
  499.         int vcoreLimit = ptInfo.getCpuVcores();
  500.         long cumulativeCpuTime = pTree.getCumulativeCpuTime();
  501.         AUDITLOG.debug(String.format(
  502.             "Resource usage of ProcessTree %s for container-id %s:" +
  503.                 " %s %%CPU: %f %%CPU-cores: %f" +
  504.                 " vCores-used: %d of %d Cumulative-CPU-ms: %d",
  505.             pId, containerId.toString(),
  506.             formatUsageString(
  507.                 currentVmemUsage, vmemLimit,
  508.                 currentPmemUsage, pmemLimit),
  509.             cpuUsagePercentPerCore,
  510.             cpuUsageTotalCoresPercentage,
  511.             milliVcoresUsed / 1000, vcoreLimit,
  512.             cumulativeCpuTime));
  513.       }
  514.  
  515.       //添加此容器的资源利用率
  516.       trackedContainersUtilization.addTo(
  517.               (int) (currentPmemUsage >> 20),
  518.               (int) (currentVmemUsage >> 20),
  519.               milliVcoresUsed / 1000.0f);
  520.  
  521.       //将使用情况添加到容器指标
  522.       if (containerMetricsEnabled) {
  523.         ContainerMetrics.forContainer(
  524.                 containerId, containerMetricsPeriodMs,
  525.                 containerMetricsUnregisterDelayMs).recordMemoryUsage(
  526.                 (int) (currentPmemUsage >> 20));
  527.         ContainerMetrics.forContainer(
  528.                 containerId, containerMetricsPeriodMs,
  529.                 containerMetricsUnregisterDelayMs).recordCpuUsage((int)
  530.                 cpuUsagePercentPerCore, milliVcoresUsed);
  531.       }
  532.     }
  533.  
  534.     
  535.     private void checkLimit(ContainerId containerId, String pId,
  536.                             ResourceCalculatorProcessTree pTree,
  537.                             ProcessTreeInfo ptInfo,
  538.                             long currentVmemUsage,
  539.                             long currentPmemUsage) {
  540.       Optional<Boolean> isMemoryOverLimit = Optional.empty();
  541.       String msg = "";
  542.       int containerExitStatus = ContainerExitStatus.INVALID;
  543.  
  544. 		//strictMemoryEnforcement 默认 true elasticMemoryEnforcement默认 false
  545. 		//因此不走这个逻辑 elasticMemoryEnforcement 开启 
  546.       if (strictMemoryEnforcement && elasticMemoryEnforcement) {
  547.         //弹性内存控制和严格内存控制都是通过cgroups实现的。如果容器超过其请求，它会被弹性内存控制机制冻结，所以我们在这里检查并杀死它。
  548.         //否则，如果节点从未超过其限制，并且基于procfs的内存核算与基于cgroup的核算不同，则不会杀死容器。
  549.  
  550. 		//默认为 CGroupsMemoryResourceHandlerImpl
  551. 		//处理程序类来处理内存控制器。YARN已经在Java中提供了一个物理内存监视器，但它不如CGroups。
  552. 		//此处理程序设置软内存和硬内存限制。软限制设置为硬限制的90%。
  553.         MemoryResourceHandler handler =
  554.             ResourceHandlerModule.getMemoryResourceHandler();
  555.         if (handler != null) {
  556. 		  //检查容器是否处于OOM状态
  557.           isMemoryOverLimit = handler.isUnderOOM(containerId);
  558.           containerExitStatus = ContainerExitStatus.KILLED_EXCEEDED_PMEM;
  559.           msg = containerId + " is under oom because it exceeded its" +
  560.               " physical memory limit";
  561.         }
  562.       } else if (strictMemoryEnforcement || elasticMemoryEnforcement) {
  563.         //如果启用了基于cgroup的内存控制
  564.         isMemoryOverLimit = Optional.of(false);
  565.       }
  566.  
  567.       if (!isMemoryOverLimit.isPresent()) {
  568.         long vmemLimit = ptInfo.getVmemLimit();
  569.         long pmemLimit = ptInfo.getPmemLimit();
  570.         //当流程从1开始时，我们想看看是否有超过1次迭代的流程。
  571.         long curMemUsageOfAgedProcesses = pTree.getVirtualMemorySize(1);
  572.         long curRssMemUsageOfAgedProcesses = pTree.getRssMemorySize(1);
  573. 		//默认为 true 对容器强制执行虚拟内存限制
  574.         if (isVmemCheckEnabled()
  575.             && isProcessTreeOverLimit(containerId.toString(),
  576.             currentVmemUsage, curMemUsageOfAgedProcesses, vmemLimit)) {
  577.           //当前使用率（年龄=0）始终高于过期使用率。我们不在消息中显示老化的大小，而是根据当前使用情况进行增量
  578.           long delta = currentVmemUsage - vmemLimit;
  579.           // 容器（根进程）仍处于活动状态，内存溢出
  580.           // 转储流程树，然后进行清理
  581.           msg = formatErrorMessage("virtual",
  582.               formatUsageString(currentVmemUsage, vmemLimit,
  583.                   currentPmemUsage, pmemLimit),
  584.               pId, containerId, pTree, delta);
  585.           isMemoryOverLimit = Optional.of(true);
  586.           containerExitStatus = ContainerExitStatus.KILLED_EXCEEDED_VMEM;
  587. 		//默认为 true 对容器强制执行物理内存限制
  588. 		//isProcessTreeOverLimit():
  589. 		//检查容器的进程树的当前内存使用量是否超过限制
  590. 		
  591. 		//当java进程exec是一个程序时，它可能会暂时占据其内存大小的两倍，因为JVM执行fork（）+exec（），在fork时间创建父内存的副本。
  592. 		//如果监视线程在同一个实例中检测到容器树使用的内存，它可能会认为它超出了限制并杀死该树，因为进程本身没有故障。
  593. 		
  594. 		//我们通过采用启发式检查来解决这个问题：如果进程树超过内存限制两倍以上，它将立即被杀死；如果进程树的进程比监控间隔早，
  595. 		//甚至超过内存限制1倍，它将被杀死。否则，它会被赋予怀疑的标志，可以再进行一次迭代。
  596.         } else if (isPmemCheckEnabled()
  597.             && isProcessTreeOverLimit(containerId.toString(),
  598.             currentPmemUsage, curRssMemUsageOfAgedProcesses,
  599.             pmemLimit)) {
  600.           //当前使用率（年龄=0）始终高于过期使用率。我们不在消息中显示老化的大小，而是根据当前使用情况进行增量
  601.           long delta = currentPmemUsage - pmemLimit;
  602.           //容器（根进程）仍处于活动状态，内存溢出
  603.           //转储流程树，然后进行清理
  604.           msg = formatErrorMessage("physical",
  605.               formatUsageString(currentVmemUsage, vmemLimit,
  606.                   currentPmemUsage, pmemLimit),
  607.               pId, containerId, pTree, delta);
  608.           isMemoryOverLimit = Optional.of(true);
  609.           containerExitStatus = ContainerExitStatus.KILLED_EXCEEDED_PMEM;
  610.         }
  611.       }
  612.  
  613.       if (isMemoryOverLimit.isPresent() && isMemoryOverLimit.get()
  614.           && trackingContainers.remove(containerId) != null) {
  615.         //虚拟内存或物理内存超出限制。使容器失败并删除相应的流程树
  616.         LOG.warn(msg);
  617.         //警告（如果不是领导者）
  618.         if (!pTree.checkPidPgrpidForMatch()) {
  619.           LOG.error("Killed container process with PID " + pId
  620.                   + " but it is not a process group leader.");
  621.         }
  622.         //杀掉容器
  623.         eventDispatcher.getEventHandler().handle(
  624.                 new ContainerKillEvent(containerId,
  625.                       containerExitStatus, msg));
  626.         LOG.info("Removed ProcessTree with root " + pId);
  627.       }
  628.     }
  629.  
  630.     
  631.  
  632.   private void onStopMonitoringContainer(
  633.       ContainersMonitorEvent monitoringEvent, ContainerId containerId) {
  634.     LOG.info("Stopping resource-monitoring for " + containerId);
  635.     updateContainerMetrics(monitoringEvent);
  636.     trackingContainers.remove(containerId);
  637.   }
  638.  
  639.   private void onStartMonitoringContainer(
  640.       ContainersMonitorEvent monitoringEvent, ContainerId containerId) {
  641.     ContainerStartMonitoringEvent startEvent =
  642.         (ContainerStartMonitoringEvent) monitoringEvent;
  643.     LOG.info("Starting resource-monitoring for " + containerId);
  644.     updateContainerMetrics(monitoringEvent);
  645.     trackingContainers.put(containerId,
  646.         new ProcessTreeInfo(containerId, null, null,
  647.             startEvent.getVmemLimit(), startEvent.getPmemLimit(),
  648.             startEvent.getCpuVcores()));
  649.   }
  650. }

四、总结

1、启动容器触发ContainerEventType.CONTAINER_LAUNCHED事件

2、ContainerImpl会处理1中事件，启动容器的同时触发容器监控事件ContainersMonitorEventType.START_MONITORING_CONTAINER

3、该事件由ContainersMonitorImpl调用onStartMonitoringContainer()处理2中事件

4、将启动的容器id、虚拟内存限制、物理内存限制、cpu核数限制封装成ProcessTreeInfo，并放到跟踪所有容器的trackingContainers中

5、ContainersMonitorImpl初始化时会获取监控容器的频率（默认3s一次）、监控容器日志目录大小频率（默认1min一次）、容器磁盘大小限制（默认1G）、全部容器总磁盘大小限制（默认10G）、系统资源计算插件（可以自己实现，默认LINUX 使用SysInfoLinux，WINDOWS 使用SysInfoWindows）、计算processTree资源利用率的类、系统为YARN容器留内存大小、YARN容器可用vcore数、虚拟内存和物理内存比率、内存控制策略等

6、ContainersMonitorImpl启动时会启动一个线程（monitoringThread）对容器的资源使用进行监控，如果超过限制就杀掉容器。默认只开启这一个线程，oomListenerThread和logMonitorThread默认不开启

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/181919
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号