| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.service.AbstractService; |
| import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix; |
| import org.apache.hadoop.yarn.api.records.ContainerExitStatus; |
| import org.apache.hadoop.yarn.api.records.ContainerId; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.ResourceUtilization; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.event.AsyncDispatcher; |
| import org.apache.hadoop.yarn.event.Dispatcher; |
| import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor; |
| import org.apache.hadoop.yarn.server.nodemanager.Context; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerImpl; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerKillEvent; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerExecutionException; |
| import org.apache.hadoop.yarn.server.nodemanager.timelineservice.NMTimelinePublisher; |
| import org.apache.hadoop.yarn.server.nodemanager.util.NodeManagerHardwareUtils; |
| import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin; |
| import org.apache.hadoop.yarn.util.ResourceCalculatorProcessTree; |
| |
| import java.util.Arrays; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.concurrent.ConcurrentHashMap; |
| |
| /** |
| * Monitors containers collecting resource usage and preempting the container |
| * if it exceeds its limits. |
| */ |
| public class ContainersMonitorImpl extends AbstractService implements |
| ContainersMonitor { |
| |
| private final static Logger LOG = |
| LoggerFactory.getLogger(ContainersMonitorImpl.class); |
| |
| private long monitoringInterval; |
| private MonitoringThread monitoringThread; |
| private boolean containerMetricsEnabled; |
| private long containerMetricsPeriodMs; |
| private long containerMetricsUnregisterDelayMs; |
| |
| @VisibleForTesting |
| final Map<ContainerId, ProcessTreeInfo> trackingContainers = |
| new ConcurrentHashMap<>(); |
| |
| private final ContainerExecutor containerExecutor; |
| private final Dispatcher eventDispatcher; |
| private final Context context; |
| private ResourceCalculatorPlugin resourceCalculatorPlugin; |
| private Configuration conf; |
| private static float vmemRatio; |
| private Class<? extends ResourceCalculatorProcessTree> processTreeClass; |
| |
| private long maxVmemAllottedForContainers = UNKNOWN_MEMORY_LIMIT; |
| private long maxPmemAllottedForContainers = UNKNOWN_MEMORY_LIMIT; |
| |
| private boolean pmemCheckEnabled; |
| private boolean vmemCheckEnabled; |
| private boolean containersMonitorEnabled; |
| |
| private long maxVCoresAllottedForContainers; |
| |
| private static final long UNKNOWN_MEMORY_LIMIT = -1L; |
| private int nodeCpuPercentageForYARN; |
| |
| /** |
| * Type of container metric. |
| */ |
| @Private |
| public enum ContainerMetric { |
| CPU, MEMORY |
| } |
| |
| private ResourceUtilization containersUtilization; |
| |
| private volatile boolean stopped = false; |
| |
| public ContainersMonitorImpl(ContainerExecutor exec, |
| AsyncDispatcher dispatcher, Context context) { |
| super("containers-monitor"); |
| |
| this.containerExecutor = exec; |
| this.eventDispatcher = dispatcher; |
| this.context = context; |
| |
| this.monitoringThread = new MonitoringThread(); |
| |
| this.containersUtilization = ResourceUtilization.newInstance(0, 0, 0.0f); |
| } |
| |
| @Override |
| protected void serviceInit(Configuration myConf) throws Exception { |
| this.conf = myConf; |
| this.monitoringInterval = |
| this.conf.getLong(YarnConfiguration.NM_CONTAINER_MON_INTERVAL_MS, |
| this.conf.getLong(YarnConfiguration.NM_RESOURCE_MON_INTERVAL_MS, |
| YarnConfiguration.DEFAULT_NM_RESOURCE_MON_INTERVAL_MS)); |
| |
| this.resourceCalculatorPlugin = |
| ResourceCalculatorPlugin.getContainersMonitorPlugin(this.conf); |
| LOG.info(" Using ResourceCalculatorPlugin : " |
| + this.resourceCalculatorPlugin); |
| processTreeClass = this.conf.getClass( |
| YarnConfiguration.NM_CONTAINER_MON_PROCESS_TREE, null, |
| ResourceCalculatorProcessTree.class); |
| LOG.info(" Using ResourceCalculatorProcessTree : " |
| + this.processTreeClass); |
| |
| this.containerMetricsEnabled = |
| this.conf.getBoolean(YarnConfiguration.NM_CONTAINER_METRICS_ENABLE, |
| YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_ENABLE); |
| this.containerMetricsPeriodMs = |
| this.conf.getLong(YarnConfiguration.NM_CONTAINER_METRICS_PERIOD_MS, |
| YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_PERIOD_MS); |
| this.containerMetricsUnregisterDelayMs = this.conf.getLong( |
| YarnConfiguration.NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS, |
| YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS); |
| |
| long configuredPMemForContainers = |
| NodeManagerHardwareUtils.getContainerMemoryMB( |
| this.resourceCalculatorPlugin, this.conf) * 1024 * 1024L; |
| |
| long configuredVCoresForContainers = |
| NodeManagerHardwareUtils.getVCores(this.resourceCalculatorPlugin, |
| this.conf); |
| |
| // Setting these irrespective of whether checks are enabled. Required in |
| // the UI. |
| // ///////// Physical memory configuration ////// |
| this.maxPmemAllottedForContainers = configuredPMemForContainers; |
| this.maxVCoresAllottedForContainers = configuredVCoresForContainers; |
| |
| // ///////// Virtual memory configuration ////// |
| vmemRatio = this.conf.getFloat(YarnConfiguration.NM_VMEM_PMEM_RATIO, |
| YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO); |
| Preconditions.checkArgument(vmemRatio > 0.99f, |
| YarnConfiguration.NM_VMEM_PMEM_RATIO + " should be at least 1.0"); |
| this.maxVmemAllottedForContainers = |
| (long) (vmemRatio * configuredPMemForContainers); |
| |
| pmemCheckEnabled = this.conf.getBoolean( |
| YarnConfiguration.NM_PMEM_CHECK_ENABLED, |
| YarnConfiguration.DEFAULT_NM_PMEM_CHECK_ENABLED); |
| vmemCheckEnabled = this.conf.getBoolean( |
| YarnConfiguration.NM_VMEM_CHECK_ENABLED, |
| YarnConfiguration.DEFAULT_NM_VMEM_CHECK_ENABLED); |
| LOG.info("Physical memory check enabled: " + pmemCheckEnabled); |
| LOG.info("Virtual memory check enabled: " + vmemCheckEnabled); |
| |
| containersMonitorEnabled = |
| isContainerMonitorEnabled() && monitoringInterval > 0; |
| LOG.info("ContainersMonitor enabled: " + containersMonitorEnabled); |
| |
| nodeCpuPercentageForYARN = |
| NodeManagerHardwareUtils.getNodeCpuPercentage(this.conf); |
| |
| if (pmemCheckEnabled) { |
| // Logging if actual pmem cannot be determined. |
| long totalPhysicalMemoryOnNM = UNKNOWN_MEMORY_LIMIT; |
| if (this.resourceCalculatorPlugin != null) { |
| totalPhysicalMemoryOnNM = this.resourceCalculatorPlugin |
| .getPhysicalMemorySize(); |
| if (totalPhysicalMemoryOnNM <= 0) { |
| LOG.warn("NodeManager's totalPmem could not be calculated. " |
| + "Setting it to " + UNKNOWN_MEMORY_LIMIT); |
| totalPhysicalMemoryOnNM = UNKNOWN_MEMORY_LIMIT; |
| } |
| } |
| |
| if (totalPhysicalMemoryOnNM != UNKNOWN_MEMORY_LIMIT && |
| this.maxPmemAllottedForContainers > totalPhysicalMemoryOnNM * 0.80f) { |
| LOG.warn("NodeManager configured with " |
| + TraditionalBinaryPrefix.long2String(maxPmemAllottedForContainers, |
| "", 1) |
| + " physical memory allocated to containers, which is more than " |
| + "80% of the total physical memory available (" |
| + TraditionalBinaryPrefix.long2String(totalPhysicalMemoryOnNM, "", |
| 1) + "). Thrashing might happen."); |
| } |
| } |
| super.serviceInit(this.conf); |
| } |
| |
| private boolean isContainerMonitorEnabled() { |
| return conf.getBoolean(YarnConfiguration.NM_CONTAINER_MONITOR_ENABLED, |
| YarnConfiguration.DEFAULT_NM_CONTAINER_MONITOR_ENABLED); |
| } |
| |
| private boolean isResourceCalculatorAvailable() { |
| if (resourceCalculatorPlugin == null) { |
| LOG.info("ResourceCalculatorPlugin is unavailable on this system. " + this |
| .getClass().getName() + " is disabled."); |
| return false; |
| } |
| if (ResourceCalculatorProcessTree |
| .getResourceCalculatorProcessTree("0", processTreeClass, conf) |
| == null) { |
| LOG.info("ResourceCalculatorProcessTree is unavailable on this system. " |
| + this.getClass().getName() + " is disabled."); |
| return false; |
| } |
| return true; |
| } |
| |
| @Override |
| protected void serviceStart() throws Exception { |
| if (containersMonitorEnabled) { |
| this.monitoringThread.start(); |
| } |
| super.serviceStart(); |
| } |
| |
| @Override |
| protected void serviceStop() throws Exception { |
| if (containersMonitorEnabled) { |
| stopped = true; |
| this.monitoringThread.interrupt(); |
| try { |
| this.monitoringThread.join(); |
| } catch (InterruptedException e) { |
| LOG.info("ContainersMonitorImpl monitoring thread interrupted"); |
| } |
| } |
| super.serviceStop(); |
| } |
| |
| /** |
| * Encapsulates resource requirements of a process and its tree. |
| */ |
| public static class ProcessTreeInfo { |
| private ContainerId containerId; |
| private String pid; |
| private ResourceCalculatorProcessTree pTree; |
| private long vmemLimit; |
| private long pmemLimit; |
| private int cpuVcores; |
| |
| public ProcessTreeInfo(ContainerId containerId, String pid, |
| ResourceCalculatorProcessTree pTree, long vmemLimit, long pmemLimit, |
| int cpuVcores) { |
| this.containerId = containerId; |
| this.pid = pid; |
| this.pTree = pTree; |
| this.vmemLimit = vmemLimit; |
| this.pmemLimit = pmemLimit; |
| this.cpuVcores = cpuVcores; |
| } |
| |
| public ContainerId getContainerId() { |
| return this.containerId; |
| } |
| |
| public String getPID() { |
| return this.pid; |
| } |
| |
| public void setPid(String pid) { |
| this.pid = pid; |
| } |
| |
| ResourceCalculatorProcessTree getProcessTree() { |
| return this.pTree; |
| } |
| |
| void setProcessTree(ResourceCalculatorProcessTree mypTree) { |
| this.pTree = mypTree; |
| } |
| |
| /** |
| * @return Virtual memory limit for the process tree in bytes |
| */ |
| public synchronized long getVmemLimit() { |
| return this.vmemLimit; |
| } |
| |
| /** |
| * @return Physical memory limit for the process tree in bytes |
| */ |
| public synchronized long getPmemLimit() { |
| return this.pmemLimit; |
| } |
| |
| /** |
| * @return Number of cpu vcores assigned |
| */ |
| public synchronized int getCpuVcores() { |
| return this.cpuVcores; |
| } |
| |
| /** |
| * Set resource limit for enforcement. |
| * @param myPmemLimit |
| * Physical memory limit for the process tree in bytes |
| * @param myVmemLimit |
| * Virtual memory limit for the process tree in bytes |
| * @param myCpuVcores |
| * Number of cpu vcores assigned |
| */ |
| synchronized void setResourceLimit( |
| long myPmemLimit, long myVmemLimit, int myCpuVcores) { |
| this.pmemLimit = myPmemLimit; |
| this.vmemLimit = myVmemLimit; |
| this.cpuVcores = myCpuVcores; |
| } |
| } |
| |
| /** |
| * Check whether a container's process tree's current memory usage is over |
| * limit. |
| * |
| * When a java process exec's a program, it could momentarily account for |
| * double the size of it's memory, because the JVM does a fork()+exec() |
| * which at fork time creates a copy of the parent's memory. If the |
| * monitoring thread detects the memory used by the container tree at the |
| * same instance, it could assume it is over limit and kill the tree, for no |
| * fault of the process itself. |
| * |
| * We counter this problem by employing a heuristic check: - if a process |
| * tree exceeds the memory limit by more than twice, it is killed |
| * immediately - if a process tree has processes older than the monitoring |
| * interval exceeding the memory limit by even 1 time, it is killed. Else it |
| * is given the benefit of doubt to lie around for one more iteration. |
| * |
| * @param containerId |
| * Container Id for the container tree |
| * @param currentMemUsage |
| * Memory usage of a container tree |
| * @param curMemUsageOfAgedProcesses |
| * Memory usage of processes older than an iteration in a container |
| * tree |
| * @param vmemLimit |
| * The limit specified for the container |
| * @return true if the memory usage is more than twice the specified limit, |
| * or if processes in the tree, older than this thread's monitoring |
| * interval, exceed the memory limit. False, otherwise. |
| */ |
| private boolean isProcessTreeOverLimit(String containerId, |
| long currentMemUsage, |
| long curMemUsageOfAgedProcesses, |
| long vmemLimit) { |
| boolean isOverLimit = false; |
| |
| if (currentMemUsage > (2 * vmemLimit)) { |
| LOG.warn("Process tree for container: " + containerId |
| + " running over twice " + "the configured limit. Limit=" + vmemLimit |
| + ", current usage = " + currentMemUsage); |
| isOverLimit = true; |
| } else if (curMemUsageOfAgedProcesses > vmemLimit) { |
| LOG.warn("Process tree for container: " + containerId |
| + " has processes older than 1 " |
| + "iteration running over the configured limit. Limit=" + vmemLimit |
| + ", current usage = " + curMemUsageOfAgedProcesses); |
| isOverLimit = true; |
| } |
| |
| return isOverLimit; |
| } |
| |
| // method provided just for easy testing purposes |
| boolean isProcessTreeOverLimit(ResourceCalculatorProcessTree pTree, |
| String containerId, long limit) { |
| long currentMemUsage = pTree.getVirtualMemorySize(); |
| // as processes begin with an age 1, we want to see if there are processes |
| // more than 1 iteration old. |
| long curMemUsageOfAgedProcesses = pTree.getVirtualMemorySize(1); |
| return isProcessTreeOverLimit(containerId, currentMemUsage, |
| curMemUsageOfAgedProcesses, limit); |
| } |
| |
| private class MonitoringThread extends Thread { |
| MonitoringThread() { |
| super("Container Monitor"); |
| } |
| |
| @Override |
| public void run() { |
| |
| while (!stopped && !Thread.currentThread().isInterrupted()) { |
| // Print the processTrees for debugging. |
| if (LOG.isDebugEnabled()) { |
| StringBuilder tmp = new StringBuilder("[ "); |
| for (ProcessTreeInfo p : trackingContainers.values()) { |
| tmp.append(p.getPID()); |
| tmp.append(" "); |
| } |
| LOG.debug("Current ProcessTree list : " |
| + tmp.substring(0, tmp.length()) + "]"); |
| } |
| |
| // Temporary structure to calculate the total resource utilization of |
| // the containers |
| ResourceUtilization trackedContainersUtilization = |
| ResourceUtilization.newInstance(0, 0, 0.0f); |
| |
| // Now do the monitoring for the trackingContainers |
| // Check memory usage and kill any overflowing containers |
| long vmemUsageByAllContainers = 0; |
| long pmemByAllContainers = 0; |
| long cpuUsagePercentPerCoreByAllContainers = 0; |
| for (Entry<ContainerId, ProcessTreeInfo> entry : trackingContainers |
| .entrySet()) { |
| ContainerId containerId = entry.getKey(); |
| ProcessTreeInfo ptInfo = entry.getValue(); |
| try { |
| // Initialize uninitialized process trees |
| initializeProcessTrees(entry); |
| |
| String pId = ptInfo.getPID(); |
| if (pId == null || !isResourceCalculatorAvailable()) { |
| continue; // processTree cannot be tracked |
| } |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Constructing ProcessTree for : PID = " + pId |
| + " ContainerId = " + containerId); |
| } |
| ResourceCalculatorProcessTree pTree = ptInfo.getProcessTree(); |
| pTree.updateProcessTree(); // update process-tree |
| long currentVmemUsage = pTree.getVirtualMemorySize(); |
| long currentPmemUsage = pTree.getRssMemorySize(); |
| |
| // if machine has 6 cores and 3 are used, |
| // cpuUsagePercentPerCore should be 300% |
| float cpuUsagePercentPerCore = pTree.getCpuUsagePercent(); |
| if (cpuUsagePercentPerCore < 0) { |
| // CPU usage is not available likely because the container just |
| // started. Let us skip this turn and consider this container |
| // in the next iteration. |
| LOG.info("Skipping monitoring container " + containerId |
| + " since CPU usage is not yet available."); |
| continue; |
| } |
| |
| recordUsage(containerId, pId, pTree, ptInfo, currentVmemUsage, |
| currentPmemUsage, trackedContainersUtilization); |
| |
| checkLimit(containerId, pId, pTree, ptInfo, |
| currentVmemUsage, currentPmemUsage); |
| |
| // Accounting the total memory in usage for all containers |
| vmemUsageByAllContainers += currentVmemUsage; |
| pmemByAllContainers += currentPmemUsage; |
| // Accounting the total cpu usage for all containers |
| cpuUsagePercentPerCoreByAllContainers += cpuUsagePercentPerCore; |
| |
| reportResourceUsage(containerId, currentPmemUsage, |
| cpuUsagePercentPerCore); |
| } catch (Exception e) { |
| // Log the exception and proceed to the next container. |
| LOG.warn("Uncaught exception in ContainersMonitorImpl " |
| + "while monitoring resource of " + containerId, e); |
| } |
| } |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Total Resource Usage stats in NM by all containers : " |
| + "Virtual Memory= " + vmemUsageByAllContainers |
| + ", Physical Memory= " + pmemByAllContainers |
| + ", Total CPU usage(% per core)= " |
| + cpuUsagePercentPerCoreByAllContainers); |
| } |
| |
| // Save the aggregated utilization of the containers |
| setContainersUtilization(trackedContainersUtilization); |
| |
| try { |
| Thread.sleep(monitoringInterval); |
| } catch (InterruptedException e) { |
| LOG.warn(ContainersMonitorImpl.class.getName() |
| + " is interrupted. Exiting."); |
| break; |
| } |
| } |
| } |
| |
| /** |
| * Initialize any uninitialized processTrees. |
| * @param entry process tree entry to fill in |
| */ |
| private void initializeProcessTrees( |
| Entry<ContainerId, ProcessTreeInfo> entry) |
| throws ContainerExecutionException { |
| ContainerId containerId = entry.getKey(); |
| ProcessTreeInfo ptInfo = entry.getValue(); |
| String pId = ptInfo.getPID(); |
| |
| // Initialize any uninitialized processTrees |
| if (pId == null) { |
| // get pid from ContainerId |
| pId = containerExecutor.getProcessId(ptInfo.getContainerId()); |
| if (pId != null) { |
| // pId will be null, either if the container is not spawned yet |
| // or if the container's pid is removed from ContainerExecutor |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Tracking ProcessTree " + pId + " for the first time"); |
| } |
| ResourceCalculatorProcessTree pt = |
| ResourceCalculatorProcessTree. |
| getResourceCalculatorProcessTree( |
| pId, processTreeClass, conf); |
| ptInfo.setPid(pId); |
| ptInfo.setProcessTree(pt); |
| |
| if (containerMetricsEnabled) { |
| ContainerMetrics usageMetrics = ContainerMetrics |
| .forContainer(containerId, containerMetricsPeriodMs, |
| containerMetricsUnregisterDelayMs); |
| usageMetrics.recordProcessId(pId); |
| } |
| |
| Container container = context.getContainers().get(containerId); |
| |
| if (container != null) { |
| String[] ipAndHost = containerExecutor.getIpAndHost(container); |
| |
| if ((ipAndHost != null) && (ipAndHost[0] != null) && |
| (ipAndHost[1] != null)) { |
| container.setIpAndHost(ipAndHost); |
| LOG.info(containerId + "'s ip = " + ipAndHost[0] |
| + ", and hostname = " + ipAndHost[1]); |
| } else { |
| LOG.info("Can not get both ip and hostname: " |
| + Arrays.toString(ipAndHost)); |
| } |
| } else { |
| LOG.info(containerId + " is missing. Not setting ip and hostname"); |
| } |
| } |
| } |
| // End of initializing any uninitialized processTrees |
| } |
| |
| /** |
| * Record usage metrics. |
| * @param containerId container id |
| * @param pId process id |
| * @param pTree valid process tree entry with CPU measurement |
| * @param ptInfo process tree info with limit information |
| * @param currentVmemUsage virtual memory measurement |
| * @param currentPmemUsage physical memory measurement |
| * @param trackedContainersUtilization utilization tracker to update |
| */ |
| private void recordUsage(ContainerId containerId, String pId, |
| ResourceCalculatorProcessTree pTree, |
| ProcessTreeInfo ptInfo, |
| long currentVmemUsage, long currentPmemUsage, |
| ResourceUtilization trackedContainersUtilization) { |
| // if machine has 6 cores and 3 are used, |
| // cpuUsagePercentPerCore should be 300% and |
| // cpuUsageTotalCoresPercentage should be 50% |
| float cpuUsagePercentPerCore = pTree.getCpuUsagePercent(); |
| float cpuUsageTotalCoresPercentage = cpuUsagePercentPerCore / |
| resourceCalculatorPlugin.getNumProcessors(); |
| |
| // Multiply by 1000 to avoid losing data when converting to int |
| int milliVcoresUsed = (int) (cpuUsageTotalCoresPercentage * 1000 |
| * maxVCoresAllottedForContainers /nodeCpuPercentageForYARN); |
| long vmemLimit = ptInfo.getVmemLimit(); |
| long pmemLimit = ptInfo.getPmemLimit(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(String.format( |
| "Memory usage of ProcessTree %s for container-id %s: ", |
| pId, containerId.toString()) + |
| formatUsageString( |
| currentVmemUsage, vmemLimit, |
| currentPmemUsage, pmemLimit)); |
| } |
| |
| // Add resource utilization for this container |
| trackedContainersUtilization.addTo( |
| (int) (currentPmemUsage >> 20), |
| (int) (currentVmemUsage >> 20), |
| milliVcoresUsed / 1000.0f); |
| |
| // Add usage to container metrics |
| if (containerMetricsEnabled) { |
| ContainerMetrics.forContainer( |
| containerId, containerMetricsPeriodMs, |
| containerMetricsUnregisterDelayMs).recordMemoryUsage( |
| (int) (currentPmemUsage >> 20)); |
| ContainerMetrics.forContainer( |
| containerId, containerMetricsPeriodMs, |
| containerMetricsUnregisterDelayMs).recordCpuUsage((int) |
| cpuUsagePercentPerCore, milliVcoresUsed); |
| } |
| } |
| |
| /** |
| * Check resource limits and take actions if the limit is exceeded. |
| * @param containerId container id |
| * @param pId process id |
| * @param pTree valid process tree entry with CPU measurement |
| * @param ptInfo process tree info with limit information |
| * @param currentVmemUsage virtual memory measurement |
| * @param currentPmemUsage physical memory measurement |
| */ |
| @SuppressWarnings("unchecked") |
| private void checkLimit(ContainerId containerId, String pId, |
| ResourceCalculatorProcessTree pTree, |
| ProcessTreeInfo ptInfo, |
| long currentVmemUsage, |
| long currentPmemUsage) { |
| boolean isMemoryOverLimit = false; |
| long vmemLimit = ptInfo.getVmemLimit(); |
| long pmemLimit = ptInfo.getPmemLimit(); |
| // as processes begin with an age 1, we want to see if there |
| // are processes more than 1 iteration old. |
| long curMemUsageOfAgedProcesses = pTree.getVirtualMemorySize(1); |
| long curRssMemUsageOfAgedProcesses = pTree.getRssMemorySize(1); |
| String msg = ""; |
| int containerExitStatus = ContainerExitStatus.INVALID; |
| if (isVmemCheckEnabled() |
| && isProcessTreeOverLimit(containerId.toString(), |
| currentVmemUsage, curMemUsageOfAgedProcesses, vmemLimit)) { |
| // Container (the root process) is still alive and overflowing |
| // memory. |
| // Dump the process-tree and then clean it up. |
| msg = formatErrorMessage("virtual", |
| formatUsageString(currentVmemUsage, vmemLimit, |
| currentPmemUsage, pmemLimit), |
| pId, containerId, pTree); |
| isMemoryOverLimit = true; |
| containerExitStatus = ContainerExitStatus.KILLED_EXCEEDED_VMEM; |
| } else if (isPmemCheckEnabled() |
| && isProcessTreeOverLimit(containerId.toString(), |
| currentPmemUsage, curRssMemUsageOfAgedProcesses, |
| pmemLimit)) { |
| // Container (the root process) is still alive and overflowing |
| // memory. |
| // Dump the process-tree and then clean it up. |
| msg = formatErrorMessage("physical", |
| formatUsageString(currentVmemUsage, vmemLimit, |
| currentPmemUsage, pmemLimit), |
| pId, containerId, pTree); |
| isMemoryOverLimit = true; |
| containerExitStatus = ContainerExitStatus.KILLED_EXCEEDED_PMEM; |
| } |
| |
| if (isMemoryOverLimit) { |
| // Virtual or physical memory over limit. Fail the container and |
| // remove |
| // the corresponding process tree |
| LOG.warn(msg); |
| // warn if not a leader |
| if (!pTree.checkPidPgrpidForMatch()) { |
| LOG.error("Killed container process with PID " + pId |
| + " but it is not a process group leader."); |
| } |
| // kill the container |
| eventDispatcher.getEventHandler().handle( |
| new ContainerKillEvent(containerId, |
| containerExitStatus, msg)); |
| trackingContainers.remove(containerId); |
| LOG.info("Removed ProcessTree with root " + pId); |
| } |
| } |
| |
| /** |
| * Report usage metrics to the timeline service. |
| * @param containerId container id |
| * @param currentPmemUsage physical memory measurement |
| * @param cpuUsagePercentPerCore CPU usage |
| */ |
| private void reportResourceUsage(ContainerId containerId, |
| long currentPmemUsage, float cpuUsagePercentPerCore) { |
| ContainerImpl container = |
| (ContainerImpl) context.getContainers().get(containerId); |
| if (container != null) { |
| NMTimelinePublisher nmMetricsPublisher = |
| container.getNMTimelinePublisher(); |
| if (nmMetricsPublisher != null) { |
| nmMetricsPublisher.reportContainerResourceUsage(container, |
| currentPmemUsage, cpuUsagePercentPerCore); |
| } |
| } else { |
| LOG.info(containerId + " does not exist to report"); |
| } |
| } |
| |
| /** |
| * Format string when memory limit has been exceeded. |
| * @param memTypeExceeded type of memory |
| * @param usageString general memory usage information string |
| * @param pId process id |
| * @param containerId container id |
| * @param pTree process tree to dump full resource utilization graph |
| * @return formatted resource usage information |
| */ |
| private String formatErrorMessage(String memTypeExceeded, |
| String usageString, String pId, ContainerId containerId, |
| ResourceCalculatorProcessTree pTree) { |
| return |
| String.format("Container [pid=%s,containerID=%s] is " + |
| "running beyond %s memory limits. ", |
| pId, containerId, memTypeExceeded) + |
| "Current usage: " + usageString + |
| ". Killing container.\n" + |
| "Dump of the process-tree for " + containerId + " :\n" + |
| pTree.getProcessTreeDump(); |
| } |
| |
| /** |
| * Format memory usage string for reporting. |
| * @param currentVmemUsage virtual memory usage |
| * @param vmemLimit virtual memory limit |
| * @param currentPmemUsage physical memory usage |
| * @param pmemLimit physical memory limit |
| * @return formatted memory information |
| */ |
| private String formatUsageString(long currentVmemUsage, long vmemLimit, |
| long currentPmemUsage, long pmemLimit) { |
| return String.format("%sB of %sB physical memory used; " + |
| "%sB of %sB virtual memory used", |
| TraditionalBinaryPrefix.long2String(currentPmemUsage, "", 1), |
| TraditionalBinaryPrefix.long2String(pmemLimit, "", 1), |
| TraditionalBinaryPrefix.long2String(currentVmemUsage, "", 1), |
| TraditionalBinaryPrefix.long2String(vmemLimit, "", 1)); |
| } |
| } |
| |
| private void updateContainerMetrics(ContainersMonitorEvent monitoringEvent) { |
| if (!containerMetricsEnabled || monitoringEvent == null) { |
| return; |
| } |
| |
| ContainerId containerId = monitoringEvent.getContainerId(); |
| ContainerMetrics usageMetrics; |
| |
| int vmemLimitMBs; |
| int pmemLimitMBs; |
| int cpuVcores; |
| switch (monitoringEvent.getType()) { |
| case START_MONITORING_CONTAINER: |
| usageMetrics = ContainerMetrics |
| .forContainer(containerId, containerMetricsPeriodMs, |
| containerMetricsUnregisterDelayMs); |
| ContainerStartMonitoringEvent startEvent = |
| (ContainerStartMonitoringEvent) monitoringEvent; |
| usageMetrics.recordStateChangeDurations( |
| startEvent.getLaunchDuration(), |
| startEvent.getLocalizationDuration()); |
| cpuVcores = startEvent.getCpuVcores(); |
| vmemLimitMBs = (int) (startEvent.getVmemLimit() >> 20); |
| pmemLimitMBs = (int) (startEvent.getPmemLimit() >> 20); |
| usageMetrics.recordResourceLimit( |
| vmemLimitMBs, pmemLimitMBs, cpuVcores); |
| break; |
| case STOP_MONITORING_CONTAINER: |
| usageMetrics = ContainerMetrics.getContainerMetrics( |
| containerId); |
| if (usageMetrics != null) { |
| usageMetrics.finished(); |
| } |
| break; |
| case CHANGE_MONITORING_CONTAINER_RESOURCE: |
| usageMetrics = ContainerMetrics |
| .forContainer(containerId, containerMetricsPeriodMs, |
| containerMetricsUnregisterDelayMs); |
| ChangeMonitoringContainerResourceEvent changeEvent = |
| (ChangeMonitoringContainerResourceEvent) monitoringEvent; |
| Resource resource = changeEvent.getResource(); |
| pmemLimitMBs = (int) resource.getMemorySize(); |
| vmemLimitMBs = (int) (pmemLimitMBs * vmemRatio); |
| cpuVcores = resource.getVirtualCores(); |
| usageMetrics.recordResourceLimit( |
| vmemLimitMBs, pmemLimitMBs, cpuVcores); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| @Override |
| public long getVmemAllocatedForContainers() { |
| return this.maxVmemAllottedForContainers; |
| } |
| |
| /** |
| * Is the total physical memory check enabled? |
| * |
| * @return true if total physical memory check is enabled. |
| */ |
| @Override |
| public boolean isPmemCheckEnabled() { |
| return this.pmemCheckEnabled; |
| } |
| |
| @Override |
| public long getPmemAllocatedForContainers() { |
| return this.maxPmemAllottedForContainers; |
| } |
| |
| @Override |
| public long getVCoresAllocatedForContainers() { |
| return this.maxVCoresAllottedForContainers; |
| } |
| |
| /** |
| * Is the total virtual memory check enabled? |
| * |
| * @return true if total virtual memory check is enabled. |
| */ |
| @Override |
| public boolean isVmemCheckEnabled() { |
| return this.vmemCheckEnabled; |
| } |
| |
| @Override |
| public ResourceUtilization getContainersUtilization() { |
| return this.containersUtilization; |
| } |
| |
| private void setContainersUtilization(ResourceUtilization utilization) { |
| this.containersUtilization = utilization; |
| } |
| |
| @Override |
| public void subtractNodeResourcesFromResourceUtilization( |
| ResourceUtilization resourceUtil) { |
| resourceUtil.subtractFrom((int) (getPmemAllocatedForContainers() >> 20), |
| (int) (getVmemAllocatedForContainers() >> 20), 1.0f); |
| } |
| |
| @Override |
| public float getVmemRatio() { |
| return vmemRatio; |
| } |
| |
| @Override |
| @SuppressWarnings("unchecked") |
| public void handle(ContainersMonitorEvent monitoringEvent) { |
| ContainerId containerId = monitoringEvent.getContainerId(); |
| |
| switch (monitoringEvent.getType()) { |
| case START_MONITORING_CONTAINER: |
| onStartMonitoringContainer(monitoringEvent, containerId); |
| break; |
| case STOP_MONITORING_CONTAINER: |
| onStopMonitoringContainer(monitoringEvent, containerId); |
| break; |
| case CHANGE_MONITORING_CONTAINER_RESOURCE: |
| onChangeMonitoringContainerResource(monitoringEvent, containerId); |
| break; |
| default: |
| // TODO: Wrong event. |
| } |
| } |
| |
| private void onChangeMonitoringContainerResource( |
| ContainersMonitorEvent monitoringEvent, ContainerId containerId) { |
| ChangeMonitoringContainerResourceEvent changeEvent = |
| (ChangeMonitoringContainerResourceEvent) monitoringEvent; |
| if (containersMonitorEnabled) { |
| ProcessTreeInfo processTreeInfo = trackingContainers.get(containerId); |
| if (processTreeInfo == null) { |
| LOG.warn("Failed to track container " |
| + containerId.toString() |
| + ". It may have already completed."); |
| return; |
| } |
| LOG.info("Changing resource-monitoring for " + containerId); |
| updateContainerMetrics(monitoringEvent); |
| long pmemLimit = |
| changeEvent.getResource().getMemorySize() * 1024L * 1024L; |
| long vmemLimit = (long) (pmemLimit * vmemRatio); |
| int cpuVcores = changeEvent.getResource().getVirtualCores(); |
| processTreeInfo.setResourceLimit(pmemLimit, vmemLimit, cpuVcores); |
| } |
| } |
| |
| private void onStopMonitoringContainer( |
| ContainersMonitorEvent monitoringEvent, ContainerId containerId) { |
| LOG.info("Stopping resource-monitoring for " + containerId); |
| updateContainerMetrics(monitoringEvent); |
| trackingContainers.remove(containerId); |
| } |
| |
| private void onStartMonitoringContainer( |
| ContainersMonitorEvent monitoringEvent, ContainerId containerId) { |
| ContainerStartMonitoringEvent startEvent = |
| (ContainerStartMonitoringEvent) monitoringEvent; |
| LOG.info("Starting resource-monitoring for " + containerId); |
| updateContainerMetrics(monitoringEvent); |
| trackingContainers.put(containerId, |
| new ProcessTreeInfo(containerId, null, null, |
| startEvent.getVmemLimit(), startEvent.getPmemLimit(), |
| startEvent.getCpuVcores())); |
| } |
| } |
