diff --git hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
index b7f264b..cbdfdbd 100644
--- hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
+++ hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
@@ -398,7 +398,8 @@ public synchronized void publishMetricsNow() {
* Sample all the sources for a snapshot of metrics/tags
* @return the metrics buffer containing the snapshot
*/
- synchronized MetricsBuffer sampleMetrics() {
+ @VisibleForTesting
+ public synchronized MetricsBuffer sampleMetrics() {
collector.clear();
MetricsBufferBuilder bufferBuilder = new MetricsBufferBuilder();
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index 6748df7..c029d90 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -882,7 +882,15 @@ private static void addDeprecatedKeys() {
NM_PREFIX + "container-metrics.period-ms";
@Private
public static final int DEFAULT_NM_CONTAINER_METRICS_PERIOD_MS = -1;
-
+
+ /** The delay time ms to unregister container metrics after completion. */
+ @Private
+ public static final String NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS =
+ NM_PREFIX + "container-metrics.unregister-delay-ms";
+ @Private
+ public static final int DEFAULT_NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS =
+ 10000;
+
/** Prefix for all node manager disk health checker configs. */
private static final String NM_DISK_HEALTH_CHECK_PREFIX =
"yarn.nodemanager.disk-health-checker.";
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index de6906e..7b2be61 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -1224,6 +1224,14 @@
${hadoop.tmp.dir}/yarn-nm-recovery
+
+
+ The delay time ms to unregister container metrics after completion.
+
+ yarn.nodemanager.container-metrics.unregister-delay-ms
+ 10000
+
+
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java
index ffa72a4..365fe84 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java
@@ -90,6 +90,7 @@
private boolean flushOnPeriod = false; // true if period elapsed
private boolean finished = false; // true if container finished
private boolean unregister = false; // unregister
+ private long unregisterDelayMs;
private Timer timer; // lazily initialized
/**
@@ -97,15 +98,21 @@
*/
protected final static Map
usageMetrics = new HashMap<>();
+ // Create a timer to unregister container metrics,
+ // whose associated thread run as a daemon.
+ private final static Timer unregisterContainerMetricsTimer =
+ new Timer("Container metrics unregistration", true);
ContainerMetrics(
- MetricsSystem ms, ContainerId containerId, long flushPeriodMs) {
+ MetricsSystem ms, ContainerId containerId, long flushPeriodMs,
+ long delayMs) {
this.recordInfo =
info(sourceName(containerId), RECORD_INFO.description());
this.registry = new MetricsRegistry(recordInfo);
this.metricsSystem = ms;
this.containerId = containerId;
this.flushPeriodMs = flushPeriodMs;
+ this.unregisterDelayMs = delayMs < 0 ? 0 : delayMs;
scheduleTimerTaskIfRequired();
this.pMemMBsStat = registry.newStat(
@@ -134,17 +141,18 @@ static String sourceName(ContainerId containerId) {
}
public static ContainerMetrics forContainer(
- ContainerId containerId, long flushPeriodMs) {
+ ContainerId containerId, long flushPeriodMs, long delayMs) {
return forContainer(
- DefaultMetricsSystem.instance(), containerId, flushPeriodMs);
+ DefaultMetricsSystem.instance(), containerId, flushPeriodMs, delayMs);
}
synchronized static ContainerMetrics forContainer(
- MetricsSystem ms, ContainerId containerId, long flushPeriodMs) {
+ MetricsSystem ms, ContainerId containerId, long flushPeriodMs,
+ long delayMs) {
ContainerMetrics metrics = usageMetrics.get(containerId);
if (metrics == null) {
- metrics = new ContainerMetrics(
- ms, containerId, flushPeriodMs).tag(RECORD_INFO, containerId);
+ metrics = new ContainerMetrics(ms, containerId, flushPeriodMs,
+ delayMs).tag(RECORD_INFO, containerId);
// Register with the MetricsSystems
if (ms != null) {
@@ -158,12 +166,15 @@ synchronized static ContainerMetrics forContainer(
return metrics;
}
+ synchronized static void unregisterContainerMetrics(ContainerMetrics cm) {
+ cm.metricsSystem.unregisterSource(cm.recordInfo.name());
+ usageMetrics.remove(cm.containerId);
+ }
+
@Override
public synchronized void getMetrics(MetricsCollector collector, boolean all) {
//Container goes through registered -> finished -> unregistered.
if (unregister) {
- metricsSystem.unregisterSource(recordInfo.name());
- usageMetrics.remove(containerId);
return;
}
@@ -185,6 +196,7 @@ public synchronized void finished() {
timer.cancel();
timer = null;
}
+ scheduleTimerTaskForUnregistration();
}
public void recordMemoryUsage(int memoryMBs) {
@@ -232,4 +244,14 @@ public void run() {
timer.schedule(timerTask, flushPeriodMs);
}
}
+
+ private void scheduleTimerTaskForUnregistration() {
+ TimerTask timerTask = new TimerTask() {
+ @Override
+ public void run() {
+ ContainerMetrics.unregisterContainerMetrics(ContainerMetrics.this);
+ }
+ };
+ unregisterContainerMetricsTimer.schedule(timerTask, unregisterDelayMs);
+ }
}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
index 5153051..20d2112 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
@@ -54,6 +54,7 @@
private MonitoringThread monitoringThread;
private boolean containerMetricsEnabled;
private long containerMetricsPeriodMs;
+ private long containerMetricsUnregisterDelayMs;
final List containersToBeRemoved;
final Map containersToBeAdded;
@@ -116,6 +117,9 @@ protected void serviceInit(Configuration conf) throws Exception {
this.containerMetricsPeriodMs =
conf.getLong(YarnConfiguration.NM_CONTAINER_METRICS_PERIOD_MS,
YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_PERIOD_MS);
+ this.containerMetricsUnregisterDelayMs = conf.getLong(
+ YarnConfiguration.NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS,
+ YarnConfiguration.DEFAULT_NM_CONTAINER_METRICS_UNREGISTER_DELAY_MS);
long configuredPMemForContainers = conf.getLong(
YarnConfiguration.NM_PMEM_MB,
@@ -379,7 +383,8 @@ public void run() {
for (ContainerId containerId : containersToBeRemoved) {
if (containerMetricsEnabled) {
ContainerMetrics.forContainer(
- containerId, containerMetricsPeriodMs).finished();
+ containerId, containerMetricsPeriodMs,
+ containerMetricsUnregisterDelayMs).finished();
}
trackingContainers.remove(containerId);
LOG.info("Stopping resource-monitoring for " + containerId);
@@ -417,7 +422,8 @@ public void run() {
if (containerMetricsEnabled) {
ContainerMetrics usageMetrics = ContainerMetrics
- .forContainer(containerId, containerMetricsPeriodMs);
+ .forContainer(containerId, containerMetricsPeriodMs,
+ containerMetricsUnregisterDelayMs);
int cpuVcores = ptInfo.getCpuVcores();
final int vmemLimit = (int) (ptInfo.getVmemLimit() >> 20);
final int pmemLimit = (int) (ptInfo.getPmemLimit() >> 20);
@@ -464,10 +470,12 @@ public void run() {
// Add usage to container metrics
if (containerMetricsEnabled) {
ContainerMetrics.forContainer(
- containerId, containerMetricsPeriodMs).recordMemoryUsage(
+ containerId, containerMetricsPeriodMs,
+ containerMetricsUnregisterDelayMs).recordMemoryUsage(
(int) (currentPmemUsage >> 20));
ContainerMetrics.forContainer(
- containerId, containerMetricsPeriodMs).recordCpuUsage
+ containerId, containerMetricsPeriodMs,
+ containerMetricsUnregisterDelayMs).recordCpuUsage
((int)cpuUsagePercentPerCore, milliVcoresUsed);
}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/TestContainerMetrics.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/TestContainerMetrics.java
index c628648..ec06856 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/TestContainerMetrics.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/TestContainerMetrics.java
@@ -22,11 +22,15 @@
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
import org.apache.hadoop.metrics2.impl.MetricsRecords;
+import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -44,7 +48,8 @@ public void testContainerMetricsFlow() throws InterruptedException {
MetricsCollectorImpl collector = new MetricsCollectorImpl();
ContainerId containerId = mock(ContainerId.class);
- ContainerMetrics metrics = ContainerMetrics.forContainer(containerId, 100);
+ ContainerMetrics metrics = ContainerMetrics.forContainer(containerId,
+ 100, 1);
metrics.recordMemoryUsage(1024);
metrics.getMetrics(collector, true);
@@ -82,7 +87,8 @@ public void testContainerMetricsLimit() throws InterruptedException {
MetricsCollectorImpl collector = new MetricsCollectorImpl();
ContainerId containerId = mock(ContainerId.class);
- ContainerMetrics metrics = ContainerMetrics.forContainer(containerId, 100);
+ ContainerMetrics metrics = ContainerMetrics.forContainer(containerId,
+ 100, 1);
int anyPmemLimit = 1024;
int anyVmemLimit = 2048;
@@ -107,4 +113,39 @@ public void testContainerMetricsLimit() throws InterruptedException {
collector.clear();
}
+
+ @Test
+ public void testContainerMetricsFinished() throws InterruptedException {
+ MetricsSystemImpl system = new MetricsSystemImpl();
+ system.init("test");
+ MetricsCollectorImpl collector = new MetricsCollectorImpl();
+ ApplicationId appId = ApplicationId.newInstance(1234, 3);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 4);
+ ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 1);
+ ContainerMetrics metrics1 = ContainerMetrics.forContainer(system,
+ containerId1, 1, 0);
+ ContainerId containerId2 = ContainerId.newContainerId(appAttemptId, 2);
+ ContainerMetrics metrics2 = ContainerMetrics.forContainer(system,
+ containerId2, 1, 0);
+ ContainerId containerId3 = ContainerId.newContainerId(appAttemptId, 3);
+ ContainerMetrics metrics3 = ContainerMetrics.forContainer(system,
+ containerId3, 1, 0);
+ metrics1.finished();
+ metrics2.finished();
+ system.sampleMetrics();
+ system.sampleMetrics();
+ Thread.sleep(100);
+ system.stop();
+ // verify metrics1 is unregistered
+ assertTrue(metrics1 != ContainerMetrics.forContainer(
+ system, containerId1, 1, 0));
+ // verify metrics2 is unregistered
+ assertTrue(metrics2 != ContainerMetrics.forContainer(
+ system, containerId2, 1, 0));
+ // verify metrics3 is still registered
+ assertTrue(metrics3 == ContainerMetrics.forContainer(
+ system, containerId3, 1, 0));
+ system.shutdown();
+ }
}