FDPINFDV-1806( YARN-6492 )

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

@@ -192,6 +192,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

@@ -41,6 +41,7 @@
 import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
@@ -89,10 +90,12 @@ public class AppSchedulingInfo {
   private final ReentrantReadWriteLock.WriteLock writeLock;
 
   public final ContainerUpdateContext updateContext;
+
+  private final RMContext rmContext;
 
   public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
       String user, Queue queue, AbstractUsersManager abstractUsersManager,
-      long epoch, ResourceUsage appResourceUsage) {
+      long epoch, ResourceUsage appResourceUsage, RMContext rmContext) {
     this.applicationAttemptId = appAttemptId;
     this.applicationId = appAttemptId.getApplicationId();
     this.queue = queue;
@@ -106,6 +109,7 @@ public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
     updateContext = new ContainerUpdateContext(this);
     readLock = lock.readLock();
     writeLock = lock.writeLock();
+    this.rmContext = rmContext;
   }
 
   public ApplicationId getApplicationId() {
@@ -437,7 +441,7 @@ public boolean isPlaceBlacklisted(String resourceName,
 
   public List<ResourceRequest> allocate(NodeType type,
       SchedulerNode node, SchedulerRequestKey schedulerKey,
-      Container containerAllocated) {
+      RMContainer containerAllocated) {
     try {
       writeLock.lock();
 
@@ -591,7 +595,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
@@ -602,14 +606,16 @@ private void updateMetricsForAllocatedContainer(NodeType type,
 
     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.getContainer().getNodeId().toString() + " user="
+          + user + " resource="
+          + containerAllocated.getContainer().getResource() + " type=" + type);
     }
-    if(node != null) {
+    if (node != null) {
       metrics.allocateResources(node.getPartition(), user, 1,
-          containerAllocated.getResource(), true);
+          containerAllocated.getContainer().getResource(), false);
+      metrics.decrPendingResources(containerAllocated.getNodeLabelExpression(),
+          user, 1, containerAllocated.getContainer().getResource());
     }
     metrics.incrNodeTypeAggregations(user, type);
   }
@@ -655,4 +661,8 @@ public boolean acceptNodePartition(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

@@ -162,10 +162,16 @@ private void cancelPreviousRequest(SchedulerNode schedulerNode,
       // Decrement the pending using a dummy RR with
       // resource = prev update req capability
       if (prevReq != null) {
+        Container container = Container.newInstance(UNDEFINED,
+            schedulerNode.getNodeID(), "host:port", prevReq.getCapability(),
+            schedulerKey.getPriority(), null);
         appSchedulingInfo.allocate(NodeType.OFF_SWITCH, schedulerNode,
-            schedulerKey, Container.newInstance(UNDEFINED,
-                schedulerNode.getNodeID(), "host:port",
-                prevReq.getCapability(), schedulerKey.getPriority(), null));
+            schedulerKey,
+            new RMContainerImpl(container, schedulerKey,
+                appSchedulingInfo.getApplicationAttemptId(),
+                schedulerNode.getNodeID(), appSchedulingInfo.getUser(),
+                appSchedulingInfo.getRMContext(),
+                schedulingPlacementSet.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;
+  }
+}