| /** |
| * 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.capacity; |
| |
| import org.apache.hadoop.ipc.WeightedTimeCostProvider; |
| import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting; |
| import org.apache.hadoop.thirdparty.com.google.common.base.Strings; |
| import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableSet; |
| import org.apache.hadoop.yarn.server.resourcemanager.placement.MappingRule; |
| import org.apache.hadoop.yarn.server.resourcemanager.placement.QueuePlacementRuleUtils; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.placement.MappingRuleCreator; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.security.authorize.AccessControlList; |
| import org.apache.hadoop.util.ReflectionUtils; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.hadoop.yarn.api.records.Priority; |
| import org.apache.hadoop.yarn.api.records.QueueACL; |
| import org.apache.hadoop.yarn.api.records.QueueState; |
| import org.apache.hadoop.yarn.api.records.ReservationACL; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.ResourceInformation; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; |
| import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager; |
| import org.apache.hadoop.yarn.security.AccessType; |
| import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.placement.QueueMapping; |
| import org.apache.hadoop.yarn.server.resourcemanager.placement.QueueMapping.QueueMappingBuilder; |
| import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSchedulerConfiguration; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.AppPriorityACLConfigurationParser.AppPriorityACLKeyType; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.WorkflowPriorityMappingsManager.WorkflowPriorityMapping; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.policy.PriorityUtilizationQueueOrderingPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.policy.QueueOrderingPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.MultiNodeLookupPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.MultiNodePolicySpec; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FairOrderingPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FifoOrderingPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FifoOrderingPolicyForPendingApps; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FifoOrderingPolicyWithExclusivePartitions; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPolicy; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.SchedulableEntity; |
| import org.apache.hadoop.yarn.util.UnitsConversionUtil; |
| import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator; |
| import org.apache.hadoop.yarn.util.resource.ResourceCalculator; |
| import org.apache.hadoop.yarn.util.resource.ResourceUtils; |
| import org.apache.hadoop.yarn.util.resource.Resources; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| import java.util.Set; |
| |
| public class CapacitySchedulerConfiguration extends ReservationSchedulerConfiguration { |
| |
| private static final Logger LOG = |
| LoggerFactory.getLogger(CapacitySchedulerConfiguration.class); |
| |
| private static final String CS_CONFIGURATION_FILE = "capacity-scheduler.xml"; |
| |
| @Private |
| public static final String PREFIX = "yarn.scheduler.capacity."; |
| |
| @Private |
| public static final String DOT = "."; |
| |
| @Private |
| public static final String MAXIMUM_APPLICATIONS_SUFFIX = |
| "maximum-applications"; |
| |
| @Private |
| public static final String MAXIMUM_SYSTEM_APPLICATIONS = |
| PREFIX + MAXIMUM_APPLICATIONS_SUFFIX; |
| |
| @Private |
| public static final String MAXIMUM_AM_RESOURCE_SUFFIX = |
| "maximum-am-resource-percent"; |
| |
| @Private |
| public static final String MAXIMUM_APPLICATION_MASTERS_RESOURCE_PERCENT = |
| PREFIX + MAXIMUM_AM_RESOURCE_SUFFIX; |
| |
| @Private |
| public static final String QUEUES = "queues"; |
| |
| @Private |
| public static final String CAPACITY = "capacity"; |
| |
| @Private |
| public static final String MAXIMUM_CAPACITY = "maximum-capacity"; |
| |
| @Private |
| public static final String USER_LIMIT = "minimum-user-limit-percent"; |
| |
| @Private |
| public static final String USER_LIMIT_FACTOR = "user-limit-factor"; |
| |
| @Private |
| public static final String USER_WEIGHT = "weight"; |
| |
| @Private |
| public static final String USER_SETTINGS = "user-settings"; |
| |
| @Private |
| public static final float DEFAULT_USER_WEIGHT = 1.0f; |
| |
| @Private |
| public static final String STATE = "state"; |
| |
| @Private |
| public static final String ACCESSIBLE_NODE_LABELS = "accessible-node-labels"; |
| |
| @Private |
| public static final String DEFAULT_NODE_LABEL_EXPRESSION = |
| "default-node-label-expression"; |
| |
| public static final String RESERVE_CONT_LOOK_ALL_NODES = PREFIX |
| + "reservations-continue-look-all-nodes"; |
| |
| @Private |
| public static final boolean DEFAULT_RESERVE_CONT_LOOK_ALL_NODES = true; |
| |
| @Private |
| public static final String MAXIMUM_ALLOCATION = "maximum-allocation"; |
| |
| @Private |
| public static final String MAXIMUM_ALLOCATION_MB = "maximum-allocation-mb"; |
| |
| @Private |
| public static final String MAXIMUM_ALLOCATION_VCORES = |
| "maximum-allocation-vcores"; |
| /** |
| * Ordering policy of queues |
| */ |
| public static final String ORDERING_POLICY = "ordering-policy"; |
| |
| /* |
| * Ordering policy inside a leaf queue to sort apps |
| */ |
| public static final String FIFO_APP_ORDERING_POLICY = "fifo"; |
| |
| public static final String FAIR_APP_ORDERING_POLICY = "fair"; |
| |
| public static final String FIFO_WITH_PARTITIONS_APP_ORDERING_POLICY |
| = "fifo-with-partitions"; |
| |
| public static final String FIFO_FOR_PENDING_APPS |
| = "fifo-for-pending-apps"; |
| |
| public static final String DEFAULT_APP_ORDERING_POLICY = |
| FIFO_APP_ORDERING_POLICY; |
| |
| @Private |
| public static final int DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS = 10000; |
| |
| @Private |
| public static final float |
| DEFAULT_MAXIMUM_APPLICATIONMASTERS_RESOURCE_PERCENT = 0.1f; |
| |
| @Private |
| public static final float UNDEFINED = -1; |
| |
| @Private |
| public static final float MINIMUM_CAPACITY_VALUE = 0; |
| |
| @Private |
| public static final float MAXIMUM_CAPACITY_VALUE = 100; |
| |
| @Private |
| public static final float DEFAULT_MAXIMUM_CAPACITY_VALUE = -1.0f; |
| |
| @Private |
| public static final int DEFAULT_USER_LIMIT = 100; |
| |
| @Private |
| public static final float DEFAULT_USER_LIMIT_FACTOR = 1.0f; |
| |
| @Private |
| public static final String ALL_ACL = "*"; |
| |
| @Private |
| public static final String NONE_ACL = " "; |
| |
| @Private public static final String ENABLE_USER_METRICS = |
| PREFIX +"user-metrics.enable"; |
| @Private public static final boolean DEFAULT_ENABLE_USER_METRICS = false; |
| |
| /** ResourceComparator for scheduling. */ |
| @Private public static final String RESOURCE_CALCULATOR_CLASS = |
| PREFIX + "resource-calculator"; |
| |
| @Private public static final Class<? extends ResourceCalculator> |
| DEFAULT_RESOURCE_CALCULATOR_CLASS = DefaultResourceCalculator.class; |
| |
| @Private |
| public static final String ROOT = "root"; |
| |
| @Private |
| public static final String NODE_LOCALITY_DELAY = |
| PREFIX + "node-locality-delay"; |
| |
| @Private |
| public static final int DEFAULT_NODE_LOCALITY_DELAY = 40; |
| |
| @Private |
| public static final String RACK_LOCALITY_ADDITIONAL_DELAY = |
| PREFIX + "rack-locality-additional-delay"; |
| |
| @Private |
| public static final int DEFAULT_RACK_LOCALITY_ADDITIONAL_DELAY = -1; |
| |
| @Private |
| public static final String RACK_LOCALITY_FULL_RESET = |
| PREFIX + "rack-locality-full-reset"; |
| |
| @Private |
| public static final int DEFAULT_OFFSWITCH_PER_HEARTBEAT_LIMIT = 1; |
| |
| @Private |
| public static final String OFFSWITCH_PER_HEARTBEAT_LIMIT = |
| PREFIX + "per-node-heartbeat.maximum-offswitch-assignments"; |
| |
| @Private |
| public static final boolean DEFAULT_RACK_LOCALITY_FULL_RESET = true; |
| |
| @Private |
| public static final String SCHEDULE_ASYNCHRONOUSLY_PREFIX = |
| PREFIX + "schedule-asynchronously"; |
| |
| @Private |
| public static final String SCHEDULE_ASYNCHRONOUSLY_ENABLE = |
| SCHEDULE_ASYNCHRONOUSLY_PREFIX + ".enable"; |
| |
| @Private |
| public static final String SCHEDULE_ASYNCHRONOUSLY_MAXIMUM_THREAD = |
| SCHEDULE_ASYNCHRONOUSLY_PREFIX + ".maximum-threads"; |
| |
| @Private |
| public static final String SCHEDULE_ASYNCHRONOUSLY_MAXIMUM_PENDING_BACKLOGS = |
| SCHEDULE_ASYNCHRONOUSLY_PREFIX + ".maximum-pending-backlogs"; |
| |
| @Private |
| public static final String APP_FAIL_FAST = PREFIX + "application.fail-fast"; |
| |
| @Private |
| public static final boolean DEFAULT_APP_FAIL_FAST = false; |
| |
| @Private |
| public static final Integer |
| DEFAULT_SCHEDULE_ASYNCHRONOUSLY_MAXIMUM_PENDING_BACKLOGS = 100; |
| |
| @Private |
| public static final boolean DEFAULT_SCHEDULE_ASYNCHRONOUSLY_ENABLE = false; |
| |
| @Private |
| public static final String QUEUE_MAPPING = PREFIX + "queue-mappings"; |
| |
| @Private |
| public static final String QUEUE_MAPPING_NAME = |
| YarnConfiguration.QUEUE_PLACEMENT_RULES + ".app-name"; |
| |
| @Private |
| public static final String ENABLE_QUEUE_MAPPING_OVERRIDE = QUEUE_MAPPING + "-override.enable"; |
| |
| @Private |
| public static final boolean DEFAULT_ENABLE_QUEUE_MAPPING_OVERRIDE = false; |
| |
| @Private |
| public static final String WORKFLOW_PRIORITY_MAPPINGS = |
| PREFIX + "workflow-priority-mappings"; |
| |
| @Private |
| public static final String ENABLE_WORKFLOW_PRIORITY_MAPPINGS_OVERRIDE = |
| WORKFLOW_PRIORITY_MAPPINGS + "-override.enable"; |
| |
| @Private |
| public static final boolean DEFAULT_ENABLE_WORKFLOW_PRIORITY_MAPPINGS_OVERRIDE = false; |
| |
| @Private |
| public static final String QUEUE_PREEMPTION_DISABLED = "disable_preemption"; |
| |
| @Private |
| public static final String DEFAULT_APPLICATION_PRIORITY = "default-application-priority"; |
| |
| @Private |
| public static final Integer DEFAULT_CONFIGURATION_APPLICATION_PRIORITY = 0; |
| |
| @Private |
| public static final String AVERAGE_CAPACITY = "average-capacity"; |
| |
| @Private |
| public static final String IS_RESERVABLE = "reservable"; |
| |
| @Private |
| public static final String RESERVATION_WINDOW = "reservation-window"; |
| |
| @Private |
| public static final String INSTANTANEOUS_MAX_CAPACITY = |
| "instantaneous-max-capacity"; |
| |
| @Private |
| public static final String RESERVATION_ADMISSION_POLICY = |
| "reservation-policy"; |
| |
| @Private |
| public static final String RESERVATION_AGENT_NAME = "reservation-agent"; |
| |
| @Private |
| public static final String RESERVATION_SHOW_RESERVATION_AS_QUEUE = |
| "show-reservations-as-queues"; |
| |
| @Private |
| public static final String RESERVATION_PLANNER_NAME = "reservation-planner"; |
| |
| @Private |
| public static final String RESERVATION_MOVE_ON_EXPIRY = |
| "reservation-move-on-expiry"; |
| |
| @Private |
| public static final String RESERVATION_ENFORCEMENT_WINDOW = |
| "reservation-enforcement-window"; |
| |
| @Private |
| public static final String LAZY_PREEMPTION_ENABLED = |
| PREFIX + "lazy-preemption-enabled"; |
| |
| @Private |
| public static final boolean DEFAULT_LAZY_PREEMPTION_ENABLED = false; |
| |
| @Private |
| public static final String ASSIGN_MULTIPLE_ENABLED = PREFIX |
| + "per-node-heartbeat.multiple-assignments-enabled"; |
| |
| @Private |
| public static final boolean DEFAULT_ASSIGN_MULTIPLE_ENABLED = true; |
| |
| /** Maximum number of containers to assign on each check-in. */ |
| @Private |
| public static final String MAX_ASSIGN_PER_HEARTBEAT = PREFIX |
| + "per-node-heartbeat.maximum-container-assignments"; |
| |
| /** |
| * Avoid potential risk that greedy assign multiple may involve |
| * */ |
| @Private |
| public static final int DEFAULT_MAX_ASSIGN_PER_HEARTBEAT = 100; |
| |
| /** Configuring absolute min/max resources in a queue. **/ |
| @Private |
| public static final String MINIMUM_RESOURCE = "min-resource"; |
| |
| @Private |
| public static final String MAXIMUM_RESOURCE = "max-resource"; |
| |
| public static final String DEFAULT_RESOURCE_TYPES = "memory,vcores"; |
| |
| public static final String PATTERN_FOR_ABSOLUTE_RESOURCE = "^\\[[\\w\\.,\\-_=\\ /]+\\]$"; |
| |
| public static final Pattern RESOURCE_PATTERN = Pattern.compile(PATTERN_FOR_ABSOLUTE_RESOURCE); |
| |
| private static final String WEIGHT_SUFFIX = "w"; |
| |
| public static final String MAX_PARALLEL_APPLICATIONS = "max-parallel-apps"; |
| |
| public static final int DEFAULT_MAX_PARALLEL_APPLICATIONS = Integer.MAX_VALUE; |
| |
| public static final String ALLOW_ZERO_CAPACITY_SUM = |
| "allow-zero-capacity-sum"; |
| |
| public static final boolean DEFAULT_ALLOW_ZERO_CAPACITY_SUM = false; |
| public static final String MAPPING_RULE_FORMAT = |
| PREFIX + "mapping-rule-format"; |
| public static final String MAPPING_RULE_JSON = |
| PREFIX + "mapping-rule-json"; |
| public static final String MAPPING_RULE_JSON_FILE = |
| PREFIX + "mapping-rule-json-file"; |
| |
| public static final String MAPPING_RULE_FORMAT_LEGACY = "legacy"; |
| public static final String MAPPING_RULE_FORMAT_JSON = "json"; |
| |
| public static final String MAPPING_RULE_FORMAT_DEFAULT = |
| MAPPING_RULE_FORMAT_LEGACY; |
| |
| /** |
| * Different resource types supported. |
| */ |
| public enum AbsoluteResourceType { |
| MEMORY, VCORES; |
| } |
| |
| AppPriorityACLConfigurationParser priorityACLConfig = new AppPriorityACLConfigurationParser(); |
| |
| public CapacitySchedulerConfiguration() { |
| this(new Configuration()); |
| } |
| |
| public CapacitySchedulerConfiguration(Configuration configuration) { |
| this(configuration, true); |
| } |
| |
| public CapacitySchedulerConfiguration(Configuration configuration, |
| boolean useLocalConfigurationProvider) { |
| super(configuration); |
| if (useLocalConfigurationProvider) { |
| addResource(CS_CONFIGURATION_FILE); |
| } |
| } |
| |
| public static String getQueuePrefix(String queue) { |
| String queueName = PREFIX + queue + DOT; |
| return queueName; |
| } |
| |
| static String getQueueOrderingPolicyPrefix(String queue) { |
| String queueName = PREFIX + queue + DOT + ORDERING_POLICY + DOT; |
| return queueName; |
| } |
| |
| static String getUserPrefix(String user) { |
| return PREFIX + "user." + user + DOT; |
| } |
| |
| private String getNodeLabelPrefix(String queue, String label) { |
| if (label.equals(CommonNodeLabelsManager.NO_LABEL)) { |
| return getQueuePrefix(queue); |
| } |
| return getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label + DOT; |
| } |
| |
| public int getMaximumSystemApplications() { |
| int maxApplications = |
| getInt(MAXIMUM_SYSTEM_APPLICATIONS, DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS); |
| return maxApplications; |
| } |
| |
| public float getMaximumApplicationMasterResourcePercent() { |
| return getFloat(MAXIMUM_APPLICATION_MASTERS_RESOURCE_PERCENT, |
| DEFAULT_MAXIMUM_APPLICATIONMASTERS_RESOURCE_PERCENT); |
| } |
| |
| |
| /** |
| * Get the maximum applications per queue setting. |
| * @param queue name of the queue |
| * @return setting specified or -1 if not set |
| */ |
| public int getMaximumApplicationsPerQueue(String queue) { |
| int maxApplicationsPerQueue = |
| getInt(getQueuePrefix(queue) + MAXIMUM_APPLICATIONS_SUFFIX, |
| (int)UNDEFINED); |
| return maxApplicationsPerQueue; |
| } |
| |
| /** |
| * Get the maximum am resource percent per queue setting. |
| * @param queue name of the queue |
| * @return per queue setting or defaults to the global am-resource-percent |
| * setting if per queue setting not present |
| */ |
| public float getMaximumApplicationMasterResourcePerQueuePercent(String queue) { |
| return getFloat(getQueuePrefix(queue) + MAXIMUM_AM_RESOURCE_SUFFIX, |
| getMaximumApplicationMasterResourcePercent()); |
| } |
| |
| public void setMaximumApplicationMasterResourcePerQueuePercent(String queue, |
| float percent) { |
| setFloat(getQueuePrefix(queue) + MAXIMUM_AM_RESOURCE_SUFFIX, percent); |
| } |
| |
| private void throwExceptionForUnexpectedWeight(float weight, String queue, |
| String label) { |
| if ((weight < -1e-6 && Math.abs(weight + 1) > 1e-6) || weight > 10000) { |
| throw new IllegalArgumentException( |
| "Illegal " + "weight=" + weight + " for queue=" + queue + "label=" |
| + label |
| + ". Acceptable values: [0, 10000], -1 is same as not set"); |
| } |
| } |
| |
| public float getNonLabeledQueueWeight(String queue) { |
| String configuredValue = get(getQueuePrefix(queue) + CAPACITY); |
| float weight = extractFloatValueFromWeightConfig(configuredValue); |
| throwExceptionForUnexpectedWeight(weight, queue, ""); |
| return weight; |
| } |
| |
| public void setNonLabeledQueueWeight(String queue, float weight) { |
| set(getQueuePrefix(queue) + CAPACITY, weight + WEIGHT_SUFFIX); |
| } |
| |
| public void setLabeledQueueWeight(String queue, String label, float weight) { |
| set(getNodeLabelPrefix(queue, label) + CAPACITY, weight + WEIGHT_SUFFIX); |
| } |
| |
| public float getLabeledQueueWeight(String queue, String label) { |
| String configuredValue = get(getNodeLabelPrefix(queue, label) + CAPACITY); |
| float weight = extractFloatValueFromWeightConfig(configuredValue); |
| throwExceptionForUnexpectedWeight(weight, queue, label); |
| return weight; |
| } |
| |
| public float getNonLabeledQueueCapacity(String queue) { |
| String configuredCapacity = get(getQueuePrefix(queue) + CAPACITY); |
| boolean absoluteResourceConfigured = (configuredCapacity != null) |
| && RESOURCE_PATTERN.matcher(configuredCapacity).find(); |
| if (absoluteResourceConfigured || configuredWeightAsCapacity( |
| configuredCapacity)) { |
| // Return capacity in percentage as 0 for non-root queues and 100 for |
| // root.From AbstractCSQueue, absolute resource will be parsed and |
| // updated. Once nodes are added/removed in cluster, capacity in |
| // percentage will also be re-calculated. |
| return queue.equals("root") ? 100.0f : 0f; |
| } |
| |
| float capacity = queue.equals("root") |
| ? 100.0f |
| : (configuredCapacity == null) |
| ? 0f |
| : Float.parseFloat(configuredCapacity); |
| if (capacity < MINIMUM_CAPACITY_VALUE |
| || capacity > MAXIMUM_CAPACITY_VALUE) { |
| throw new IllegalArgumentException( |
| "Illegal " + "capacity of " + capacity + " for queue " + queue); |
| } |
| LOG.debug("CSConf - getCapacity: queuePrefix={}, capacity={}", |
| getQueuePrefix(queue), capacity); |
| |
| return capacity; |
| } |
| |
| public void setCapacity(String queue, float capacity) { |
| if (queue.equals("root")) { |
| throw new IllegalArgumentException( |
| "Cannot set capacity, root queue has a fixed capacity of 100.0f"); |
| } |
| setFloat(getQueuePrefix(queue) + CAPACITY, capacity); |
| LOG.debug("CSConf - setCapacity: queuePrefix={}, capacity={}", |
| getQueuePrefix(queue), capacity); |
| |
| } |
| |
| @VisibleForTesting |
| public void setCapacity(String queue, String absoluteResourceCapacity) { |
| if (queue.equals("root")) { |
| throw new IllegalArgumentException( |
| "Cannot set capacity, root queue has a fixed capacity"); |
| } |
| set(getQueuePrefix(queue) + CAPACITY, absoluteResourceCapacity); |
| LOG.debug("CSConf - setCapacity: queuePrefix={}, capacity={}", |
| getQueuePrefix(queue), absoluteResourceCapacity); |
| |
| } |
| |
| public float getNonLabeledQueueMaximumCapacity(String queue) { |
| String configuredCapacity = get(getQueuePrefix(queue) + MAXIMUM_CAPACITY); |
| boolean matcher = (configuredCapacity != null) |
| && RESOURCE_PATTERN.matcher(configuredCapacity).find(); |
| if (matcher) { |
| // Return capacity in percentage as 0 for non-root queues and 100 for |
| // root.From AbstractCSQueue, absolute resource will be parsed and |
| // updated. Once nodes are added/removed in cluster, capacity in |
| // percentage will also be re-calculated. |
| return 100.0f; |
| } |
| |
| float maxCapacity = (configuredCapacity == null) |
| ? MAXIMUM_CAPACITY_VALUE |
| : Float.parseFloat(configuredCapacity); |
| maxCapacity = (maxCapacity == DEFAULT_MAXIMUM_CAPACITY_VALUE) |
| ? MAXIMUM_CAPACITY_VALUE |
| : maxCapacity; |
| return maxCapacity; |
| } |
| |
| public void setMaximumCapacity(String queue, float maxCapacity) { |
| if (maxCapacity > MAXIMUM_CAPACITY_VALUE) { |
| throw new IllegalArgumentException("Illegal " + |
| "maximum-capacity of " + maxCapacity + " for queue " + queue); |
| } |
| setFloat(getQueuePrefix(queue) + MAXIMUM_CAPACITY, maxCapacity); |
| LOG.debug("CSConf - setMaxCapacity: queuePrefix={}, maxCapacity={}", |
| getQueuePrefix(queue), maxCapacity); |
| } |
| |
| public void setCapacityByLabel(String queue, String label, float capacity) { |
| setFloat(getNodeLabelPrefix(queue, label) + CAPACITY, capacity); |
| } |
| |
| @VisibleForTesting |
| public void setCapacityByLabel(String queue, String label, |
| String absoluteResourceCapacity) { |
| set(getNodeLabelPrefix(queue, label) + CAPACITY, absoluteResourceCapacity); |
| } |
| |
| 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, |
| DEFAULT_USER_LIMIT); |
| return userLimit; |
| } |
| |
| // TODO (wangda): We need to better distinguish app ordering policy and queue |
| // ordering policy's classname / configuration options, etc. And dedup code |
| // if possible. |
| @SuppressWarnings("unchecked") |
| public <S extends SchedulableEntity> OrderingPolicy<S> getAppOrderingPolicy( |
| String queue) { |
| |
| String policyType = get(getQueuePrefix(queue) + ORDERING_POLICY, |
| DEFAULT_APP_ORDERING_POLICY); |
| |
| OrderingPolicy<S> orderingPolicy; |
| |
| if (policyType.trim().equals(FIFO_APP_ORDERING_POLICY)) { |
| policyType = FifoOrderingPolicy.class.getName(); |
| } |
| if (policyType.trim().equals(FAIR_APP_ORDERING_POLICY)) { |
| policyType = FairOrderingPolicy.class.getName(); |
| } |
| if (policyType.trim().equals(FIFO_WITH_PARTITIONS_APP_ORDERING_POLICY)) { |
| policyType = FifoOrderingPolicyWithExclusivePartitions.class.getName(); |
| } |
| if (policyType.trim().equals(FIFO_FOR_PENDING_APPS)) { |
| policyType = FifoOrderingPolicyForPendingApps.class.getName(); |
| } |
| |
| try { |
| orderingPolicy = (OrderingPolicy<S>) |
| Class.forName(policyType).newInstance(); |
| } catch (Exception e) { |
| String message = "Unable to construct ordering policy for: " + policyType + ", " + e.getMessage(); |
| throw new RuntimeException(message, e); |
| } |
| |
| Map<String, String> config = new HashMap<String, String>(); |
| String confPrefix = getQueuePrefix(queue) + ORDERING_POLICY + "."; |
| for (Map.Entry<String, String> kv : this) { |
| if (kv.getKey().startsWith(confPrefix)) { |
| config.put(kv.getKey().substring(confPrefix.length()), kv.getValue()); |
| } |
| } |
| orderingPolicy.configure(config); |
| return orderingPolicy; |
| } |
| |
| public void setUserLimit(String queue, int userLimit) { |
| setInt(getQueuePrefix(queue) + USER_LIMIT, userLimit); |
| LOG.debug("here setUserLimit: queuePrefix={}, userLimit={}", |
| getQueuePrefix(queue), getUserLimit(queue)); |
| } |
| |
| public float getUserLimitFactor(String queue) { |
| float userLimitFactor = |
| getFloat(getQueuePrefix(queue) + USER_LIMIT_FACTOR, |
| DEFAULT_USER_LIMIT_FACTOR); |
| return userLimitFactor; |
| } |
| |
| public void setUserLimitFactor(String queue, float userLimitFactor) { |
| setFloat(getQueuePrefix(queue) + USER_LIMIT_FACTOR, userLimitFactor); |
| } |
| |
| public QueueState getConfiguredState(String queue) { |
| String state = get(getQueuePrefix(queue) + STATE); |
| if (state == null) { |
| return null; |
| } else { |
| return QueueState.valueOf(StringUtils.toUpperCase(state)); |
| } |
| } |
| |
| public QueueState getState(String queue) { |
| QueueState state = getConfiguredState(queue); |
| return (state == null) ? QueueState.RUNNING : state; |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setState(String queue, QueueState state) { |
| set(getQueuePrefix(queue) + STATE, state.name()); |
| } |
| |
| public void setAccessibleNodeLabels(String queue, Set<String> labels) { |
| if (labels == null) { |
| return; |
| } |
| String str = StringUtils.join(",", labels); |
| set(getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS, str); |
| } |
| |
| public Set<String> getAccessibleNodeLabels(String queue) { |
| String accessibleLabelStr = |
| get(getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS); |
| |
| // When accessible-label is null, |
| if (accessibleLabelStr == null) { |
| // Only return null when queue is not ROOT |
| if (!queue.equals(ROOT)) { |
| return null; |
| } |
| } else { |
| // print a warning when accessibleNodeLabel specified in config and queue |
| // is ROOT |
| if (queue.equals(ROOT)) { |
| LOG.warn("Accessible node labels for root queue will be ignored," |
| + " it will be automatically set to \"*\"."); |
| } |
| } |
| |
| // always return ANY for queue root |
| if (queue.equals(ROOT)) { |
| return ImmutableSet.of(RMNodeLabelsManager.ANY); |
| } |
| |
| // In other cases, split the accessibleLabelStr by "," |
| Set<String> set = new HashSet<String>(); |
| for (String str : accessibleLabelStr.split(",")) { |
| if (!str.trim().isEmpty()) { |
| set.add(str.trim()); |
| } |
| } |
| |
| // if labels contains "*", only keep ANY behind |
| if (set.contains(RMNodeLabelsManager.ANY)) { |
| set.clear(); |
| set.add(RMNodeLabelsManager.ANY); |
| } |
| return Collections.unmodifiableSet(set); |
| } |
| |
| private boolean configuredWeightAsCapacity(String configureValue) { |
| if (configureValue == null) { |
| return false; |
| } |
| return configureValue.endsWith(WEIGHT_SUFFIX); |
| } |
| |
| private float extractFloatValueFromWeightConfig(String configureValue) { |
| if (!configuredWeightAsCapacity(configureValue)) { |
| return -1f; |
| } else { |
| return Float.valueOf( |
| configureValue.substring(0, configureValue.indexOf(WEIGHT_SUFFIX))); |
| } |
| } |
| |
| private float internalGetLabeledQueueCapacity(String queue, String label, |
| String suffix, float defaultValue) { |
| String capacityPropertyName = getNodeLabelPrefix(queue, label) + suffix; |
| String configuredCapacity = get(capacityPropertyName); |
| boolean absoluteResourceConfigured = |
| (configuredCapacity != null) && RESOURCE_PATTERN.matcher( |
| configuredCapacity).find(); |
| if (absoluteResourceConfigured || configuredWeightAsCapacity( |
| configuredCapacity)) { |
| // Return capacity in percentage as 0 for non-root queues and 100 for |
| // root.From AbstractCSQueue, absolute resource, and weight will be parsed |
| // and updated separately. Once nodes are added/removed in cluster, |
| // capacity is percentage will also be re-calculated. |
| return defaultValue; |
| } |
| |
| float capacity = getFloat(capacityPropertyName, defaultValue); |
| if (capacity < MINIMUM_CAPACITY_VALUE |
| || capacity > MAXIMUM_CAPACITY_VALUE) { |
| throw new IllegalArgumentException( |
| "Illegal capacity of " + capacity + " for node-label=" + label |
| + " in queue=" + queue |
| + ", valid capacity should in range of [0, 100]."); |
| } |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| "CSConf - getCapacityOfLabel: prefix=" + getNodeLabelPrefix(queue, |
| label) + ", capacity=" + capacity); |
| } |
| return capacity; |
| } |
| |
| public float getLabeledQueueCapacity(String queue, String label) { |
| return internalGetLabeledQueueCapacity(queue, label, CAPACITY, 0f); |
| } |
| |
| public float getLabeledQueueMaximumCapacity(String queue, String label) { |
| return internalGetLabeledQueueCapacity(queue, label, MAXIMUM_CAPACITY, 100f); |
| } |
| |
| public String getDefaultNodeLabelExpression(String queue) { |
| String defaultLabelExpression = get(getQueuePrefix(queue) |
| + DEFAULT_NODE_LABEL_EXPRESSION); |
| if (defaultLabelExpression == null) { |
| return null; |
| } |
| return defaultLabelExpression.trim(); |
| } |
| |
| public void setDefaultNodeLabelExpression(String queue, String exp) { |
| set(getQueuePrefix(queue) + DEFAULT_NODE_LABEL_EXPRESSION, exp); |
| } |
| |
| public float getMaximumAMResourcePercentPerPartition(String queue, |
| String label) { |
| // If per-partition max-am-resource-percent is not configured, |
| // use default value as max-am-resource-percent for this queue. |
| return getFloat(getNodeLabelPrefix(queue, label) |
| + MAXIMUM_AM_RESOURCE_SUFFIX, |
| getMaximumApplicationMasterResourcePerQueuePercent(queue)); |
| } |
| |
| public void setMaximumAMResourcePercentPerPartition(String queue, |
| String label, float percent) { |
| setFloat(getNodeLabelPrefix(queue, label) |
| + MAXIMUM_AM_RESOURCE_SUFFIX, percent); |
| } |
| |
| /* |
| * Returns whether we should continue to look at all heart beating nodes even |
| * after the reservation limit was hit. The node heart beating in could |
| * satisfy the request thus could be a better pick then waiting for the |
| * reservation to be fullfilled. This config is refreshable. |
| */ |
| public boolean getReservationContinueLook() { |
| return getBoolean(RESERVE_CONT_LOOK_ALL_NODES, |
| DEFAULT_RESERVE_CONT_LOOK_ALL_NODES); |
| } |
| |
| private static String getAclKey(QueueACL acl) { |
| return "acl_" + StringUtils.toLowerCase(acl.toString()); |
| } |
| |
| public AccessControlList getAcl(String queue, QueueACL acl) { |
| String queuePrefix = getQueuePrefix(queue); |
| // The root queue defaults to all access if not defined |
| // Sub queues inherit access if not defined |
| String defaultAcl = queue.equals(ROOT) ? ALL_ACL : NONE_ACL; |
| String aclString = get(queuePrefix + getAclKey(acl), defaultAcl); |
| return new AccessControlList(aclString); |
| } |
| |
| public void setAcl(String queue, QueueACL acl, String aclString) { |
| String queuePrefix = getQueuePrefix(queue); |
| set(queuePrefix + getAclKey(acl), aclString); |
| } |
| |
| private static String getAclKey(ReservationACL acl) { |
| return "acl_" + StringUtils.toLowerCase(acl.toString()); |
| } |
| |
| private static String getAclKey(AccessType acl) { |
| return "acl_" + StringUtils.toLowerCase(acl.toString()); |
| } |
| |
| @Override |
| public Map<ReservationACL, AccessControlList> getReservationAcls(String |
| queue) { |
| Map<ReservationACL, AccessControlList> resAcls = new HashMap<>(); |
| for (ReservationACL acl : ReservationACL.values()) { |
| resAcls.put(acl, getReservationAcl(queue, acl)); |
| } |
| return resAcls; |
| } |
| |
| private AccessControlList getReservationAcl(String queue, ReservationACL |
| acl) { |
| String queuePrefix = getQueuePrefix(queue); |
| // The root queue defaults to all access if not defined |
| // Sub queues inherit access if not defined |
| String defaultAcl = ALL_ACL; |
| String aclString = get(queuePrefix + getAclKey(acl), defaultAcl); |
| return new AccessControlList(aclString); |
| } |
| |
| private void setAcl(String queue, ReservationACL acl, String aclString) { |
| String queuePrefix = getQueuePrefix(queue); |
| set(queuePrefix + getAclKey(acl), aclString); |
| } |
| |
| private void setAcl(String queue, AccessType acl, String aclString) { |
| String queuePrefix = getQueuePrefix(queue); |
| set(queuePrefix + getAclKey(acl), aclString); |
| } |
| |
| public Map<AccessType, AccessControlList> getAcls(String queue) { |
| Map<AccessType, AccessControlList> acls = |
| new HashMap<AccessType, AccessControlList>(); |
| for (QueueACL acl : QueueACL.values()) { |
| acls.put(SchedulerUtils.toAccessType(acl), getAcl(queue, acl)); |
| } |
| return acls; |
| } |
| |
| public void setAcls(String queue, Map<QueueACL, AccessControlList> acls) { |
| for (Map.Entry<QueueACL, AccessControlList> e : acls.entrySet()) { |
| setAcl(queue, e.getKey(), e.getValue().getAclString()); |
| } |
| } |
| |
| @VisibleForTesting |
| public void setReservationAcls(String queue, |
| Map<ReservationACL, AccessControlList> acls) { |
| for (Map.Entry<ReservationACL, AccessControlList> e : acls.entrySet()) { |
| setAcl(queue, e.getKey(), e.getValue().getAclString()); |
| } |
| } |
| |
| @VisibleForTesting |
| public void setPriorityAcls(String queue, Priority priority, |
| Priority defaultPriority, String[] acls) { |
| StringBuilder aclString = new StringBuilder(); |
| |
| StringBuilder userAndGroup = new StringBuilder(); |
| for (int i = 0; i < acls.length; i++) { |
| userAndGroup.append(AppPriorityACLKeyType.values()[i] + "=" + acls[i].trim()) |
| .append(" "); |
| } |
| |
| aclString.append("[" + userAndGroup.toString().trim() + " " |
| + "max_priority=" + priority.getPriority() + " " + "default_priority=" |
| + defaultPriority.getPriority() + "]"); |
| |
| setAcl(queue, AccessType.APPLICATION_MAX_PRIORITY, aclString.toString()); |
| } |
| |
| public List<AppPriorityACLGroup> getPriorityAcls(String queue, |
| Priority clusterMaxPriority) { |
| String queuePrefix = getQueuePrefix(queue); |
| String defaultAcl = ALL_ACL; |
| String aclString = get( |
| queuePrefix + getAclKey(AccessType.APPLICATION_MAX_PRIORITY), |
| defaultAcl); |
| |
| return priorityACLConfig.getPriorityAcl(clusterMaxPriority, aclString); |
| } |
| |
| public String[] getQueues(String queue) { |
| LOG.debug("CSConf - getQueues called for: queuePrefix={}", |
| getQueuePrefix(queue)); |
| String[] queues = getStrings(getQueuePrefix(queue) + QUEUES); |
| List<String> trimmedQueueNames = new ArrayList<String>(); |
| if (null != queues) { |
| for (String s : queues) { |
| trimmedQueueNames.add(s.trim()); |
| } |
| queues = trimmedQueueNames.toArray(new String[0]); |
| } |
| |
| LOG.debug("CSConf - getQueues: queuePrefix={}, queues={}", |
| getQueuePrefix(queue), |
| ((queues == null) ? "" : StringUtils.arrayToString(queues))); |
| |
| return queues; |
| } |
| |
| public void setQueues(String queue, String[] subQueues) { |
| set(getQueuePrefix(queue) + QUEUES, StringUtils.arrayToString(subQueues)); |
| LOG.debug("CSConf - setQueues: qPrefix={}, queues={}", |
| getQueuePrefix(queue), StringUtils.arrayToString(subQueues)); |
| } |
| |
| public Resource getMinimumAllocation() { |
| int minimumMemory = getInt( |
| YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, |
| YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB); |
| int minimumCores = getInt( |
| YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, |
| YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES); |
| return Resources.createResource(minimumMemory, minimumCores); |
| } |
| |
| @Private |
| public Priority getQueuePriority(String queue) { |
| String queuePolicyPrefix = getQueuePrefix(queue); |
| Priority pri = Priority.newInstance( |
| getInt(queuePolicyPrefix + "priority", 0)); |
| return pri; |
| } |
| |
| @Private |
| public void setQueuePriority(String queue, int priority) { |
| String queuePolicyPrefix = getQueuePrefix(queue); |
| setInt(queuePolicyPrefix + "priority", priority); |
| } |
| |
| /** |
| * Get maximum_allocation setting for the specified queue from the |
| * configuration. |
| * |
| * @param queue |
| * name of the queue |
| * @return Resource object or Resource.none if not set |
| */ |
| public Resource getQueueMaximumAllocation(String queue) { |
| String queuePrefix = getQueuePrefix(queue); |
| String rawQueueMaxAllocation = get(queuePrefix + MAXIMUM_ALLOCATION, null); |
| if (Strings.isNullOrEmpty(rawQueueMaxAllocation)) { |
| return Resources.none(); |
| } else { |
| return ResourceUtils.createResourceFromString(rawQueueMaxAllocation, |
| ResourceUtils.getResourcesTypeInfo()); |
| } |
| } |
| |
| public long getQueueMaximumAllocationMb(String queue) { |
| String queuePrefix = getQueuePrefix(queue); |
| return getInt(queuePrefix + MAXIMUM_ALLOCATION_MB, (int)UNDEFINED); |
| } |
| |
| public int getQueueMaximumAllocationVcores(String queue) { |
| String queuePrefix = getQueuePrefix(queue); |
| return getInt(queuePrefix + MAXIMUM_ALLOCATION_VCORES, (int)UNDEFINED); |
| } |
| |
| public boolean getEnableUserMetrics() { |
| return getBoolean(ENABLE_USER_METRICS, DEFAULT_ENABLE_USER_METRICS); |
| } |
| |
| public int getOffSwitchPerHeartbeatLimit() { |
| int limit = getInt(OFFSWITCH_PER_HEARTBEAT_LIMIT, |
| DEFAULT_OFFSWITCH_PER_HEARTBEAT_LIMIT); |
| if (limit < 1) { |
| LOG.warn(OFFSWITCH_PER_HEARTBEAT_LIMIT + "(" + limit + ") < 1. Using 1."); |
| limit = 1; |
| } |
| return limit; |
| } |
| |
| public void setOffSwitchPerHeartbeatLimit(int limit) { |
| setInt(OFFSWITCH_PER_HEARTBEAT_LIMIT, limit); |
| } |
| |
| public int getNodeLocalityDelay() { |
| return getInt(NODE_LOCALITY_DELAY, DEFAULT_NODE_LOCALITY_DELAY); |
| } |
| |
| @VisibleForTesting |
| public void setNodeLocalityDelay(int nodeLocalityDelay) { |
| setInt(NODE_LOCALITY_DELAY, nodeLocalityDelay); |
| } |
| |
| public int getRackLocalityAdditionalDelay() { |
| return getInt(RACK_LOCALITY_ADDITIONAL_DELAY, |
| DEFAULT_RACK_LOCALITY_ADDITIONAL_DELAY); |
| } |
| |
| public boolean getRackLocalityFullReset() { |
| return getBoolean(RACK_LOCALITY_FULL_RESET, |
| DEFAULT_RACK_LOCALITY_FULL_RESET); |
| } |
| |
| public ResourceCalculator getResourceCalculator() { |
| return ReflectionUtils.newInstance( |
| getClass( |
| RESOURCE_CALCULATOR_CLASS, |
| DEFAULT_RESOURCE_CALCULATOR_CLASS, |
| ResourceCalculator.class), |
| this); |
| } |
| |
| public boolean getUsePortForNodeName() { |
| return getBoolean(YarnConfiguration.RM_SCHEDULER_INCLUDE_PORT_IN_NODE_NAME, |
| YarnConfiguration.DEFAULT_RM_SCHEDULER_USE_PORT_FOR_NODE_NAME); |
| } |
| |
| public void setResourceComparator( |
| Class<? extends ResourceCalculator> resourceCalculatorClass) { |
| setClass( |
| RESOURCE_CALCULATOR_CLASS, |
| resourceCalculatorClass, |
| ResourceCalculator.class); |
| } |
| |
| public boolean getScheduleAynschronously() { |
| return getBoolean(SCHEDULE_ASYNCHRONOUSLY_ENABLE, |
| DEFAULT_SCHEDULE_ASYNCHRONOUSLY_ENABLE); |
| } |
| |
| public void setScheduleAynschronously(boolean async) { |
| setBoolean(SCHEDULE_ASYNCHRONOUSLY_ENABLE, async); |
| } |
| |
| public boolean getOverrideWithQueueMappings() { |
| return getBoolean(ENABLE_QUEUE_MAPPING_OVERRIDE, |
| DEFAULT_ENABLE_QUEUE_MAPPING_OVERRIDE); |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setOverrideWithQueueMappings(boolean overrideWithQueueMappings) { |
| setBoolean(ENABLE_QUEUE_MAPPING_OVERRIDE, overrideWithQueueMappings); |
| } |
| |
| public List<QueueMapping> getQueueMappingEntity( |
| String queueMappingSuffix) { |
| String queueMappingName = buildQueueMappingRuleProperty(queueMappingSuffix); |
| |
| List<QueueMapping> mappings = |
| new ArrayList<QueueMapping>(); |
| Collection<String> mappingsString = |
| getTrimmedStringCollection(queueMappingName); |
| for (String mappingValue : mappingsString) { |
| String[] mapping = |
| StringUtils.getTrimmedStringCollection(mappingValue, ":") |
| .toArray(new String[] {}); |
| if (mapping.length != 2 || mapping[1].length() == 0) { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping " + mappingValue); |
| } |
| |
| //Mappings should be consistent, and have the parent path parsed |
| // from the beginning |
| QueueMapping m = QueueMapping.QueueMappingBuilder.create() |
| .type(QueueMapping.MappingType.APPLICATION) |
| .source(mapping[0]) |
| .parsePathString(mapping[1]) |
| .build(); |
| mappings.add(m); |
| } |
| |
| return mappings; |
| } |
| |
| private String buildQueueMappingRuleProperty (String queueMappingSuffix) { |
| StringBuilder queueMapping = new StringBuilder(); |
| queueMapping.append(YarnConfiguration.QUEUE_PLACEMENT_RULES) |
| .append(".").append(queueMappingSuffix); |
| return queueMapping.toString(); |
| } |
| |
| @VisibleForTesting |
| public void setQueueMappingEntities(List<QueueMapping> queueMappings, |
| String queueMappingSuffix) { |
| if (queueMappings == null) { |
| return; |
| } |
| |
| List<String> queueMappingStrs = new ArrayList<>(); |
| for (QueueMapping mapping : queueMappings) { |
| queueMappingStrs.add(mapping.toTypelessString()); |
| } |
| |
| String mappingRuleProp = buildQueueMappingRuleProperty(queueMappingSuffix); |
| setStrings(mappingRuleProp, StringUtils.join(",", queueMappingStrs)); |
| } |
| |
| public boolean getOverrideWithWorkflowPriorityMappings() { |
| return getBoolean(ENABLE_WORKFLOW_PRIORITY_MAPPINGS_OVERRIDE, |
| DEFAULT_ENABLE_WORKFLOW_PRIORITY_MAPPINGS_OVERRIDE); |
| } |
| |
| public Collection<String> getWorkflowPriorityMappings() { |
| return getTrimmedStringCollection(WORKFLOW_PRIORITY_MAPPINGS); |
| } |
| |
| /** |
| * Get user/group mappings to queues. |
| * |
| * @return user/groups mappings or null on illegal configs |
| */ |
| public List<QueueMapping> getQueueMappings() { |
| List<QueueMapping> mappings = |
| new ArrayList<QueueMapping>(); |
| Collection<String> mappingsString = |
| getTrimmedStringCollection(QUEUE_MAPPING); |
| for (String mappingValue : mappingsString) { |
| String[] mapping = |
| StringUtils.getTrimmedStringCollection(mappingValue, ":") |
| .toArray(new String[] {}); |
| if (mapping.length != 3 || mapping[1].length() == 0 |
| || mapping[2].length() == 0) { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping " + mappingValue); |
| } |
| |
| QueueMapping m; |
| try { |
| QueueMapping.MappingType mappingType; |
| if (mapping[0].equals("u")) { |
| mappingType = QueueMapping.MappingType.USER; |
| } else if (mapping[0].equals("g")) { |
| mappingType = QueueMapping.MappingType.GROUP; |
| } else { |
| throw new IllegalArgumentException( |
| "unknown mapping prefix " + mapping[0]); |
| } |
| //forcing the queue path to be split to parent and leafQueue, to make |
| //queue mapping parentPath and queueName consistent |
| m = QueueMappingBuilder.create() |
| .type(mappingType) |
| .source(mapping[1]) |
| .parsePathString(mapping[2]) |
| .build(); |
| } catch (Throwable t) { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping " + mappingValue); |
| } |
| |
| if (m != null) { |
| mappings.add(m); |
| } |
| } |
| |
| return mappings; |
| } |
| |
| public List<MappingRule> parseLegacyMappingRules() { |
| List<MappingRule> mappings = new ArrayList<MappingRule>(); |
| Collection<String> mappingsString = |
| getTrimmedStringCollection(QUEUE_MAPPING); |
| |
| for (String mappingValue : mappingsString) { |
| String[] mapping = |
| StringUtils.getTrimmedStringCollection(mappingValue, ":") |
| .toArray(new String[] {}); |
| if (mapping.length != 3 || mapping[1].length() == 0 |
| || mapping[2].length() == 0) { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping " + mappingValue); |
| } |
| |
| if (mapping[0].equals("u") || mapping[0].equals("g")) { |
| mappings.add(MappingRule.createLegacyRule( |
| mapping[0], mapping[1], mapping[2])); |
| } else { |
| throw new IllegalArgumentException( |
| "unknown mapping prefix " + mapping[0]); |
| } |
| } |
| |
| mappingsString = getTrimmedStringCollection(QUEUE_MAPPING_NAME); |
| for (String mappingValue : mappingsString) { |
| String[] mapping = |
| StringUtils.getTrimmedStringCollection(mappingValue, ":") |
| .toArray(new String[] {}); |
| if (mapping.length != 2 || mapping[1].length() == 0) { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping " + mappingValue); |
| } |
| |
| mappings.add(MappingRule.createLegacyRule(mapping[0], mapping[1])); |
| } |
| |
| return mappings; |
| } |
| |
| public List<MappingRule> parseJSONMappingRules() throws IOException { |
| String mappingJson = get(MAPPING_RULE_JSON, ""); |
| String mappingJsonFile = get(MAPPING_RULE_JSON_FILE, ""); |
| MappingRuleCreator creator = new MappingRuleCreator(); |
| |
| if (!mappingJson.equals("")) { |
| LOG.info("Reading mapping rules from provided inline JSON '{}'.", |
| mappingJson); |
| try { |
| return creator.getMappingRulesFromString(mappingJson); |
| } catch (IOException e) { |
| LOG.error("Error parsing mapping rule inline JSON."); |
| throw e; |
| } |
| } else if (!mappingJsonFile.equals("")) { |
| LOG.info("Reading mapping rules from JSON file '{}'.", |
| mappingJsonFile); |
| try { |
| return creator.getMappingRulesFromFile(mappingJsonFile.trim()); |
| } catch (IOException e) { |
| LOG.error("Error reading or parsing mapping rule JSON file '{}'.", |
| mappingJsonFile); |
| throw e; |
| } |
| } else { |
| LOG.warn("Mapping rule is set to JSON, but no inline JSON nor a JSON " + |
| "file was provided! Starting with no mapping rules!"); |
| } |
| |
| return new ArrayList<>(); |
| } |
| |
| public List<MappingRule> getMappingRules() throws IOException { |
| String mappingFormat = |
| get(MAPPING_RULE_FORMAT, MAPPING_RULE_FORMAT_DEFAULT); |
| if (mappingFormat.equals(MAPPING_RULE_FORMAT_LEGACY)) { |
| return parseLegacyMappingRules(); |
| } else if (mappingFormat.equals(MAPPING_RULE_FORMAT_JSON)) { |
| return parseJSONMappingRules(); |
| } else { |
| throw new IllegalArgumentException( |
| "Illegal queue mapping format '" + mappingFormat + "' please use '" + |
| MAPPING_RULE_FORMAT_LEGACY + "' or '" + MAPPING_RULE_FORMAT_JSON + |
| "'"); |
| } |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setQueuePlacementRules(Collection<String> queuePlacementRules) { |
| if (queuePlacementRules == null) { |
| return; |
| } |
| String str = StringUtils.join(",", queuePlacementRules); |
| setStrings(YarnConfiguration.QUEUE_PLACEMENT_RULES, str); |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setQueueMappings(List<QueueMapping> queueMappings) { |
| if (queueMappings == null) { |
| return; |
| } |
| |
| List<String> queueMappingStrs = new ArrayList<>(); |
| for (QueueMapping mapping : queueMappings) { |
| queueMappingStrs.add(mapping.toString()); |
| } |
| |
| setStrings(QUEUE_MAPPING, StringUtils.join(",", queueMappingStrs)); |
| } |
| |
| |
| @Private |
| @VisibleForTesting |
| public void setAppNameMappings(List<QueueMapping> queueMappings) { |
| if (queueMappings == null) { |
| return; |
| } |
| |
| List<String> queueMappingStrs = new ArrayList<>(); |
| for (QueueMapping mapping : queueMappings) { |
| String rule = mapping.toString(); |
| String[] parts = rule.split(":"); |
| queueMappingStrs.add(parts[1] + ":" + parts[2]); |
| } |
| |
| setStrings(QUEUE_MAPPING_NAME, StringUtils.join(",", queueMappingStrs)); |
| } |
| |
| @Private |
| @VisibleForTesting |
| void setWorkflowPriorityMappings( |
| List<WorkflowPriorityMapping> workflowPriorityMappings) { |
| setStrings(WORKFLOW_PRIORITY_MAPPINGS, WorkflowPriorityMappingsManager |
| .getWorkflowPriorityMappingStr(workflowPriorityMappings)); |
| } |
| |
| public boolean isReservable(String queue) { |
| boolean isReservable = |
| getBoolean(getQueuePrefix(queue) + IS_RESERVABLE, false); |
| return isReservable; |
| } |
| |
| public void setReservable(String queue, boolean isReservable) { |
| setBoolean(getQueuePrefix(queue) + IS_RESERVABLE, isReservable); |
| LOG.debug("here setReservableQueue: queuePrefix={}, isReservableQueue={}", |
| getQueuePrefix(queue), isReservable(queue)); |
| } |
| |
| @Override |
| public long getReservationWindow(String queue) { |
| long reservationWindow = |
| getLong(getQueuePrefix(queue) + RESERVATION_WINDOW, |
| DEFAULT_RESERVATION_WINDOW); |
| return reservationWindow; |
| } |
| |
| @Override |
| public float getAverageCapacity(String queue) { |
| float avgCapacity = |
| getFloat(getQueuePrefix(queue) + AVERAGE_CAPACITY, |
| MAXIMUM_CAPACITY_VALUE); |
| return avgCapacity; |
| } |
| |
| @Override |
| public float getInstantaneousMaxCapacity(String queue) { |
| float instMaxCapacity = |
| getFloat(getQueuePrefix(queue) + INSTANTANEOUS_MAX_CAPACITY, |
| MAXIMUM_CAPACITY_VALUE); |
| return instMaxCapacity; |
| } |
| |
| public void setInstantaneousMaxCapacity(String queue, float instMaxCapacity) { |
| setFloat(getQueuePrefix(queue) + INSTANTANEOUS_MAX_CAPACITY, |
| instMaxCapacity); |
| } |
| |
| public void setReservationWindow(String queue, long reservationWindow) { |
| setLong(getQueuePrefix(queue) + RESERVATION_WINDOW, reservationWindow); |
| } |
| |
| public void setAverageCapacity(String queue, float avgCapacity) { |
| setFloat(getQueuePrefix(queue) + AVERAGE_CAPACITY, avgCapacity); |
| } |
| |
| @Override |
| public String getReservationAdmissionPolicy(String queue) { |
| String reservationPolicy = |
| get(getQueuePrefix(queue) + RESERVATION_ADMISSION_POLICY, |
| DEFAULT_RESERVATION_ADMISSION_POLICY); |
| return reservationPolicy; |
| } |
| |
| public void setReservationAdmissionPolicy(String queue, |
| String reservationPolicy) { |
| set(getQueuePrefix(queue) + RESERVATION_ADMISSION_POLICY, reservationPolicy); |
| } |
| |
| @Override |
| public String getReservationAgent(String queue) { |
| String reservationAgent = |
| get(getQueuePrefix(queue) + RESERVATION_AGENT_NAME, |
| DEFAULT_RESERVATION_AGENT_NAME); |
| return reservationAgent; |
| } |
| |
| public void setReservationAgent(String queue, String reservationPolicy) { |
| set(getQueuePrefix(queue) + RESERVATION_AGENT_NAME, reservationPolicy); |
| } |
| |
| @Override |
| public boolean getShowReservationAsQueues(String queuePath) { |
| boolean showReservationAsQueues = |
| getBoolean(getQueuePrefix(queuePath) |
| + RESERVATION_SHOW_RESERVATION_AS_QUEUE, |
| DEFAULT_SHOW_RESERVATIONS_AS_QUEUES); |
| return showReservationAsQueues; |
| } |
| |
| @Override |
| public String getReplanner(String queue) { |
| String replanner = |
| get(getQueuePrefix(queue) + RESERVATION_PLANNER_NAME, |
| DEFAULT_RESERVATION_PLANNER_NAME); |
| return replanner; |
| } |
| |
| @Override |
| public boolean getMoveOnExpiry(String queue) { |
| boolean killOnExpiry = |
| getBoolean(getQueuePrefix(queue) + RESERVATION_MOVE_ON_EXPIRY, |
| DEFAULT_RESERVATION_MOVE_ON_EXPIRY); |
| return killOnExpiry; |
| } |
| |
| @Override |
| public long getEnforcementWindow(String queue) { |
| long enforcementWindow = |
| getLong(getQueuePrefix(queue) + RESERVATION_ENFORCEMENT_WINDOW, |
| DEFAULT_RESERVATION_ENFORCEMENT_WINDOW); |
| return enforcementWindow; |
| } |
| |
| /** |
| * Sets the <em>disable_preemption</em> property in order to indicate |
| * whether or not container preemption will be disabled for the specified |
| * queue. |
| * |
| * @param queue queue path |
| * @param preemptionDisabled true if preemption is disabled on queue |
| */ |
| public void setPreemptionDisabled(String queue, boolean preemptionDisabled) { |
| setBoolean(getQueuePrefix(queue) + QUEUE_PREEMPTION_DISABLED, |
| preemptionDisabled); |
| } |
| |
| /** |
| * Indicates whether preemption is disabled on the specified queue. |
| * |
| * @param queue queue path to query |
| * @param defaultVal used as default if the <em>disable_preemption</em> |
| * is not set in the configuration |
| * @return true if preemption is disabled on <em>queue</em>, false otherwise |
| */ |
| public boolean getPreemptionDisabled(String queue, boolean defaultVal) { |
| boolean preemptionDisabled = |
| getBoolean(getQueuePrefix(queue) + QUEUE_PREEMPTION_DISABLED, |
| defaultVal); |
| return preemptionDisabled; |
| } |
| |
| /** |
| * Indicates whether intra-queue preemption is disabled on the specified queue |
| * |
| * @param queue queue path to query |
| * @param defaultVal used as default if the property is not set in the |
| * configuration |
| * @return true if preemption is disabled on queue, false otherwise |
| */ |
| public boolean getIntraQueuePreemptionDisabled(String queue, |
| boolean defaultVal) { |
| return |
| getBoolean(getQueuePrefix(queue) + INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX |
| + QUEUE_PREEMPTION_DISABLED, defaultVal); |
| } |
| |
| /** |
| * Get configured node labels in a given queuePath |
| */ |
| public Set<String> getConfiguredNodeLabels(String queuePath) { |
| Set<String> configuredNodeLabels = new HashSet<String>(); |
| Entry<String, String> e = null; |
| |
| Iterator<Entry<String, String>> iter = iterator(); |
| while (iter.hasNext()) { |
| e = iter.next(); |
| String key = e.getKey(); |
| |
| if (key.startsWith(getQueuePrefix(queuePath) + ACCESSIBLE_NODE_LABELS |
| + DOT)) { |
| // Find <label-name> in |
| // <queue-path>.accessible-node-labels.<label-name>.property |
| int labelStartIdx = |
| key.indexOf(ACCESSIBLE_NODE_LABELS) |
| + ACCESSIBLE_NODE_LABELS.length() + 1; |
| int labelEndIndx = key.indexOf('.', labelStartIdx); |
| String labelName = key.substring(labelStartIdx, labelEndIndx); |
| configuredNodeLabels.add(labelName); |
| } |
| } |
| |
| // always add NO_LABEL |
| configuredNodeLabels.add(RMNodeLabelsManager.NO_LABEL); |
| |
| return configuredNodeLabels; |
| } |
| |
| public Integer getDefaultApplicationPriorityConfPerQueue(String queue) { |
| Integer defaultPriority = getInt(getQueuePrefix(queue) |
| + DEFAULT_APPLICATION_PRIORITY, |
| DEFAULT_CONFIGURATION_APPLICATION_PRIORITY); |
| return defaultPriority; |
| } |
| |
| @VisibleForTesting |
| public void setOrderingPolicy(String queue, String policy) { |
| set(getQueuePrefix(queue) + ORDERING_POLICY, policy); |
| } |
| |
| @VisibleForTesting |
| public void setOrderingPolicyParameter(String queue, |
| String parameterKey, String parameterValue) { |
| set(getQueuePrefix(queue) + ORDERING_POLICY + "." + parameterKey, |
| parameterValue); |
| } |
| |
| public boolean getLazyPreemptionEnabled() { |
| return getBoolean(LAZY_PREEMPTION_ENABLED, DEFAULT_LAZY_PREEMPTION_ENABLED); |
| } |
| |
| public boolean shouldAppFailFast(Configuration conf) { |
| return conf.getBoolean(APP_FAIL_FAST, DEFAULT_APP_FAIL_FAST); |
| } |
| |
| public Integer getMaxParallelAppsForQueue(String queue) { |
| int defaultMaxParallelAppsForQueue = |
| getInt(PREFIX + MAX_PARALLEL_APPLICATIONS, |
| DEFAULT_MAX_PARALLEL_APPLICATIONS); |
| |
| String maxParallelAppsForQueue = get(getQueuePrefix(queue) |
| + MAX_PARALLEL_APPLICATIONS); |
| |
| return (maxParallelAppsForQueue != null) ? |
| Integer.parseInt(maxParallelAppsForQueue) |
| : defaultMaxParallelAppsForQueue; |
| } |
| |
| public Integer getMaxParallelAppsForUser(String user) { |
| int defaultMaxParallelAppsForUser = |
| getInt(PREFIX + "user." + MAX_PARALLEL_APPLICATIONS, |
| DEFAULT_MAX_PARALLEL_APPLICATIONS); |
| String maxParallelAppsForUser = get(getUserPrefix(user) |
| + MAX_PARALLEL_APPLICATIONS); |
| |
| return (maxParallelAppsForUser != null) ? |
| Integer.parseInt(maxParallelAppsForUser) |
| : defaultMaxParallelAppsForUser; |
| } |
| |
| public boolean getAllowZeroCapacitySum(String queue) { |
| return getBoolean(getQueuePrefix(queue) |
| + ALLOW_ZERO_CAPACITY_SUM, DEFAULT_ALLOW_ZERO_CAPACITY_SUM); |
| } |
| |
| public void setAllowZeroCapacitySum(String queue, boolean value) { |
| setBoolean(getQueuePrefix(queue) |
| + ALLOW_ZERO_CAPACITY_SUM, value); |
| } |
| private static final String PREEMPTION_CONFIG_PREFIX = |
| "yarn.resourcemanager.monitor.capacity.preemption."; |
| |
| private static final String INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX = |
| "intra-queue-preemption."; |
| |
| /** If true, run the policy but do not affect the cluster with preemption and |
| * kill events. */ |
| public static final String PREEMPTION_OBSERVE_ONLY = |
| PREEMPTION_CONFIG_PREFIX + "observe_only"; |
| public static final boolean DEFAULT_PREEMPTION_OBSERVE_ONLY = false; |
| |
| /** Time in milliseconds between invocations of this policy */ |
| public static final String PREEMPTION_MONITORING_INTERVAL = |
| PREEMPTION_CONFIG_PREFIX + "monitoring_interval"; |
| public static final long DEFAULT_PREEMPTION_MONITORING_INTERVAL = 3000L; |
| |
| /** Time in milliseconds between requesting a preemption from an application |
| * and killing the container. */ |
| public static final String PREEMPTION_WAIT_TIME_BEFORE_KILL = |
| PREEMPTION_CONFIG_PREFIX + "max_wait_before_kill"; |
| public static final long DEFAULT_PREEMPTION_WAIT_TIME_BEFORE_KILL = 15000L; |
| |
| /** Maximum percentage of resources preemptionCandidates in a single round. By |
| * controlling this value one can throttle the pace at which containers are |
| * reclaimed from the cluster. After computing the total desired preemption, |
| * the policy scales it back within this limit. */ |
| public static final String TOTAL_PREEMPTION_PER_ROUND = |
| PREEMPTION_CONFIG_PREFIX + "total_preemption_per_round"; |
| public static final float DEFAULT_TOTAL_PREEMPTION_PER_ROUND = 0.1f; |
| |
| /** Maximum amount of resources above the target capacity ignored for |
| * preemption. This defines a deadzone around the target capacity that helps |
| * prevent thrashing and oscillations around the computed target balance. |
| * High values would slow the time to capacity and (absent natural |
| * completions) it might prevent convergence to guaranteed capacity. */ |
| public static final String PREEMPTION_MAX_IGNORED_OVER_CAPACITY = |
| PREEMPTION_CONFIG_PREFIX + "max_ignored_over_capacity"; |
| public static final double DEFAULT_PREEMPTION_MAX_IGNORED_OVER_CAPACITY = 0.1; |
| /** |
| * Given a computed preemption target, account for containers naturally |
| * expiring and preempt only this percentage of the delta. This determines |
| * the rate of geometric convergence into the deadzone ({@link |
| * #PREEMPTION_MAX_IGNORED_OVER_CAPACITY}). For example, a termination factor of 0.5 |
| * will reclaim almost 95% of resources within 5 * {@link |
| * #PREEMPTION_WAIT_TIME_BEFORE_KILL}, even absent natural termination. */ |
| public static final String PREEMPTION_NATURAL_TERMINATION_FACTOR = |
| PREEMPTION_CONFIG_PREFIX + "natural_termination_factor"; |
| public static final double DEFAULT_PREEMPTION_NATURAL_TERMINATION_FACTOR = |
| 0.2; |
| |
| /** |
| * By default, reserved resource will be excluded while balancing capacities |
| * of queues. |
| * |
| * Why doing this? In YARN-4390, we added preemption-based-on-reserved-container |
| * Support. To reduce unnecessary preemption for large containers. We will |
| * not include reserved resources while calculating ideal-allocation in |
| * FifoCandidatesSelector. |
| * |
| * Changes in YARN-4390 will significantly reduce number of containers preempted |
| * When cluster has heterogeneous container requests. (Please check test |
| * report: https://issues.apache.org/jira/secure/attachment/12796197/YARN-4390-test-results.pdf |
| * |
| * However, on the other hand, in some corner cases, especially for |
| * fragmented cluster. It could lead to preemption cannot kick in in some |
| * cases. Please see YARN-5731. |
| * |
| * So to solve the problem, make this change to be configurable, and please |
| * note that it is an experimental option. |
| */ |
| public static final String |
| ADDITIONAL_RESOURCE_BALANCE_BASED_ON_RESERVED_CONTAINERS = |
| PREEMPTION_CONFIG_PREFIX |
| + "additional_res_balance_based_on_reserved_containers"; |
| public static final boolean |
| DEFAULT_ADDITIONAL_RESOURCE_BALANCE_BASED_ON_RESERVED_CONTAINERS = false; |
| |
| /** |
| * When calculating which containers to be preempted, we will try to preempt |
| * containers for reserved containers first. By default is false. |
| */ |
| public static final String PREEMPTION_SELECT_CANDIDATES_FOR_RESERVED_CONTAINERS = |
| PREEMPTION_CONFIG_PREFIX + "select_based_on_reserved_containers"; |
| public static final boolean DEFAULT_PREEMPTION_SELECT_CANDIDATES_FOR_RESERVED_CONTAINERS = |
| false; |
| |
| /** |
| * For intra-queue preemption, priority/user-limit/fairness based selectors |
| * can help to preempt containers. |
| */ |
| public static final String INTRAQUEUE_PREEMPTION_ENABLED = |
| PREEMPTION_CONFIG_PREFIX + |
| INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX + "enabled"; |
| public static final boolean DEFAULT_INTRAQUEUE_PREEMPTION_ENABLED = false; |
| |
| /** |
| * For intra-queue preemption, consider those queues which are above used cap |
| * limit. |
| */ |
| public static final String INTRAQUEUE_PREEMPTION_MINIMUM_THRESHOLD = |
| PREEMPTION_CONFIG_PREFIX + |
| INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX + "minimum-threshold"; |
| public static final float DEFAULT_INTRAQUEUE_PREEMPTION_MINIMUM_THRESHOLD = |
| 0.5f; |
| |
| /** |
| * For intra-queue preemption, allowable maximum-preemptable limit per queue. |
| */ |
| public static final String INTRAQUEUE_PREEMPTION_MAX_ALLOWABLE_LIMIT = |
| PREEMPTION_CONFIG_PREFIX + |
| INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX + "max-allowable-limit"; |
| public static final float DEFAULT_INTRAQUEUE_PREEMPTION_MAX_ALLOWABLE_LIMIT = |
| 0.2f; |
| |
| /** |
| * For intra-queue preemption, enforce a preemption order such as |
| * "userlimit_first" or "priority_first". |
| */ |
| public static final String INTRAQUEUE_PREEMPTION_ORDER_POLICY = PREEMPTION_CONFIG_PREFIX |
| + INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX + "preemption-order-policy"; |
| public static final String DEFAULT_INTRAQUEUE_PREEMPTION_ORDER_POLICY = "userlimit_first"; |
| |
| /** |
| * Flag to determine whether or not to preempt containers from apps where some |
| * used resources are less than the user's user limit. |
| */ |
| public static final String CROSS_QUEUE_PREEMPTION_CONSERVATIVE_DRF = |
| PREEMPTION_CONFIG_PREFIX + "conservative-drf"; |
| public static final Boolean DEFAULT_CROSS_QUEUE_PREEMPTION_CONSERVATIVE_DRF = |
| false; |
| |
| public static final String IN_QUEUE_PREEMPTION_CONSERVATIVE_DRF = |
| PREEMPTION_CONFIG_PREFIX + INTRA_QUEUE_PREEMPTION_CONFIG_PREFIX + |
| "conservative-drf"; |
| public static final Boolean DEFAULT_IN_QUEUE_PREEMPTION_CONSERVATIVE_DRF = |
| true; |
| |
| /** |
| * Should we allow queues continue grow after all queue reaches their |
| * guaranteed capacity. |
| */ |
| public static final String PREEMPTION_TO_BALANCE_QUEUES_BEYOND_GUARANTEED = |
| PREEMPTION_CONFIG_PREFIX + "preemption-to-balance-queue-after-satisfied.enabled"; |
| public static final boolean DEFAULT_PREEMPTION_TO_BALANCE_QUEUES_BEYOND_GUARANTEED = false; |
| |
| /** |
| * How long we will wait to balance queues, by default it is 5 mins. |
| */ |
| public static final String MAX_WAIT_BEFORE_KILL_FOR_QUEUE_BALANCE_PREEMPTION = |
| PREEMPTION_CONFIG_PREFIX + "preemption-to-balance-queue-after-satisfied.max-wait-before-kill"; |
| public static final long |
| DEFAULT_MAX_WAIT_BEFORE_KILL_FOR_QUEUE_BALANCE_PREEMPTION = |
| 300 * 1000; |
| |
| /** |
| * Maximum application for a queue to be used when application per queue is |
| * not defined.To be consistent with previous version the default value is set |
| * as UNDEFINED. |
| */ |
| @Private |
| public static final String QUEUE_GLOBAL_MAX_APPLICATION = |
| PREFIX + "global-queue-max-application"; |
| |
| public int getGlobalMaximumApplicationsPerQueue() { |
| int maxApplicationsPerQueue = |
| getInt(QUEUE_GLOBAL_MAX_APPLICATION, (int) UNDEFINED); |
| return maxApplicationsPerQueue; |
| } |
| |
| public void setGlobalMaximumApplicationsPerQueue(int val) { |
| setInt(QUEUE_GLOBAL_MAX_APPLICATION, val); |
| } |
| |
| /** |
| * Ordering policy inside a parent queue to sort queues |
| */ |
| |
| /** |
| * Less relative usage queue can get next resource, this is default |
| */ |
| public static final String QUEUE_UTILIZATION_ORDERING_POLICY = "utilization"; |
| |
| /** |
| * Combination of relative usage and priority |
| */ |
| public static final String QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY = |
| "priority-utilization"; |
| |
| public static final String DEFAULT_QUEUE_ORDERING_POLICY = |
| QUEUE_UTILIZATION_ORDERING_POLICY; |
| |
| |
| @Private |
| public void setQueueOrderingPolicy(String queue, String policy) { |
| set(getQueuePrefix(queue) + ORDERING_POLICY, policy); |
| } |
| |
| @Private |
| public QueueOrderingPolicy getQueueOrderingPolicy(String queue, |
| String parentPolicy) { |
| String defaultPolicy = parentPolicy; |
| if (null == defaultPolicy) { |
| defaultPolicy = DEFAULT_QUEUE_ORDERING_POLICY; |
| } |
| |
| String policyType = get(getQueuePrefix(queue) + ORDERING_POLICY, |
| defaultPolicy).trim(); |
| |
| QueueOrderingPolicy qop; |
| if (policyType.equals(QUEUE_UTILIZATION_ORDERING_POLICY)) { |
| // Doesn't respect priority |
| qop = new PriorityUtilizationQueueOrderingPolicy(false); |
| } else if (policyType.equals( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY)) { |
| qop = new PriorityUtilizationQueueOrderingPolicy(true); |
| } else { |
| try { |
| qop = (QueueOrderingPolicy) Class.forName(policyType).newInstance(); |
| } catch (Exception e) { |
| String message = "Unable to construct queue ordering policy=" |
| + policyType + " queue=" + queue; |
| throw new YarnRuntimeException(message, e); |
| } |
| } |
| |
| return qop; |
| } |
| |
| /* |
| * Get global configuration for ordering policies |
| */ |
| private String getOrderingPolicyGlobalConfigKey(String orderPolicyName, |
| String configKey) { |
| return PREFIX + ORDERING_POLICY + DOT + orderPolicyName + DOT + configKey; |
| } |
| |
| /** |
| * Global configurations of queue-priority-utilization ordering policy |
| */ |
| private static final String UNDER_UTILIZED_PREEMPTION_ENABLED = |
| "underutilized-preemption.enabled"; |
| |
| /** |
| * Do we allow under-utilized queue with higher priority to preempt queue |
| * with lower priority *even if queue with lower priority is not satisfied*. |
| * |
| * For example, two queues, a and b |
| * a.priority = 1, (a.used-capacity - a.reserved-capacity) = 40% |
| * b.priority = 0, b.used-capacity = 30% |
| * |
| * Set this configuration to true to allow queue-a to preempt container from |
| * queue-b. |
| * |
| * (The reason why deduct reserved-capacity from used-capacity for queue with |
| * higher priority is: the reserved-capacity is just scheduler's internal |
| * implementation to allocate large containers, it is not possible for |
| * application to use such reserved-capacity. It is possible that a queue with |
| * large container requests have a large number of containers but cannot |
| * allocate from any of them. But scheduler will make sure a satisfied queue |
| * will not preempt resource from any other queues. A queue is considered to |
| * be satisfied when queue's used-capacity - reserved-capacity ≥ |
| * guaranteed-capacity.) |
| * |
| * @return allowed or not |
| */ |
| public boolean getPUOrderingPolicyUnderUtilizedPreemptionEnabled() { |
| return getBoolean(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_ENABLED), false); |
| } |
| |
| @VisibleForTesting |
| public void setPUOrderingPolicyUnderUtilizedPreemptionEnabled( |
| boolean enabled) { |
| setBoolean(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_ENABLED), enabled); |
| } |
| |
| private static final String UNDER_UTILIZED_PREEMPTION_DELAY = |
| "underutilized-preemption.reserved-container-delay-ms"; |
| |
| /** |
| * When a reserved container of an underutilized queue is created. Preemption |
| * will kick in after specified delay (in ms). |
| * |
| * The total time to preempt resources for a reserved container from higher |
| * priority queue will be: reserved-container-delay-ms + |
| * {@link CapacitySchedulerConfiguration#PREEMPTION_WAIT_TIME_BEFORE_KILL}. |
| * |
| * This parameter is added to make preemption from lower priority queue which |
| * is underutilized to be more careful. This parameter takes effect when |
| * underutilized-preemption.enabled set to true. |
| * |
| * @return delay |
| */ |
| public long getPUOrderingPolicyUnderUtilizedPreemptionDelay() { |
| return getLong(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_DELAY), 60000L); |
| } |
| |
| @VisibleForTesting |
| public void setPUOrderingPolicyUnderUtilizedPreemptionDelay( |
| long timeout) { |
| setLong(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_DELAY), timeout); |
| } |
| |
| private static final String UNDER_UTILIZED_PREEMPTION_MOVE_RESERVATION = |
| "underutilized-preemption.allow-move-reservation"; |
| |
| /** |
| * When doing preemption from under-satisfied queues for priority queue. |
| * Do we allow move reserved container from one host to another? |
| * |
| * @return allow or not |
| */ |
| public boolean getPUOrderingPolicyUnderUtilizedPreemptionMoveReservation() { |
| return getBoolean(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_MOVE_RESERVATION), false); |
| } |
| |
| @VisibleForTesting |
| public void setPUOrderingPolicyUnderUtilizedPreemptionMoveReservation( |
| boolean allowMoveReservation) { |
| setBoolean(getOrderingPolicyGlobalConfigKey( |
| QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, |
| UNDER_UTILIZED_PREEMPTION_MOVE_RESERVATION), allowMoveReservation); |
| } |
| |
| /** |
| * Get the weights of all users at this queue level from the configuration. |
| * Used in computing user-specific user limit, relative to other users. |
| * @param queuePath full queue path |
| * @return map of user weights, if they exists. Otherwise, return empty map. |
| */ |
| public Map<String, Float> getAllUserWeightsForQueue(String queuePath) { |
| Map <String, Float> userWeights = new HashMap <String, Float>(); |
| String qPathPlusPrefix = |
| getQueuePrefix(queuePath).replaceAll("\\.", "\\\\.") |
| + USER_SETTINGS + "\\."; |
| String weightKeyRegex = |
| qPathPlusPrefix + "\\w+\\." + USER_WEIGHT; |
| Map<String, String> props = getValByRegex(weightKeyRegex); |
| for (Entry<String, String> e : props.entrySet()) { |
| String userName = |
| e.getKey().replaceFirst(qPathPlusPrefix, "") |
| .replaceFirst("\\." + USER_WEIGHT, ""); |
| if (userName != null && !userName.isEmpty()) { |
| userWeights.put(userName, new Float(e.getValue())); |
| } |
| } |
| return userWeights; |
| } |
| |
| public boolean getAssignMultipleEnabled() { |
| return getBoolean(ASSIGN_MULTIPLE_ENABLED, DEFAULT_ASSIGN_MULTIPLE_ENABLED); |
| } |
| |
| public int getMaxAssignPerHeartbeat() { |
| return getInt(MAX_ASSIGN_PER_HEARTBEAT, DEFAULT_MAX_ASSIGN_PER_HEARTBEAT); |
| } |
| |
| public static final String MAXIMUM_LIFETIME_SUFFIX = |
| "maximum-application-lifetime"; |
| |
| public static final String DEFAULT_LIFETIME_SUFFIX = |
| "default-application-lifetime"; |
| |
| public long getMaximumLifetimePerQueue(String queue) { |
| long maximumLifetimePerQueue = getLong( |
| getQueuePrefix(queue) + MAXIMUM_LIFETIME_SUFFIX, (long) UNDEFINED); |
| return maximumLifetimePerQueue; |
| } |
| |
| public void setMaximumLifetimePerQueue(String queue, long maximumLifetime) { |
| setLong(getQueuePrefix(queue) + MAXIMUM_LIFETIME_SUFFIX, maximumLifetime); |
| } |
| |
| public long getDefaultLifetimePerQueue(String queue) { |
| long maximumLifetimePerQueue = getLong( |
| getQueuePrefix(queue) + DEFAULT_LIFETIME_SUFFIX, (long) UNDEFINED); |
| return maximumLifetimePerQueue; |
| } |
| |
| public void setDefaultLifetimePerQueue(String queue, long defaultLifetime) { |
| setLong(getQueuePrefix(queue) + DEFAULT_LIFETIME_SUFFIX, defaultLifetime); |
| } |
| |
| @Private |
| public static final boolean DEFAULT_AUTO_CREATE_CHILD_QUEUE_ENABLED = false; |
| |
| @Private |
| private static final String AUTO_CREATE_CHILD_QUEUE_PREFIX = |
| "auto-create-child-queue."; |
| |
| @Private |
| public static final String AUTO_CREATE_CHILD_QUEUE_ENABLED = |
| AUTO_CREATE_CHILD_QUEUE_PREFIX + "enabled"; |
| |
| @Private |
| private static final String AUTO_QUEUE_CREATION_V2_PREFIX = |
| "auto-queue-creation-v2."; |
| |
| @Private |
| public static final String AUTO_QUEUE_CREATION_V2_ENABLED = |
| AUTO_QUEUE_CREATION_V2_PREFIX + "enabled"; |
| |
| @Private |
| public static final String AUTO_QUEUE_CREATION_V2_MAX_QUEUES = |
| AUTO_QUEUE_CREATION_V2_PREFIX + "max-queues"; |
| |
| @Private |
| public static final int |
| DEFAULT_AUTO_QUEUE_CREATION_V2_MAX_QUEUES = 1000; |
| |
| @Private |
| public static final boolean DEFAULT_AUTO_QUEUE_CREATION_ENABLED = false; |
| |
| @Private |
| public static final String AUTO_CREATED_LEAF_QUEUE_TEMPLATE_PREFIX = |
| "leaf-queue-template"; |
| |
| @Private |
| public static final String AUTO_CREATE_QUEUE_MAX_QUEUES = |
| "auto-create-child-queue.max-queues"; |
| |
| @Private |
| public static final int DEFAULT_AUTO_CREATE_QUEUE_MAX_QUEUES = 1000; |
| |
| /** |
| * If true, this queue will be created as a Parent Queue which Auto Created |
| * leaf child queues |
| * |
| * @param queuePath The queues path |
| * @return true if auto create is enabled for child queues else false. Default |
| * is false |
| */ |
| @Private |
| public boolean isAutoCreateChildQueueEnabled(String queuePath) { |
| boolean isAutoCreateEnabled = getBoolean( |
| getQueuePrefix(queuePath) + AUTO_CREATE_CHILD_QUEUE_ENABLED, |
| DEFAULT_AUTO_CREATE_CHILD_QUEUE_ENABLED); |
| return isAutoCreateEnabled; |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setAutoCreateChildQueueEnabled(String queuePath, |
| boolean autoCreationEnabled) { |
| setBoolean(getQueuePrefix(queuePath) + |
| AUTO_CREATE_CHILD_QUEUE_ENABLED, |
| autoCreationEnabled); |
| } |
| |
| public void setAutoQueueCreationV2Enabled(String queuePath, |
| boolean autoQueueCreation) { |
| setBoolean( |
| getQueuePrefix(queuePath) + AUTO_QUEUE_CREATION_V2_ENABLED, |
| autoQueueCreation); |
| } |
| |
| public boolean isAutoQueueCreationV2Enabled(String queuePath) { |
| boolean isAutoQueueCreation = getBoolean( |
| getQueuePrefix(queuePath) + AUTO_QUEUE_CREATION_V2_ENABLED, |
| DEFAULT_AUTO_QUEUE_CREATION_ENABLED); |
| return isAutoQueueCreation; |
| } |
| |
| /** |
| * Get the auto created leaf queue's template configuration prefix |
| * Leaf queue's template capacities are configured at the parent queue |
| * |
| * @param queuePath parent queue's path |
| * @return Config prefix for leaf queue template configurations |
| */ |
| @Private |
| public String getAutoCreatedQueueTemplateConfPrefix(String queuePath) { |
| return queuePath + DOT + AUTO_CREATED_LEAF_QUEUE_TEMPLATE_PREFIX; |
| } |
| |
| @Private |
| public static final String FAIL_AUTO_CREATION_ON_EXCEEDING_CAPACITY = |
| "auto-create-child-queue.fail-on-exceeding-parent-capacity"; |
| |
| @Private |
| public static final boolean DEFAULT_FAIL_AUTO_CREATION_ON_EXCEEDING_CAPACITY = |
| false; |
| |
| /** |
| * Fail further auto leaf queue creation when parent's guaranteed capacity is |
| * exceeded. |
| * |
| * @param queuePath the parent queue's path |
| * @return true if configured to fail else false |
| */ |
| @Private |
| public boolean getShouldFailAutoQueueCreationWhenGuaranteedCapacityExceeded( |
| String queuePath) { |
| boolean shouldFailAutoQueueCreationOnExceedingGuaranteedCapacity = |
| getBoolean(getQueuePrefix(queuePath) |
| + FAIL_AUTO_CREATION_ON_EXCEEDING_CAPACITY, |
| DEFAULT_FAIL_AUTO_CREATION_ON_EXCEEDING_CAPACITY); |
| return shouldFailAutoQueueCreationOnExceedingGuaranteedCapacity; |
| } |
| |
| @VisibleForTesting |
| @Private |
| public void setShouldFailAutoQueueCreationWhenGuaranteedCapacityExceeded( |
| String queuePath, boolean autoCreationEnabled) { |
| setBoolean( |
| getQueuePrefix(queuePath) + |
| FAIL_AUTO_CREATION_ON_EXCEEDING_CAPACITY, |
| autoCreationEnabled); |
| } |
| |
| /** |
| * Get the max number of leaf queues that are allowed to be created under |
| * a parent queue |
| * |
| * @param queuePath the paret queue's path |
| * @return the max number of leaf queues allowed to be auto created |
| */ |
| @Private |
| public int getAutoCreatedQueuesMaxChildQueuesLimit(String queuePath) { |
| return getInt(getQueuePrefix(queuePath) + |
| AUTO_CREATE_QUEUE_MAX_QUEUES, |
| DEFAULT_AUTO_CREATE_QUEUE_MAX_QUEUES); |
| } |
| |
| /** |
| * Get the max number of queues that are allowed to be created under |
| * a parent queue which allowed auto creation v2. |
| * |
| * @param queuePath the parent queue's path |
| * @return the max number of queues allowed to be auto created, |
| * in new auto created. |
| */ |
| @Private |
| public int getAutoCreatedQueuesV2MaxChildQueuesLimit(String queuePath) { |
| return getInt(getQueuePrefix(queuePath) + |
| AUTO_QUEUE_CREATION_V2_MAX_QUEUES, |
| DEFAULT_AUTO_QUEUE_CREATION_V2_MAX_QUEUES); |
| } |
| |
| @VisibleForTesting |
| public void setAutoCreatedQueuesV2MaxChildQueuesLimit(String queuePath, |
| int maxQueues) { |
| setInt(getQueuePrefix(queuePath) + |
| AUTO_QUEUE_CREATION_V2_MAX_QUEUES, maxQueues); |
| } |
| |
| @Private |
| public static final String AUTO_CREATED_QUEUE_MANAGEMENT_POLICY = |
| AUTO_CREATE_CHILD_QUEUE_PREFIX + "management-policy"; |
| |
| @Private |
| public static final String DEFAULT_AUTO_CREATED_QUEUE_MANAGEMENT_POLICY = |
| "org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity" |
| + ".queuemanagement." |
| + "GuaranteedOrZeroCapacityOverTimePolicy"; |
| |
| @Private |
| private static final String QUEUE_MANAGEMENT_CONFIG_PREFIX = |
| "yarn.resourcemanager.monitor.capacity.queue-management."; |
| |
| /** |
| * Time in milliseconds between invocations of this policy |
| */ |
| @Private |
| public static final String QUEUE_MANAGEMENT_MONITORING_INTERVAL = |
| QUEUE_MANAGEMENT_CONFIG_PREFIX + "monitoring-interval"; |
| |
| @Private |
| public static final long DEFAULT_QUEUE_MANAGEMENT_MONITORING_INTERVAL = |
| 1500L; |
| |
| /** |
| * Time in milliseconds between invocations |
| * of QueueConfigurationAutoRefreshPolicy. |
| */ |
| @Private |
| public static final String QUEUE_AUTO_REFRESH_MONITORING_INTERVAL = |
| PREFIX + "queue.auto.refresh.monitoring-interval"; |
| |
| @Private |
| public static final long DEFAULT_QUEUE_AUTO_REFRESH_MONITORING_INTERVAL = |
| 5000L; |
| |
| /** |
| * Queue Management computation policy for Auto Created queues |
| * @param queue The queue's path |
| * @return Configured policy class name |
| */ |
| @Private |
| public String getAutoCreatedQueueManagementPolicy(String queue) { |
| String autoCreatedQueueManagementPolicy = |
| get(getQueuePrefix(queue) + AUTO_CREATED_QUEUE_MANAGEMENT_POLICY, |
| DEFAULT_AUTO_CREATED_QUEUE_MANAGEMENT_POLICY); |
| return autoCreatedQueueManagementPolicy; |
| } |
| |
| /** |
| * Get The policy class configured to manage capacities for auto created leaf |
| * queues under the specified parent |
| * |
| * @param queueName The parent queue's name |
| * @return The policy class configured to manage capacities for auto created |
| * leaf queues under the specified parent queue |
| */ |
| @Private |
| protected AutoCreatedQueueManagementPolicy |
| getAutoCreatedQueueManagementPolicyClass( |
| String queueName) { |
| |
| String queueManagementPolicyClassName = |
| getAutoCreatedQueueManagementPolicy(queueName); |
| LOG.info("Using Auto Created Queue Management Policy: " |
| + queueManagementPolicyClassName + " for queue: " + queueName); |
| try { |
| Class<?> queueManagementPolicyClazz = getClassByName( |
| queueManagementPolicyClassName); |
| if (AutoCreatedQueueManagementPolicy.class.isAssignableFrom( |
| queueManagementPolicyClazz)) { |
| return (AutoCreatedQueueManagementPolicy) ReflectionUtils.newInstance( |
| queueManagementPolicyClazz, this); |
| } else{ |
| throw new YarnRuntimeException( |
| "Class: " + queueManagementPolicyClassName + " not instance of " |
| + AutoCreatedQueueManagementPolicy.class.getCanonicalName()); |
| } |
| } catch (ClassNotFoundException e) { |
| throw new YarnRuntimeException( |
| "Could not instantiate " + "AutoCreatedQueueManagementPolicy: " |
| + queueManagementPolicyClassName + " for queue: " + queueName, |
| e); |
| } |
| } |
| |
| @VisibleForTesting |
| @Private |
| public void setAutoCreatedLeafQueueConfigCapacity(String queuePath, |
| float val) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| queuePath); |
| setCapacity(leafQueueConfPrefix, val); |
| } |
| |
| @VisibleForTesting |
| @Private |
| public void setAutoCreatedLeafQueueTemplateCapacityByLabel(String queuePath, |
| String label, float val) { |
| 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, |
| float val) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| queuePath); |
| setMaximumCapacity(leafQueueConfPrefix, val); |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setAutoCreatedLeafQueueTemplateMaxCapacity(String queuePath, |
| String label, float val) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| 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, |
| int val) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| queuePath); |
| setUserLimit(leafQueueConfPrefix, val); |
| } |
| |
| @VisibleForTesting |
| @Private |
| public void setAutoCreatedLeafQueueConfigUserLimitFactor(String queuePath, |
| float val) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| queuePath); |
| setUserLimitFactor(leafQueueConfPrefix, val); |
| } |
| |
| @Private |
| @VisibleForTesting |
| public void setAutoCreatedLeafQueueConfigDefaultNodeLabelExpression(String |
| queuePath, |
| String expression) { |
| String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix( |
| queuePath); |
| setDefaultNodeLabelExpression(leafQueueConfPrefix, expression); |
| } |
| |
| public static String getUnits(String resourceValue) { |
| String units; |
| for (int i = 0; i < resourceValue.length(); i++) { |
| if (Character.isAlphabetic(resourceValue.charAt(i))) { |
| units = resourceValue.substring(i); |
| if (StringUtils.isAlpha(units)) { |
| return units; |
| } |
| } |
| } |
| return ""; |
| } |
| |
| /** |
| * Get absolute minimum resource requirement for a queue. |
| * |
| * @param label |
| * NodeLabel |
| * @param queue |
| * queue path |
| * @param resourceTypes |
| * Resource types |
| * @return ResourceInformation |
| */ |
| public Resource getMinimumResourceRequirement(String label, String queue, |
| Set<String> resourceTypes) { |
| return internalGetLabeledResourceRequirementForQueue(queue, label, |
| resourceTypes, CAPACITY); |
| } |
| |
| /** |
| * Get absolute maximum resource requirement for a queue. |
| * |
| * @param label |
| * NodeLabel |
| * @param queue |
| * queue path |
| * @param resourceTypes |
| * Resource types |
| * @return Resource |
| */ |
| public Resource getMaximumResourceRequirement(String label, String queue, |
| Set<String> resourceTypes) { |
| return internalGetLabeledResourceRequirementForQueue(queue, label, |
| resourceTypes, MAXIMUM_CAPACITY); |
| } |
| |
| @VisibleForTesting |
| public void setMinimumResourceRequirement(String label, String queue, |
| Resource resource) { |
| updateMinMaxResourceToConf(label, queue, resource, CAPACITY); |
| } |
| |
| @VisibleForTesting |
| public void setMaximumResourceRequirement(String label, String queue, |
| Resource resource) { |
| updateMinMaxResourceToConf(label, queue, resource, MAXIMUM_CAPACITY); |
| } |
| |
| private void updateMinMaxResourceToConf(String label, String queue, |
| Resource resource, String type) { |
| if (queue.equals("root")) { |
| throw new IllegalArgumentException( |
| "Cannot set resource, root queue will take 100% of cluster capacity"); |
| } |
| |
| StringBuilder resourceString = new StringBuilder(); |
| resourceString |
| .append("[" + AbsoluteResourceType.MEMORY.toString().toLowerCase() + "=" |
| + resource.getMemorySize() + "," |
| + AbsoluteResourceType.VCORES.toString().toLowerCase() + "=" |
| + resource.getVirtualCores() + "]"); |
| |
| String prefix = getQueuePrefix(queue) + type; |
| if (!label.isEmpty()) { |
| prefix = getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label |
| + DOT + type; |
| } |
| set(prefix, resourceString.toString()); |
| } |
| |
| public boolean checkConfigTypeIsAbsoluteResource(String label, String queue, |
| Set<String> resourceTypes) { |
| String propertyName = getNodeLabelPrefix(queue, label) + CAPACITY; |
| String resourceString = get(propertyName); |
| if (resourceString == null || resourceString.isEmpty()) { |
| return false; |
| } |
| |
| Matcher matcher = RESOURCE_PATTERN.matcher(resourceString); |
| if (matcher.find()) { |
| return true; |
| } |
| return false; |
| } |
| |
| private Resource internalGetLabeledResourceRequirementForQueue(String queue, |
| String label, Set<String> resourceTypes, String suffix) { |
| String propertyName = getNodeLabelPrefix(queue, label) + suffix; |
| String resourceString = get(propertyName); |
| if (resourceString == null || resourceString.isEmpty()) { |
| return Resources.none(); |
| } |
| |
| // Define resource here. |
| Resource resource = Resource.newInstance(0L, 0); |
| Matcher matcher = RESOURCE_PATTERN.matcher(resourceString); |
| |
| /* |
| * Absolute resource configuration for a queue will be grouped by "[]". |
| * Syntax of absolute resource config could be like below |
| * "memory=4Gi vcores=2". Ideally this means "4GB of memory and 2 vcores". |
| */ |
| if (matcher.find()) { |
| // Get the sub-group. |
| String subGroup = matcher.group(0); |
| if (subGroup.trim().isEmpty()) { |
| return Resources.none(); |
| } |
| subGroup = subGroup.substring(1, subGroup.length() - 1); |
| for (String kvPair : subGroup.trim().split(",")) { |
| String[] splits = kvPair.split("="); |
| |
| // Ensure that each sub string is key value pair separated by '='. |
| if (splits != null && splits.length > 1) { |
| updateResourceValuesFromConfig(resourceTypes, resource, splits); |
| } |
| } |
| } |
| |
| // Memory has to be configured always. |
| if (resource.getMemorySize() == 0L) { |
| return Resources.none(); |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("CSConf - getAbsolueResourcePerQueue: prefix=" |
| + getNodeLabelPrefix(queue, label) + ", capacity=" + resource); |
| } |
| return resource; |
| } |
| |
| private void updateResourceValuesFromConfig(Set<String> resourceTypes, |
| Resource resource, String[] splits) { |
| |
| // If key is not a valid type, skip it. |
| if (!resourceTypes.contains(splits[0])) { |
| return; |
| } |
| |
| String units = getUnits(splits[1]); |
| Long resourceValue = Long |
| .valueOf(splits[1].substring(0, splits[1].length() - units.length())); |
| |
| // Convert all incoming units to MB if units is configured. |
| if (!units.isEmpty()) { |
| resourceValue = UnitsConversionUtil.convert(units, "Mi", resourceValue); |
| } |
| |
| // map it based on key. |
| AbsoluteResourceType resType = AbsoluteResourceType |
| .valueOf(StringUtils.toUpperCase(splits[0].trim())); |
| switch (resType) { |
| case MEMORY : |
| resource.setMemorySize(resourceValue); |
| break; |
| case VCORES : |
| resource.setVirtualCores(resourceValue.intValue()); |
| break; |
| default : |
| resource.setResourceInformation(splits[0].trim(), ResourceInformation |
| .newInstance(splits[0].trim(), units, resourceValue)); |
| break; |
| } |
| } |
| |
| @Private public static final String MULTI_NODE_SORTING_POLICIES = |
| PREFIX + "multi-node-sorting.policy.names"; |
| |
| @Private public static final String MULTI_NODE_SORTING_POLICY_NAME = |
| PREFIX + "multi-node-sorting.policy"; |
| |
| /** |
| * resource usage based node sorting algorithm. |
| */ |
| public static final String DEFAULT_NODE_SORTING_POLICY = "default"; |
| public static final String DEFAULT_NODE_SORTING_POLICY_CLASSNAME |
| = "org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.ResourceUsageMultiNodeLookupPolicy"; |
| public static final long DEFAULT_MULTI_NODE_SORTING_INTERVAL = 1000L; |
| |
| @Private |
| public static final String MULTI_NODE_PLACEMENT_ENABLED = PREFIX |
| + "multi-node-placement-enabled"; |
| |
| @Private |
| public static final boolean DEFAULT_MULTI_NODE_PLACEMENT_ENABLED = false; |
| |
| public String getMultiNodesSortingAlgorithmPolicy( |
| String queue) { |
| |
| String policyName = get( |
| getQueuePrefix(queue) + "multi-node-sorting.policy"); |
| |
| if (policyName == null) { |
| policyName = get(MULTI_NODE_SORTING_POLICY_NAME); |
| } |
| |
| // If node sorting policy is not configured in queue and in cluster level, |
| // it is been assumed that this queue is not enabled with multi-node lookup. |
| if (policyName == null || policyName.isEmpty()) { |
| return null; |
| } |
| |
| String policyClassName = get(MULTI_NODE_SORTING_POLICY_NAME + DOT |
| + policyName.trim() + DOT + "class"); |
| |
| if (policyClassName == null || policyClassName.isEmpty()) { |
| throw new YarnRuntimeException( |
| policyName.trim() + " Class is not configured or not an instance of " |
| + MultiNodeLookupPolicy.class.getCanonicalName()); |
| } |
| |
| return normalizePolicyName(policyClassName.trim()); |
| } |
| |
| public boolean getMultiNodePlacementEnabled() { |
| return getBoolean(MULTI_NODE_PLACEMENT_ENABLED, |
| DEFAULT_MULTI_NODE_PLACEMENT_ENABLED); |
| } |
| |
| public Set<MultiNodePolicySpec> getMultiNodePlacementPolicies() { |
| String[] policies = getTrimmedStrings(MULTI_NODE_SORTING_POLICIES); |
| |
| // In other cases, split the accessibleLabelStr by "," |
| Set<MultiNodePolicySpec> set = new HashSet<MultiNodePolicySpec>(); |
| for (String str : policies) { |
| if (!str.trim().isEmpty()) { |
| String policyClassName = get( |
| MULTI_NODE_SORTING_POLICY_NAME + DOT + str.trim() + DOT + "class"); |
| if (str.trim().equals(DEFAULT_NODE_SORTING_POLICY)) { |
| policyClassName = get( |
| MULTI_NODE_SORTING_POLICY_NAME + DOT + str.trim() + DOT + "class", |
| DEFAULT_NODE_SORTING_POLICY_CLASSNAME); |
| } |
| |
| // This check is needed as default class name is loaded only for |
| // DEFAULT_NODE_SORTING_POLICY. |
| if (policyClassName == null) { |
| throw new YarnRuntimeException( |
| str.trim() + " Class is not configured or not an instance of " |
| + MultiNodeLookupPolicy.class.getCanonicalName()); |
| } |
| policyClassName = normalizePolicyName(policyClassName.trim()); |
| long policySortingInterval = getLong( |
| MULTI_NODE_SORTING_POLICY_NAME + DOT + str.trim() |
| + DOT + "sorting-interval.ms", |
| DEFAULT_MULTI_NODE_SORTING_INTERVAL); |
| if (policySortingInterval < 0) { |
| throw new YarnRuntimeException( |
| str.trim() |
| + " multi-node policy is configured with invalid" |
| + " sorting-interval:" + policySortingInterval); |
| } |
| set.add( |
| new MultiNodePolicySpec(policyClassName, policySortingInterval)); |
| } |
| } |
| |
| return Collections.unmodifiableSet(set); |
| } |
| |
| private String normalizePolicyName(String policyName) { |
| |
| // Ensure that custom node sorting algorithm class is valid. |
| try { |
| Class<?> nodeSortingPolicyClazz = getClassByName(policyName); |
| if (MultiNodeLookupPolicy.class |
| .isAssignableFrom(nodeSortingPolicyClazz)) { |
| return policyName; |
| } else { |
| throw new YarnRuntimeException( |
| "Class: " + policyName + " not instance of " |
| + MultiNodeLookupPolicy.class.getCanonicalName()); |
| } |
| } catch (ClassNotFoundException e) { |
| throw new YarnRuntimeException( |
| "Could not instantiate " + "NodesSortingPolicy: " + policyName, e); |
| } |
| } |
| } |
