YARN-6492. Generate queue metrics for each partition. Contributed by Manikandan R

(cherry picked from commit c30c23c)
(cherry picked from commit 7a323a4)
(cherry picked from commit a80595a6deb3124a3d6d99057e9d5298cd7237d8)

Author: hungj

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/Resources.java

@@ -298,6 +298,19 @@ public static Resource subtractFromNonNegative(Resource lhs, Resource rhs) {
     return lhs;
   }
 
+  /**
+   * Subtract {@code rhs} from {@code lhs} and reset any negative values to
+   * zero. This call will operate on a copy of {@code lhs}, leaving {@code lhs}
+   * unmodified.
+   *
+   * @param lhs {@link Resource} to subtract from
+   * @param rhs {@link Resource} to subtract
+   * @return the value of lhs after subtraction
+   */
+  public static Resource subtractNonNegative(Resource lhs, Resource rhs) {
+    return subtractFromNonNegative(clone(lhs), rhs);
+  }
+
   public static Resource negate(Resource resource) {
     return subtract(NONE, resource);
   }

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

@@ -37,7 +37,6 @@
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -531,7 +530,7 @@ public boolean isPlaceBlacklisted(String resourceName,
 
   public ContainerRequest allocate(NodeType type,
       SchedulerNode node, SchedulerRequestKey schedulerKey,
-      Container containerAllocated) {
+      RMContainer containerAllocated) {
     try {
       writeLock.lock();
 
@@ -690,7 +689,7 @@ public boolean checkAllocation(NodeType type, SchedulerNode node,
   }
 
   private void updateMetricsForAllocatedContainer(NodeType type,
-      SchedulerNode node, Container containerAllocated) {
+      SchedulerNode node, RMContainer containerAllocated) {
     QueueMetrics metrics = queue.getMetrics();
     if (pending) {
       // once an allocation is done we assume the application is
@@ -703,18 +702,21 @@ private void updateMetricsForAllocatedContainer(NodeType type,
   }
 
   public static void updateMetrics(ApplicationId applicationId, NodeType type,
-      SchedulerNode node, Container containerAllocated, String user,
+      SchedulerNode node, RMContainer containerAllocated, String user,
       Queue queue) {
     if (LOG.isDebugEnabled()) {
       LOG.debug("allocate: applicationId=" + applicationId + " container="
-          + containerAllocated.getId() + " host=" + containerAllocated
-          .getNodeId().toString() + " user=" + user + " resource="
-          + containerAllocated.getResource() + " type="
-          + type);
+          + containerAllocated.getContainer().getId() + " host="
+          + containerAllocated.getNodeId().toString() + " user="
+          + user + " resource="
+          + containerAllocated.getContainer().getResource() + " type=" + type);
     }
     if(node != null) {
       queue.getMetrics().allocateResources(node.getPartition(), user, 1,
-          containerAllocated.getResource(), true);
+          containerAllocated.getContainer().getResource(), false);
+      queue.getMetrics().decrPendingResources(
+          containerAllocated.getNodeLabelExpression(), user, 1,
+          containerAllocated.getContainer().getResource());
     }
     queue.getMetrics().incrNodeTypeAggregations(user, type);
   }
@@ -768,4 +770,8 @@ public boolean precheckNode(SchedulerRequestKey schedulerKey,
       this.readLock.unlock();
     }
   }
+
+  public RMContext getRMContext() {
+    return this.rmContext;
+  }
 }

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

@@ -161,11 +161,17 @@ private void cancelPreviousRequest(SchedulerNode schedulerNode,
       // Decrement the pending using a dummy RR with
       // resource = prev update req capability
       if (pendingAsk != null && pendingAsk.getCount() > 0) {
+        Container container = Container.newInstance(UNDEFINED,
+            schedulerNode.getNodeID(), "host:port",
+            pendingAsk.getPerAllocationResource(),
+            schedulerKey.getPriority(), null);
         appSchedulingInfo.allocate(NodeType.OFF_SWITCH, schedulerNode,
-            schedulerKey, Container.newInstance(UNDEFINED,
-                schedulerNode.getNodeID(), "host:port",
-                pendingAsk.getPerAllocationResource(),
-                schedulerKey.getPriority(), null));
+            schedulerKey,
+            new RMContainerImpl(container, schedulerKey,
+                appSchedulingInfo.getApplicationAttemptId(),
+                schedulerNode.getNodeID(), appSchedulingInfo.getUser(),
+                appSchedulingInfo.getRMContext(),
+                appPlacementAllocator.getPrimaryRequestedNodePartition()));
       }
     }
   }

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

@@ -0,0 +1,89 @@
+/**
+ * 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.resourcemanager.scheduler;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.annotation.Metrics;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
+
+@Metrics(context = "yarn")
+public class PartitionQueueMetrics extends QueueMetrics {
+
+  private String partition;
+
+  protected PartitionQueueMetrics(MetricsSystem ms, String queueName,
+      Queue parent, boolean enableUserMetrics, Configuration conf,
+      String partition) {
+    super(ms, queueName, parent, enableUserMetrics, conf);
+    this.partition = partition;
+    if (getParentQueue() != null) {
+      String newQueueName = (getParentQueue() instanceof CSQueue)
+          ? ((CSQueue) getParentQueue()).getQueuePath()
+          : getParentQueue().getQueueName();
+      String parentMetricName =
+          partition + METRIC_NAME_DELIMITER + newQueueName;
+      setParent(getQueueMetrics().get(parentMetricName));
+    }
+  }
+
+  /**
+   * Partition * Queue * User Metrics
+   *
+   * Computes Metrics at Partition (Node Label) * Queue * User Level.
+   *
+   * Sample JMX O/P Structure:
+   *
+   * PartitionQueueMetrics (labelX)
+   *  QueueMetrics (A)
+   *    usermetrics
+   *  QueueMetrics (A1)
+   *    usermetrics
+   *    QueueMetrics (A2)
+   *      usermetrics
+   *    QueueMetrics (B)
+   *      usermetrics
+   *
+   * @return QueueMetrics
+   */
+  @Override
+  public synchronized QueueMetrics getUserMetrics(String userName) {
+    if (users == null) {
+      return null;
+    }
+
+    String partitionJMXStr =
+        (partition.equals(DEFAULT_PARTITION)) ? DEFAULT_PARTITION_JMX_STR
+            : partition;
+
+    QueueMetrics metrics = (PartitionQueueMetrics) users.get(userName);
+    if (metrics == null) {
+      metrics = new PartitionQueueMetrics(this.metricsSystem, this.queueName,
+          null, false, this.conf, this.partition);
+      users.put(userName, metrics);
+      metricsSystem.register(
+          pSourceName(partitionJMXStr).append(qSourceName(queueName))
+              .append(",user=").append(userName).toString(),
+          "Metrics for user '" + userName + "' in queue '" + queueName + "'",
+          ((PartitionQueueMetrics) metrics.tag(PARTITION_INFO, partitionJMXStr)
+              .tag(QUEUE_INFO, queueName)).tag(USER_INFO, userName));
+    }
+    return metrics;
+  }
+}