YARN-11005. Implement the core QUEUE_LENGTH_THEN_RESOURCES OContainer allocation policy

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

@@ -404,18 +404,51 @@ public boolean isAllInvalidDivisor(Resource r) {
 
   @Override
   public float ratio(Resource a, Resource b) {
-    float ratio = 0.0f;
+    return ratio(a, b, true);
+  }
+
+  /**
+   * Computes the ratio of resource a over resource b,
+   * where the boolean flag {@literal isDominantShare} allows
+   * specification of whether the max- or min-share should be computed.
+   * @param a the numerator resource.
+   * @param b the denominator resource.
+   * @param isDominantShare whether the dominant (max) share should be computed,
+   *                        computes the min-share if false.
+   * @return the max- or min-share ratio of the resources.
+   */
+  private float ratio(Resource a, Resource b, boolean isDominantShare) {
+    float ratio = isDominantShare ? 0.0f : 1.0f;
     int maxLength = ResourceUtils.getNumberOfCountableResourceTypes();
     for (int i = 0; i < maxLength; i++) {
       ResourceInformation aResourceInformation = a.getResourceInformation(i);
       ResourceInformation bResourceInformation = b.getResourceInformation(i);
       final float tmp = divideSafelyAsFloat(aResourceInformation.getValue(),
           bResourceInformation.getValue());
-      ratio = ratio > tmp ? ratio : tmp;
+      if (isDominantShare) {
+        ratio = Math.max(ratio, tmp);
+      } else {
+        ratio = Math.min(ratio, tmp);
+      }
     }
     return ratio;
   }
 
+  /**
+   * Computes the ratio of resource a over resource b.
+   * However, different from ratio(Resource, Resource),
+   * this returns the min-share of the resources.
+   * For example, ratio(Resource(10, 50), Resource(100, 100)) would return 0.5,
+   * whereas minRatio(Resource(10, 50), Resource(100, 100)) would return 0.1.
+   * @param a the numerator resource.
+   * @param b the denominator resource.
+   * @return the min-share ratio of the resources.
+   */
+  @Unstable
+  public float minRatio(Resource a, Resource b) {
+    return ratio(a, b, false);
+  }
+
   @Override
   public Resource divideAndCeil(Resource numerator, int denominator) {
     return divideAndCeil(numerator, (long) denominator);

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

@@ -257,7 +257,9 @@ public OpportunisticContainerAllocatorAMService(RMContext rmContext,
 
     int limitMin, limitMax;
 
-    if (comparator == NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH) {
+    if (comparator == NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH ||
+        comparator ==
+            NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES) {
       limitMin = rmContext.getYarnConfiguration()
           .getInt(YarnConfiguration.NM_CONTAINER_QUEUING_MIN_QUEUE_LENGTH,
               YarnConfiguration.

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

@@ -251,14 +251,15 @@ private List<Container> allocateNodeLocal(
       String userName, Map<Resource, List<Allocation>> allocations)
       throws YarnException {
     List<Container> allocatedContainers = new ArrayList<>();
+    final ResourceRequest resourceRequest = enrichedAsk.getRequest();
     while (toAllocate > 0) {
       RMNode node = nodeQueueLoadMonitor.selectLocalNode(nodeLocation,
-          blacklist);
+          blacklist, resourceRequest.getCapability());
       if (node != null) {
         toAllocate--;
         Container container = createContainer(rmIdentifier, appParams,
             idCounter, id, userName, allocations, nodeLocation,
-            enrichedAsk.getRequest(), convertToRemoteNode(node));
+            resourceRequest, convertToRemoteNode(node));
         allocatedContainers.add(container);
         LOG.info("Allocated [{}] as opportunistic at location [{}]",
             container.getId(), nodeLocation);
@@ -280,14 +281,15 @@ private List<Container> allocateRackLocal(EnrichedResourceRequest enrichedAsk,
       String userName, Map<Resource, List<Allocation>> allocations)
       throws YarnException {
     List<Container> allocatedContainers = new ArrayList<>();
+    final ResourceRequest resourceRequest = enrichedAsk.getRequest();
     while (toAllocate > 0) {
       RMNode node = nodeQueueLoadMonitor.selectRackLocalNode(rackLocation,
-          blacklist);
+          blacklist, resourceRequest.getCapability());
       if (node != null) {
         toAllocate--;
         Container container = createContainer(rmIdentifier, appParams,
             idCounter, id, userName, allocations, rackLocation,
-            enrichedAsk.getRequest(), convertToRemoteNode(node));
+            resourceRequest, convertToRemoteNode(node));
         allocatedContainers.add(container);
         metrics.incrRackLocalOppContainers();
         LOG.info("Allocated [{}] as opportunistic at location [{}]",
@@ -309,13 +311,15 @@ private List<Container> allocateAny(EnrichedResourceRequest enrichedAsk,
       String userName, Map<Resource, List<Allocation>> allocations)
       throws YarnException {
     List<Container> allocatedContainers = new ArrayList<>();
+    final ResourceRequest resourceRequest = enrichedAsk.getRequest();
     while (toAllocate > 0) {
-      RMNode node = nodeQueueLoadMonitor.selectAnyNode(blacklist);
+      RMNode node = nodeQueueLoadMonitor.selectAnyNode(
+          blacklist, resourceRequest.getCapability());
       if (node != null) {
         toAllocate--;
         Container container = createContainer(rmIdentifier, appParams,
             idCounter, id, userName, allocations, ResourceRequest.ANY,
-            enrichedAsk.getRequest(), convertToRemoteNode(node));
+            resourceRequest, convertToRemoteNode(node));
         allocatedContainers.add(container);
         metrics.incrOffSwitchOppContainers();
         LOG.info("Allocated [{}] as opportunistic at location [{}]",

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

@@ -22,6 +22,8 @@
 import java.util.HashSet;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.util.resource.Resources;
 
 /**
  * Represents a node in the cluster from the NodeQueueLoadMonitor's perspective
@@ -33,13 +35,42 @@ public class ClusterNode {
   final NodeId nodeId;
   private int queueCapacity = 0;
   private final HashSet<String> labels;
+  private Resource capability = Resources.none();
+  private Resource allocatedResource = Resources.none();
 
   public ClusterNode(NodeId nodeId) {
     this.nodeId = nodeId;
     this.labels = new HashSet<>();
     updateTimestamp();
   }
 
+  public ClusterNode setCapability(Resource nodeCapability) {
+    if (nodeCapability == null) {
+      this.capability = Resources.none();
+    } else {
+      this.capability = nodeCapability;
+    }
+    return this;
+  }
+
+  public ClusterNode setAllocatedResource(
+      Resource allocResource) {
+    if (allocResource == null) {
+      this.allocatedResource = Resources.none();
+    } else {
+      this.allocatedResource = allocResource;
+    }
+    return this;
+  }
+
+  public Resource getAllocatedResource() {
+    return this.allocatedResource;
+  }
+
+  public Resource getCapability() {
+    return this.capability;
+  }
+
   public ClusterNode setQueueLength(int qLength) {
     this.queueLength.set(qLength);
     return this;