YARN-10154. CS Dynamic Queues cannot be configured with absolute resources. Contributed by Manikandan R.

Author: sunilgovind

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 
+import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler
     .SchedulerDynamicEditException;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common
@@ -67,6 +68,22 @@ public void setEntitlement(QueueEntitlement entitlement)
      setEntitlement(NO_LABEL, entitlement);
   }
 
+  @Override
+  protected Resource getMinimumAbsoluteResource(String queuePath,
+      String label) {
+    return super.getMinimumAbsoluteResource(csContext.getConfiguration()
+        .getAutoCreatedQueueTemplateConfPrefix(this.getParent().getQueuePath()),
+        label);
+  }
+
+  @Override
+  protected Resource getMaximumAbsoluteResource(String queuePath,
+      String label) {
+    return super.getMaximumAbsoluteResource(csContext.getConfiguration()
+        .getAutoCreatedQueueTemplateConfPrefix(this.getParent().getQueuePath()),
+        label);
+  }
+
   /**
    * This methods to change capacity for a queue and adjusts its
    * absoluteCapacity

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

@@ -524,16 +524,26 @@ private void setupMaximumAllocation(CapacitySchedulerConfiguration csConf) {
     return unionInheritedWeights;
   }
 
+  protected Resource getMinimumAbsoluteResource(String queuePath, String label) {
+    Resource minResource = csContext.getConfiguration()
+        .getMinimumResourceRequirement(label, queuePath, resourceTypes);
+    return minResource;
+  }
+
+  protected Resource getMaximumAbsoluteResource(String queuePath, String label) {
+    Resource maxResource = csContext.getConfiguration()
+        .getMaximumResourceRequirement(label, queuePath, resourceTypes);
+    return maxResource;
+  }
+
   protected void updateConfigurableResourceRequirement(String queuePath,
       Resource clusterResource) {
     CapacitySchedulerConfiguration conf = csContext.getConfiguration();
     Set<String> configuredNodelabels = conf.getConfiguredNodeLabels(queuePath);
 
     for (String label : configuredNodelabels) {
-      Resource minResource = conf.getMinimumResourceRequirement(label,
-          queuePath, resourceTypes);
-      Resource maxResource = conf.getMaximumResourceRequirement(label,
-          queuePath, resourceTypes);
+      Resource minResource = getMinimumAbsoluteResource(queuePath, label);
+      Resource maxResource = getMaximumAbsoluteResource(queuePath, label);
 
       LOG.debug("capacityConfigType is '{}' for queue {}",
           capacityConfigType, getQueuePath());

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

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 
 import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
 
@@ -74,6 +75,9 @@ public void reinitializeFromTemplate(AutoCreatedLeafQueueConfig
     writeLock.lock();
     try {
 
+      this.getParent().updateClusterResource(this.csContext.getClusterResource(),
+          new ResourceLimits(this.csContext.getClusterResource()));
+
       // TODO:
       // reinitialize only capacities for now since 0 capacity updates
       // can cause
@@ -100,7 +104,7 @@ public void reinitializeFromTemplate(AutoCreatedLeafQueueConfig
     }
   }
 
-  private void mergeCapacities(QueueCapacities capacities) {
+  public void mergeCapacities(QueueCapacities capacities) {
     for ( String nodeLabel : capacities.getExistingNodeLabels()) {
       queueCapacities.setCapacity(nodeLabel,
           capacities.getCapacity(nodeLabel));

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

@@ -557,6 +557,12 @@ public void setMaximumCapacityByLabel(String queue, String label,
       float capacity) {
     setFloat(getNodeLabelPrefix(queue, label) + MAXIMUM_CAPACITY, capacity);
   }
+
+  public void setMaximumCapacityByLabel(String queue, String label,
+      String absoluteResourceCapacity) {
+    set(getNodeLabelPrefix(queue, label) + MAXIMUM_CAPACITY,
+        absoluteResourceCapacity);
+  }
 
   public int getUserLimit(String queue) {
     int userLimit = getInt(getQueuePrefix(queue) + USER_LIMIT,
@@ -1960,11 +1966,29 @@ public void setAutoCreatedLeafQueueConfigCapacity(String queuePath,
   @Private
   public void setAutoCreatedLeafQueueTemplateCapacityByLabel(String queuePath,
       String label, float val) {
-    String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix(
-        queuePath);
+    String leafQueueConfPrefix =
+        getAutoCreatedQueueTemplateConfPrefix(queuePath);
     setCapacityByLabel(leafQueueConfPrefix, label, val);
   }
 
+  @VisibleForTesting
+  @Private
+  public void setAutoCreatedLeafQueueTemplateCapacityByLabel(String queuePath,
+      String label, Resource resource) {
+
+    String leafQueueConfPrefix =
+        getAutoCreatedQueueTemplateConfPrefix(queuePath);
+
+    StringBuilder resourceString = new StringBuilder();
+    resourceString
+        .append("[" + AbsoluteResourceType.MEMORY.toString().toLowerCase() + "="
+            + resource.getMemorySize() + ","
+            + AbsoluteResourceType.VCORES.toString().toLowerCase() + "="
+            + resource.getVirtualCores() + "]");
+
+    setCapacityByLabel(leafQueueConfPrefix, label, resourceString.toString());
+  }
+
   @Private
   @VisibleForTesting
   public void setAutoCreatedLeafQueueConfigMaxCapacity(String queuePath,
@@ -1983,6 +2007,23 @@ public void setAutoCreatedLeafQueueTemplateMaxCapacity(String queuePath,
     setMaximumCapacityByLabel(leafQueueConfPrefix, label, val);
   }
 
+  @Private
+  @VisibleForTesting
+  public void setAutoCreatedLeafQueueTemplateMaxCapacity(String queuePath,
+      String label, Resource resource) {
+    String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix(
+        queuePath);
+
+    StringBuilder resourceString = new StringBuilder();
+    resourceString
+        .append("[" + AbsoluteResourceType.MEMORY.toString().toLowerCase() + "="
+            + resource.getMemorySize() + ","
+            + AbsoluteResourceType.VCORES.toString().toLowerCase() + "="
+            + resource.getVirtualCores() + "]");
+
+    setMaximumCapacityByLabel(leafQueueConfPrefix, label, resourceString.toString());
+  }
+
   @VisibleForTesting
   @Private
   public void setAutoCreatedLeafQueueConfigUserLimit(String queuePath,
@@ -2115,7 +2156,6 @@ private Resource internalGetLabeledResourceRequirementForQueue(String queue,
       if (subGroup.trim().isEmpty()) {
         return Resources.none();
       }
-
       subGroup = subGroup.substring(1, subGroup.length() - 1);
       for (String kvPair : subGroup.trim().split(",")) {
         String[] splits = kvPair.split("=");

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

@@ -22,9 +22,10 @@
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler
     .SchedulerDynamicEditException;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.AbstractCSQueue.CapacityConfigType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica
     .FiCaSchedulerApp;
-
+import org.apache.hadoop.yarn.util.resource.Resources;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -150,24 +151,78 @@ private void reinitializeQueueManagementPolicy() throws IOException {
     }
   }
 
-  protected AutoCreatedLeafQueueConfig.Builder initializeLeafQueueConfigs() {
+  protected AutoCreatedLeafQueueConfig.Builder initializeLeafQueueConfigs() throws IOException {
 
     AutoCreatedLeafQueueConfig.Builder builder =
         new AutoCreatedLeafQueueConfig.Builder();
 
     String leafQueueTemplateConfPrefix = getLeafQueueConfigPrefix(
         csContext.getConfiguration());
     //Load template configuration
-    builder.configuration(
-        super.initializeLeafQueueConfigs(leafQueueTemplateConfPrefix));
+    CapacitySchedulerConfiguration conf =
+        super.initializeLeafQueueConfigs(leafQueueTemplateConfPrefix);
+    builder.configuration(conf);
+
+    for (String nodeLabel : conf
+        .getConfiguredNodeLabels(csContext.getConfiguration()
+            .getAutoCreatedQueueTemplateConfPrefix(getQueuePath()))) {
+      Resource templateMinResource = conf.getMinimumResourceRequirement(
+          nodeLabel, csContext.getConfiguration()
+              .getAutoCreatedQueueTemplateConfPrefix(getQueuePath()),
+          resourceTypes);
+
+      if (this.capacityConfigType.equals(CapacityConfigType.PERCENTAGE)
+          && !templateMinResource.equals(Resources.none())) {
+        throw new IOException("Managed Parent Queue " + this.getQueuePath()
+            + " config type is different from leaf queue template config type");
+      }
+    }
 
     //Load template capacities
     QueueCapacities queueCapacities = new QueueCapacities(false);
     CSQueueUtils.loadUpdateAndCheckCapacities(csContext.getConfiguration()
             .getAutoCreatedQueueTemplateConfPrefix(getQueuePath()),
         csContext.getConfiguration(), queueCapacities, getQueueCapacities());
-    builder.capacities(queueCapacities);
 
+
+    /**
+     * Populate leaf queue template (of Parent resources configured in
+     * ABSOLUTE_RESOURCE) capacities with actual values for which configured has
+     * been defined in ABSOLUTE_RESOURCE format.
+     *
+     */
+    if (this.capacityConfigType.equals(CapacityConfigType.ABSOLUTE_RESOURCE)) {
+      for (String label : queueCapacities.getExistingNodeLabels()) {
+        queueCapacities.setCapacity(label,
+            this.csContext.getResourceCalculator().divide(
+                this.csContext.getClusterResource(),
+                this.csContext.getConfiguration().getMinimumResourceRequirement(
+                    label,
+                    this.csContext.getConfiguration()
+                        .getAutoCreatedQueueTemplateConfPrefix(getQueuePath()),
+                    resourceTypes),
+                getQueueResourceQuotas().getConfiguredMinResource(label)));
+
+        queueCapacities.setMaximumCapacity(label,
+            this.csContext.getResourceCalculator().divide(
+                this.csContext.getClusterResource(),
+                this.csContext.getConfiguration().getMaximumResourceRequirement(
+                    label,
+                    this.csContext.getConfiguration()
+                        .getAutoCreatedQueueTemplateConfPrefix(getQueuePath()),
+                    resourceTypes),
+                getQueueResourceQuotas().getConfiguredMaxResource(label)));
+
+        queueCapacities.setAbsoluteCapacity(label,
+            queueCapacities.getCapacity(label)
+                * getQueueCapacities().getAbsoluteCapacity(label));
+
+        queueCapacities.setAbsoluteMaximumCapacity(label,
+            queueCapacities.getMaximumCapacity(label)
+                * getQueueCapacities().getAbsoluteMaximumCapacity(label));
+      }
+    }
+    builder.capacities(queueCapacities);
     return builder;
   }
 

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

@@ -933,7 +933,6 @@ public boolean hasChildQueues() {
 
   private void calculateEffectiveResourcesAndCapacity(String label,
       Resource clusterResource) {
-
     // For root queue, ensure that max/min resource is updated to latest
     // cluster resource.
     Resource resourceByLabel = labelManager.getResourceByLabel(label,
@@ -1134,7 +1133,9 @@ private void deriveCapacityFromAbsoluteConfigurations(String label,
       LOG.info("LeafQueue:" + leafQueue.getQueuePath() + ", maxApplications="
           + maxApplications + ", maxApplicationsPerUser="
           + maxApplicationsPerUser + ", Abs Cap:"
-          + childQueue.getQueueCapacities().getAbsoluteCapacity(label));
+          + childQueue.getQueueCapacities().getAbsoluteCapacity(label) + ", Cap: "
+          + childQueue.getQueueCapacities().getCapacity(label) + ", MaxCap : "
+          + childQueue.getQueueCapacities().getMaximumCapacity(label));
     }
   }
 

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

@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler
     .SchedulerDynamicEditException;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
@@ -51,6 +52,7 @@
     .FiCaSchedulerApp;
 import org.apache.hadoop.yarn.util.Clock;
 import org.apache.hadoop.yarn.util.MonotonicClock;
+import org.apache.hadoop.yarn.util.resource.Resources;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -679,6 +681,19 @@ public void commitQueueManagementChanges(
               LOG.debug("Queue is already de-activated. Skipping "
                   + "de-activation : {}", leafQueue.getQueuePath());
             } else{
+              /**
+               * While deactivating queues of type ABSOLUTE_RESOURCE, configured
+               * min resource has to be set based on updated capacity (which is
+               * again based on updated queue entitlements). Otherwise,
+               * ParentQueue#calculateEffectiveResourcesAndCapacity calculations
+               * leads to incorrect results.
+               */
+              leafQueue
+                  .mergeCapacities(updatedQueueTemplate.getQueueCapacities());
+              leafQueue.getQueueResourceQuotas()
+                  .setConfiguredMinResource(Resources.multiply(
+                      this.scheduler.getClusterResource(), updatedQueueTemplate
+                          .getQueueCapacities().getCapacity(nodeLabel)));
               deactivate(leafQueue, nodeLabel);
             }
           }