YARN-9756: Create metric that sums total memory/vcores preempted per round. Contributed by Manikandan R (manirajv06).

Author: Eric E Payne

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java

@@ -77,6 +77,10 @@ public class QueueMetrics implements MetricsSource {
   @Metric("# of active applications") MutableGaugeInt activeApplications;
   @Metric("App Attempt First Container Allocation Delay")
     MutableRate appAttemptFirstContainerAllocationDelay;
+  @Metric("Aggregate total of preempted memory MB")
+    MutableCounterLong aggregateMemoryMBPreempted;
+  @Metric("Aggregate total of preempted vcores")
+    MutableCounterLong aggregateVcoresPreempted;
 
   //Metrics updated only for "default" partition
   @Metric("Allocated memory in MB") MutableGaugeLong allocatedMB;
@@ -586,6 +590,23 @@ public void updatePreemptedVcoreSeconds(long vcoreSeconds) {
     }
   }
 
+  public void updatePreemptedResources(Resource res) {
+    aggregateMemoryMBPreempted.incr(res.getMemorySize());
+    aggregateVcoresPreempted.incr(res.getVirtualCores());
+    if (parent != null) {
+      parent.updatePreemptedResources(res);
+    }
+  }
+
+  public void updatePreemptedForCustomResources(Resource res) {
+    if (queueMetricsForCustomResources != null) {
+      queueMetricsForCustomResources.increaseAggregatedPreempted(res);
+    }
+    if (parent != null) {
+      parent.updatePreemptedForCustomResources(res);
+    }
+  }
+
   public void updatePreemptedSecondsForCustomResources(Resource res,
           long seconds) {
     if (queueMetricsForCustomResources != null) {
@@ -753,6 +774,16 @@ public MutableCounterLong getAggregateVcoreSecondsPreempted() {
     return aggregateVcoreSecondsPreempted;
   }
 
+  @VisibleForTesting
+  public long getAggregateMemoryMBPreempted() {
+    return aggregateMemoryMBPreempted.value();
+  }
+
+  @VisibleForTesting
+  public long getAggregateVcoresPreempted() {
+    return aggregateVcoresPreempted.value();
+  }
+
   public long getAllocatedMB() {
     return allocatedMB.value();
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetricsForCustomResources.java

@@ -28,6 +28,8 @@
 public class QueueMetricsForCustomResources {
   private final QueueMetricsCustomResource aggregatePreemptedSeconds =
       new QueueMetricsCustomResource();
+  private final QueueMetricsCustomResource aggregatePreempted =
+    new QueueMetricsCustomResource();
   private final QueueMetricsCustomResource allocated =
       new QueueMetricsCustomResource();
   private final QueueMetricsCustomResource available =
@@ -82,6 +84,10 @@ public void increaseAggregatedPreemptedSeconds(Resource res, long seconds) {
     aggregatePreemptedSeconds.increaseWithMultiplier(res, seconds);
   }
 
+  public void increaseAggregatedPreempted(Resource res) {
+    aggregatePreempted.increase(res);
+  }
+
   Map<String, Long> getAllocatedValues() {
     return allocated.getValues();
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java

@@ -2222,7 +2222,9 @@ private void updateQueuePreemptionMetrics(RMContainer rmc) {
         / DateUtils.MILLIS_PER_SECOND;
     metrics.updatePreemptedMemoryMBSeconds(mbSeconds);
     metrics.updatePreemptedVcoreSeconds(vcSeconds);
+    metrics.updatePreemptedResources(containerResource);
     metrics.updatePreemptedSecondsForCustomResources(containerResource,
         usedSeconds);
+    metrics.updatePreemptedForCustomResources(containerResource);
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerSurgicalPreemption.java

@@ -185,6 +185,13 @@ protected RMNodeLabelsManager createNodeLabelManager() {
     Assert.assertEquals("Number of preempted containers incorrectly recorded:",
         4, cs.getQueue("root").getMetrics().getAggregatePreemptedContainers());
 
+    Assert.assertEquals("Amount of preempted memory incorrectly recorded:",
+        4 * GB,
+        cs.getQueue("root").getMetrics().getAggregateMemoryMBPreempted());
+
+    Assert.assertEquals("Number of preempted vcores incorrectly recorded:", 4,
+        cs.getQueue("root").getMetrics().getAggregateVcoresPreempted());
+
     rm1.close();
   }