YARN-10965. Centralize queue resource calculation based on CapacityVectors

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceUtils.java

@@ -818,6 +818,28 @@ public static Resource createResourceWithSameValue(long value) {
     return res;
   }
 
+  public static Resource multiply(Resource resource, float multiplier) {
+    Resource newResource = Resource.newInstance(0, 0);
+
+    for (ResourceInformation resourceInformation : resource.getResources()) {
+      newResource.setResourceValue(resourceInformation.getName(),
+          (long) Math.floor(resourceInformation.getValue() * multiplier));
+    }
+
+    return newResource;
+  }
+
+  public static Resource multiplyRound(Resource resource, float multiplier) {
+    Resource newResource = Resource.newInstance(0, 0);
+
+    for (ResourceInformation resourceInformation : resource.getResources()) {
+      newResource.setResourceValue(resourceInformation.getName(),
+          Math.round(resourceInformation.getValue() * multiplier));
+    }
+
+    return newResource;
+  }
+
   @InterfaceAudience.Private
   @InterfaceStability.Unstable
   public static Resource createResourceFromString(

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

@@ -0,0 +1,56 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.capacity;
+
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacityVector.ResourceUnitCapacityType;
+
+public class AbsoluteResourceCapacityCalculator extends AbstractQueueCapacityCalculator {
+
+  @Override
+  public float calculateMinimumResource(
+       ResourceCalculationDriver resourceCalculationDriver, String label) {
+    String resourceName = resourceCalculationDriver.getCurrentResourceName();
+    float normalizedRatio = resourceCalculationDriver.getNormalizedResourceRatios().getOrDefault(
+        label, ResourceVector.of(1)).getValue(resourceName);
+    float remainingResourceRatio = resourceCalculationDriver.getRemainingRatioOfResource(
+        label, resourceName);
+
+    return normalizedRatio * remainingResourceRatio * resourceCalculationDriver
+        .getCurrentMinimumCapacityEntry(label).getResourceValue();
+  }
+
+  @Override
+  public float calculateMaximumResource(
+       ResourceCalculationDriver resourceCalculationDriver, String label) {
+    return resourceCalculationDriver.getCurrentMaximumCapacityEntry(label).getResourceValue();
+  }
+
+  @Override
+  public void updateCapacitiesAfterCalculation(
+      ResourceCalculationDriver resourceCalculationDriver, String label) {
+    setQueueCapacities(resourceCalculationDriver.getUpdateContext().getUpdatedClusterResource(
+        label), resourceCalculationDriver.getCurrentChild(), label);
+  }
+
+  @Override
+  public ResourceUnitCapacityType getCapacityType() {
+    return ResourceUnitCapacityType.ABSOLUTE;
+  }
+
+}

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

@@ -114,6 +114,7 @@ public enum CapacityConfigType {
       CapacityConfigType.NONE;
 
   protected Map<String, QueueCapacityVector> configuredCapacityVectors;
+  protected Map<String, QueueCapacityVector> configuredMaxCapacityVectors;
 
   private final RecordFactory recordFactory =
       RecordFactoryProvider.getRecordFactory(null);
@@ -199,6 +200,20 @@ public float getAbsoluteCapacity() {
     return queueCapacities.getAbsoluteCapacity();
   }
 
+  @Override
+  public ResourceVector getOrCreateAbsoluteMinCapacityVector(String label) {
+    usageTracker.getAbsoluteMinCapacityVector().putIfAbsent(label, ResourceVector.newInstance());
+
+    return usageTracker.getAbsoluteMinCapacityVector().get(label);
+  }
+
+  @Override
+  public ResourceVector getOrCreateAbsoluteMaxCapacityVector(String label) {
+    usageTracker.getAbsoluteMaxCapacityVector().putIfAbsent(label, ResourceVector.newInstance());
+
+    return usageTracker.getAbsoluteMaxCapacityVector().get(label);
+  }
+
   @Override
   public float getAbsoluteMaximumCapacity() {
     return queueCapacities.getAbsoluteMaximumCapacity();
@@ -373,7 +388,8 @@ protected void setupQueueConfigs(Resource clusterResource) throws
       this.configuredCapacityVectors = configuration
           .parseConfiguredResourceVector(queuePath.getFullPath(),
               this.queueNodeLabelsSettings.getConfiguredNodeLabels());
-
+      this.configuredMaxCapacityVectors = configuration
+          .parseConfiguredMaximumCapacityVector(queuePath.getFullPath(), this.queueNodeLabelsSettings.getConfiguredNodeLabels(), QueueCapacityVector.newInstance());
       // Update metrics
       CSQueueUtils.updateQueueStatistics(resourceCalculator, clusterResource,
           this, labelManager, null);
@@ -519,7 +535,8 @@ private void validateMinResourceIsNotGreaterThanMaxResource(Resource minResource
   private void validateAbsoluteVsPercentageCapacityConfig(
       CapacityConfigType localType) {
     if (!queuePath.isRoot()
-        && !this.capacityConfigType.equals(localType)) {
+        && !this.capacityConfigType.equals(localType) &&
+        queueContext.getConfiguration().isLegacyQueueMode()) {
       throw new IllegalArgumentException("Queue '" + getQueuePath()
           + "' should use either percentage based capacity"
           + " configuration or absolute resource.");
@@ -558,11 +575,25 @@ public Resource getEffectiveMaxCapacityDown(String label, Resource factor) {
   }
 
   @Override
-  public QueueCapacityVector getConfiguredCapacityVector(
-      String label) {
+  public QueueCapacityVector getConfiguredCapacityVector(String label) {
     return configuredCapacityVectors.get(label);
   }
 
+  @Override
+  public QueueCapacityVector getConfiguredMaxCapacityVector(String label) {
+    return configuredMaxCapacityVectors.get(label);
+  }
+
+  @Override
+  public void setConfiguredMinCapacityVector(String label, QueueCapacityVector minCapacityVector) {
+    configuredCapacityVectors.put(label, minCapacityVector);
+  }
+
+  @Override
+  public void setConfiguredMaxCapacityVector(String label, QueueCapacityVector maxCapacityVector) {
+    configuredMaxCapacityVectors.put(label, maxCapacityVector);
+  }
+
   private void initializeQueueState() {
     QueueState previousState = getState();
     QueueState configuredState = queueContext.getConfiguration()
@@ -789,6 +820,11 @@ public ReentrantReadWriteLock.ReadLock getReadLock() {
     return readLock;
   }
 
+  @Override
+  public ReentrantReadWriteLock.WriteLock getWriteLock() {
+    return writeLock;
+  }
+
   private Resource getCurrentLimitResource(String nodePartition,
       Resource clusterResource, ResourceLimits currentResourceLimits,
       SchedulingMode schedulingMode) {
@@ -916,6 +952,11 @@ boolean canAssignToThisQueue(Resource clusterResource,
 
   }
 
+  @Override
+  public Set<String> getConfiguredNodeLabels() {
+    return queueNodeLabelsSettings.getConfiguredNodeLabels();
+  }
+
   private static String ensurePartition(String partition) {
     return Optional.ofNullable(partition).orElse(RMNodeLabelsManager.NO_LABEL);
   }

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

@@ -86,6 +86,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager.NO_LABEL;
+import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacityVector.ResourceUnitCapacityType.PERCENTAGE;
+
 public class AbstractLeafQueue extends AbstractCSQueue {
   private static final Logger LOG =
       LoggerFactory.getLogger(AbstractLeafQueue.class);
@@ -1919,6 +1922,48 @@ private void updateCurrentResourceLimits(
         currentResourceLimits.getLimit()));
   }
 
+  @Override
+  public void refreshAfterResourceCalculation(Resource clusterResource, ResourceLimits resourceLimits) {
+    lastClusterResource = clusterResource;
+    // Update maximum applications for the queue and for users
+    updateMaximumApplications();
+
+    updateCurrentResourceLimits(resourceLimits, clusterResource);
+
+    // Update headroom info based on new cluster resource value
+    // absoluteMaxCapacity now,  will be replaced with absoluteMaxAvailCapacity
+    // during allocation
+    setQueueResourceLimitsInfo(clusterResource);
+
+    // Update user consumedRatios
+    recalculateQueueUsageRatio(clusterResource, null);
+
+    // Update metrics
+    CSQueueUtils.updateQueueStatistics(resourceCalculator, clusterResource,
+        this, labelManager, null);
+    // Update configured capacity/max-capacity for default partition only
+    CSQueueUtils.updateConfiguredCapacityMetrics(resourceCalculator,
+        labelManager.getResourceByLabel(null, clusterResource),
+        NO_LABEL, this);
+
+    // queue metrics are updated, more resource may be available
+    // activate the pending applications if possible
+    activateApplications();
+
+    // In case of any resource change, invalidate recalculateULCount to clear
+    // the computed user-limit.
+    usersManager.userLimitNeedsRecompute();
+
+    // Update application properties
+    for (FiCaSchedulerApp application : orderingPolicy
+        .getSchedulableEntities()) {
+      computeUserLimitAndSetHeadroom(application, clusterResource,
+          NO_LABEL,
+          SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null);
+
+    }
+  }
+
   @Override
   public void updateClusterResource(Resource clusterResource,
       ResourceLimits currentResourceLimits) {
@@ -2207,11 +2252,14 @@ public Map<String, TreeSet<RMContainer>> getIgnoreExclusivityRMContainers() {
     }
   }
 
+
   public void setCapacity(float capacity) {
+    configuredCapacityVectors.put(NO_LABEL, QueueCapacityVector.of(capacity * 100, PERCENTAGE));
     queueCapacities.setCapacity(capacity);
   }
 
   public void setCapacity(String nodeLabel, float capacity) {
+    configuredCapacityVectors.put(nodeLabel, QueueCapacityVector.of(capacity * 100, PERCENTAGE));
     queueCapacities.setCapacity(nodeLabel, capacity);
   }