You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by kk...@apache.org on 2018/04/04 17:55:10 UTC
[1/2] hadoop git commit: YARN-8013. Support application tags when
defining application namespaces for placement constraints. Contributed by
Weiwei Yang.
Repository: hadoop
Updated Branches:
refs/heads/branch-3.1 4db13cb94 -> fa464c12a
refs/heads/trunk 42cd367c9 -> 7853ec8d2
YARN-8013. Support application tags when defining application namespaces for placement constraints. Contributed by Weiwei Yang.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/7853ec8d
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/7853ec8d
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/7853ec8d
Branch: refs/heads/trunk
Commit: 7853ec8d2fb8731b7f7c28fd87491a0a2d47967e
Parents: 42cd367
Author: Konstantinos Karanasos <kk...@apache.org>
Authored: Wed Apr 4 10:51:58 2018 -0700
Committer: Konstantinos Karanasos <kk...@apache.org>
Committed: Wed Apr 4 10:51:58 2018 -0700
----------------------------------------------------------------------
.../api/records/AllocationTagNamespaceType.java | 2 +-
.../constraint/AllocationTagNamespace.java | 312 ------------------
.../scheduler/constraint/AllocationTags.java | 44 ++-
.../constraint/AllocationTagsManager.java | 47 ++-
.../constraint/PlacementConstraintsUtil.java | 41 +--
.../constraint/TargetApplications.java | 53 ++-
.../constraint/TargetApplicationsNamespace.java | 326 +++++++++++++++++++
.../SingleConstraintAppPlacementAllocator.java | 21 --
.../server/resourcemanager/rmapp/MockRMApp.java | 9 +-
...estSchedulingRequestContainerAllocation.java | 5 +-
.../constraint/TestAllocationTagsManager.java | 22 +-
.../constraint/TestAllocationTagsNamespace.java | 89 ++++-
.../TestPlacementConstraintsUtil.java | 125 ++++++-
13 files changed, 654 insertions(+), 442 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
index de5492e..f304600 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
@@ -26,7 +26,7 @@ public enum AllocationTagNamespaceType {
SELF("self"),
NOT_SELF("not-self"),
APP_ID("app-id"),
- APP_LABEL("app-label"),
+ APP_TAG("app-tag"),
ALL("all");
private String typeKeyword;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
deleted file mode 100644
index 7b9f3be..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
+++ /dev/null
@@ -1,312 +0,0 @@
-/**
- * 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.constraint;
-
-import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableSet;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.SELF;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.NOT_SELF;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_LABEL;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_ID;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.ALL;
-
-/**
- * Class to describe the namespace of an allocation tag.
- * Each namespace can be evaluated against a set of applications.
- * After evaluation, the namespace should have an implicit set of
- * applications which defines its scope.
- */
-public abstract class AllocationTagNamespace implements
- Evaluable<TargetApplications> {
-
- public final static String NAMESPACE_DELIMITER = "/";
-
- private AllocationTagNamespaceType nsType;
- // Namespace scope value will be delay binding by eval method.
- private Set<ApplicationId> nsScope;
-
- public AllocationTagNamespace(AllocationTagNamespaceType
- allocationTagNamespaceType) {
- this.nsType = allocationTagNamespaceType;
- }
-
- protected void setScopeIfNotNull(Set<ApplicationId> appIds) {
- if (appIds != null) {
- this.nsScope = appIds;
- }
- }
-
- /**
- * Get the type of the namespace.
- * @return namespace type.
- */
- public AllocationTagNamespaceType getNamespaceType() {
- return nsType;
- }
-
- /**
- * Get the scope of the namespace, in form of a set of applications.
- *
- * @return a set of applications.
- */
- public Set<ApplicationId> getNamespaceScope() {
- if (this.nsScope == null) {
- throw new IllegalStateException("Invalid namespace scope,"
- + " it is not initialized. Evaluate must be called before"
- + " a namespace can be consumed.");
- }
- return this.nsScope;
- }
-
- /**
- * Evaluate the namespace against given target applications
- * if it is necessary. Only self/not-self/app-label namespace types
- * require this evaluation step, because they are not binding to a
- * specific scope during initiating. So we do lazy binding for them
- * in this method.
- *
- * @param target a generic type target that impacts this evaluation.
- * @throws InvalidAllocationTagsQueryException
- */
- @Override
- public void evaluate(TargetApplications target)
- throws InvalidAllocationTagsQueryException {
- // Sub-class needs to override this when it requires the eval step.
- }
-
- @Override
- public String toString() {
- return this.nsType.toString();
- }
-
- /**
- * Namespace within application itself.
- */
- public static class Self extends AllocationTagNamespace {
-
- public Self() {
- super(SELF);
- }
-
- @Override
- public void evaluate(TargetApplications target)
- throws InvalidAllocationTagsQueryException {
- if (target == null || target.getCurrentApplicationId() == null) {
- throw new InvalidAllocationTagsQueryException("Namespace Self must"
- + " be evaluated against a single application ID.");
- }
- ApplicationId applicationId = target.getCurrentApplicationId();
- setScopeIfNotNull(ImmutableSet.of(applicationId));
- }
- }
-
- /**
- * Namespace to all applications except itself.
- */
- public static class NotSelf extends AllocationTagNamespace {
-
- private ApplicationId applicationId;
-
- public NotSelf() {
- super(NOT_SELF);
- }
-
- /**
- * The scope of self namespace is to an application itself,
- * the application ID can be delay binding to the namespace.
- *
- * @param appId application ID.
- */
- public void setApplicationId(ApplicationId appId) {
- this.applicationId = appId;
- }
-
- public ApplicationId getApplicationId() {
- return this.applicationId;
- }
-
- @Override
- public void evaluate(TargetApplications target) {
- Set<ApplicationId> otherAppIds = target.getOtherApplicationIds();
- setScopeIfNotNull(otherAppIds);
- }
- }
-
- /**
- * Namespace to all applications in the cluster.
- */
- public static class All extends AllocationTagNamespace {
-
- public All() {
- super(ALL);
- }
- }
-
- /**
- * Namespace to all applications in the cluster.
- */
- public static class AppLabel extends AllocationTagNamespace {
-
- public AppLabel() {
- super(APP_LABEL);
- }
-
- @Override
- public void evaluate(TargetApplications target) {
- // TODO Implement app-label namespace evaluation
- }
- }
-
- /**
- * Namespace defined by a certain application ID.
- */
- public static class AppID extends AllocationTagNamespace {
-
- private ApplicationId targetAppId;
- // app-id namespace requires an extra value of an application id.
- public AppID(ApplicationId applicationId) {
- super(APP_ID);
- this.targetAppId = applicationId;
- setScopeIfNotNull(ImmutableSet.of(targetAppId));
- }
-
- @Override
- public String toString() {
- return APP_ID.toString() + NAMESPACE_DELIMITER + this.targetAppId;
- }
- }
-
- /**
- * Parse namespace from a string. The string must be in legal format
- * defined by each {@link AllocationTagNamespaceType}.
- *
- * @param namespaceStr namespace string.
- * @return an instance of {@link AllocationTagNamespace}.
- * @throws InvalidAllocationTagsQueryException
- * if given string is not in valid format
- */
- public static AllocationTagNamespace parse(String namespaceStr)
- throws InvalidAllocationTagsQueryException {
- // Return the default namespace if no valid string is given.
- if (Strings.isNullOrEmpty(namespaceStr)) {
- return new Self();
- }
-
- // Normalize the input, escape additional chars.
- List<String> nsValues = normalize(namespaceStr);
- // The first string should be the prefix.
- String nsPrefix = nsValues.get(0);
- AllocationTagNamespaceType allocationTagNamespaceType =
- fromString(nsPrefix);
- switch (allocationTagNamespaceType) {
- case SELF:
- return new Self();
- case NOT_SELF:
- return new NotSelf();
- case ALL:
- return new All();
- case APP_ID:
- if (nsValues.size() != 2) {
- throw new InvalidAllocationTagsQueryException(
- "Missing the application ID in the namespace string: "
- + namespaceStr);
- }
- String appIDStr = nsValues.get(1);
- return parseAppID(appIDStr);
- case APP_LABEL:
- return new AppLabel();
- default:
- throw new InvalidAllocationTagsQueryException(
- "Invalid namespace string " + namespaceStr);
- }
- }
-
- private static AllocationTagNamespaceType fromString(String prefix) throws
- InvalidAllocationTagsQueryException {
- for (AllocationTagNamespaceType type :
- AllocationTagNamespaceType.values()) {
- if(type.getTypeKeyword().equals(prefix)) {
- return type;
- }
- }
-
- Set<String> values = Arrays.stream(AllocationTagNamespaceType.values())
- .map(AllocationTagNamespaceType::toString)
- .collect(Collectors.toSet());
- throw new InvalidAllocationTagsQueryException(
- "Invalid namespace prefix: " + prefix
- + ", valid values are: " + String.join(",", values));
- }
-
- private static AllocationTagNamespace parseAppID(String appIDStr)
- throws InvalidAllocationTagsQueryException {
- try {
- ApplicationId applicationId = ApplicationId.fromString(appIDStr);
- return new AppID(applicationId);
- } catch (IllegalArgumentException e) {
- throw new InvalidAllocationTagsQueryException(
- "Invalid application ID for "
- + APP_ID.getTypeKeyword() + ": " + appIDStr);
- }
- }
-
- /**
- * Valid given namespace string and parse it to a list of sub-strings
- * that can be consumed by the parser according to the type of the
- * namespace. Currently the size of return list should be either 1 or 2.
- * Extra slash is escaped during the normalization.
- *
- * @param namespaceStr namespace string.
- * @return a list of parsed strings.
- * @throws InvalidAllocationTagsQueryException
- * if namespace format is unexpected.
- */
- private static List<String> normalize(String namespaceStr)
- throws InvalidAllocationTagsQueryException {
- List<String> result = new ArrayList<>();
- if (namespaceStr == null) {
- return result;
- }
-
- String[] nsValues = namespaceStr.split(NAMESPACE_DELIMITER);
- for (String str : nsValues) {
- if (!Strings.isNullOrEmpty(str)) {
- result.add(str);
- }
- }
-
- // Currently we only allow 1 or 2 values for a namespace string
- if (result.size() == 0 || result.size() > 2) {
- throw new InvalidAllocationTagsQueryException("Invalid namespace string: "
- + namespaceStr + ", the syntax is <namespace_prefix> or"
- + " <namespace_prefix>/<namespace_value>");
- }
-
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
index dc0237e..5cae92c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import java.util.Set;
@@ -29,22 +28,34 @@ import java.util.Set;
*/
public final class AllocationTags {
- private AllocationTagNamespace ns;
+ private TargetApplicationsNamespace ns;
private Set<String> tags;
+ private ApplicationId applicationId;
- private AllocationTags(AllocationTagNamespace namespace,
+ private AllocationTags(TargetApplicationsNamespace namespace,
Set<String> allocationTags) {
this.ns = namespace;
this.tags = allocationTags;
}
+ private AllocationTags(TargetApplicationsNamespace namespace,
+ Set<String> allocationTags, ApplicationId currentAppId) {
+ this.ns = namespace;
+ this.tags = allocationTags;
+ this.applicationId = currentAppId;
+ }
+
/**
* @return the namespace of these tags.
*/
- public AllocationTagNamespace getNamespace() {
+ public TargetApplicationsNamespace getNamespace() {
return this.ns;
}
+ public ApplicationId getCurrentApplicationId() {
+ return this.applicationId;
+ }
+
/**
* @return the allocation tags.
*/
@@ -55,28 +66,31 @@ public final class AllocationTags {
@VisibleForTesting
public static AllocationTags createSingleAppAllocationTags(
ApplicationId appId, Set<String> tags) {
- AllocationTagNamespace namespace = new AllocationTagNamespace.AppID(appId);
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppID(appId);
return new AllocationTags(namespace, tags);
}
@VisibleForTesting
public static AllocationTags createGlobalAllocationTags(Set<String> tags) {
- AllocationTagNamespace namespace = new AllocationTagNamespace.All();
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.All();
return new AllocationTags(namespace, tags);
}
@VisibleForTesting
public static AllocationTags createOtherAppAllocationTags(
- ApplicationId currentApp, Set<ApplicationId> allIds, Set<String> tags)
- throws InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace = new AllocationTagNamespace.NotSelf();
- TargetApplications ta = new TargetApplications(currentApp, allIds);
- namespace.evaluate(ta);
- return new AllocationTags(namespace, tags);
+ ApplicationId currentApp, Set<String> tags) {
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.NotSelf();
+ return new AllocationTags(namespace, tags, currentApp);
}
- public static AllocationTags newAllocationTags(
- AllocationTagNamespace namespace, Set<String> tags) {
- return new AllocationTags(namespace, tags);
+ public static AllocationTags createAllocationTags(
+ ApplicationId currentApplicationId, String namespaceString,
+ Set<String> tags) throws InvalidAllocationTagsQueryException {
+ TargetApplicationsNamespace namespace = TargetApplicationsNamespace
+ .parse(namespaceString);
+ return new AllocationTags(namespace, tags, currentApplicationId);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
index 830566a..4fc2bab 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
@@ -21,7 +21,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -32,12 +31,14 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.log4j.Logger;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.LongBinaryOperator;
@@ -292,13 +293,21 @@ public class AllocationTagsManager {
/**
* Aggregates multiple {@link TypeToCountedTags} to a single one based on
- * a given set of application IDs, the values are properly merged.
+ * the scope defined in the allocation tags, the values are properly merged.
*
- * @param appIds a set of application IDs.
+ * @param allocationTags {@link AllocationTags}.
* @return an aggregated {@link TypeToCountedTags}.
*/
- private TypeToCountedTags aggregateAllocationTags(Set<ApplicationId> appIds,
- Map<ApplicationId, TypeToCountedTags> mapping) {
+ private TypeToCountedTags aggregateAllocationTags(
+ AllocationTags allocationTags,
+ Map<ApplicationId, TypeToCountedTags> mapping)
+ throws InvalidAllocationTagsQueryException {
+ // Based on the namespace type of the given allocation tags
+ TargetApplicationsNamespace namespace = allocationTags.getNamespace();
+ TargetApplications ta = new TargetApplications(
+ allocationTags.getCurrentApplicationId(), getApplicationIdToTags());
+ namespace.evaluate(ta);
+ Set<ApplicationId> appIds = namespace.getNamespaceScope();
TypeToCountedTags result = new TypeToCountedTags();
if (appIds != null) {
if (appIds.size() == 1) {
@@ -571,9 +580,7 @@ public class AllocationTagsManager {
mapping = globalNodeMapping;
} else {
// Aggregate app tags cardinality by applications.
- mapping = aggregateAllocationTags(
- tags.getNamespace().getNamespaceScope(),
- perAppNodeMappings);
+ mapping = aggregateAllocationTags(tags, perAppNodeMappings);
}
return mapping == null ? 0 :
@@ -618,9 +625,7 @@ public class AllocationTagsManager {
mapping = globalRackMapping;
} else {
// Aggregates cardinality by rack.
- mapping = aggregateAllocationTags(
- tags.getNamespace().getNamespaceScope(),
- perAppRackMappings);
+ mapping = aggregateAllocationTags(tags, perAppRackMappings);
}
return mapping == null ? 0 :
@@ -642,10 +647,22 @@ public class AllocationTagsManager {
}
/**
- * @return all application IDs in a set that currently visible by
- * the allocation tags manager.
+ * @return all applications that is known to the
+ * {@link AllocationTagsManager}, along with their application tags.
+ * The result is a map, where key is an application ID, and value is the
+ * application-tags attached to this application. If there is no
+ * application-tag exists for the application, the value is an empty set.
*/
- public Set<ApplicationId> getAllApplicationIds() {
- return ImmutableSet.copyOf(perAppNodeMappings.keySet());
+ private Map<ApplicationId, Set<String>> getApplicationIdToTags() {
+ Map<ApplicationId, Set<String>> result = new HashMap<>();
+ ConcurrentMap<ApplicationId, RMApp> allApps = rmContext.getRMApps();
+ if (allApps != null) {
+ for (Map.Entry<ApplicationId, RMApp> app : allApps.entrySet()) {
+ if (perAppNodeMappings.containsKey(app.getKey())) {
+ result.put(app.getKey(), app.getValue().getApplicationTags());
+ }
+ }
+ }
+ return result;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
index 389fc5c..efa7b65 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
@@ -24,7 +24,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.api.resource.PlacementConstraint;
@@ -58,35 +57,6 @@ public final class PlacementConstraintsUtil {
}
/**
- * Try to the namespace of the allocation tags from the given target key.
- *
- * @param targetKey
- * @return allocation tag namespace.
- * @throws InvalidAllocationTagsQueryException
- * if fail to parse the target key to a valid namespace.
- */
- private static AllocationTagNamespace getAllocationTagNamespace(
- ApplicationId currentAppId, String targetKey, AllocationTagsManager atm)
- throws InvalidAllocationTagsQueryException {
- // Parse to a valid namespace.
- AllocationTagNamespace namespace = AllocationTagNamespace.parse(targetKey);
-
- // TODO Complete remove this check once we support app-label.
- if (AllocationTagNamespaceType.APP_LABEL
- .equals(namespace.getNamespaceType())) {
- throw new InvalidAllocationTagsQueryException(
- namespace.toString() + " is not supported yet!");
- }
-
- // Evaluate the namespace according to the given target
- // before it can be consumed.
- TargetApplications ta =
- new TargetApplications(currentAppId, atm.getAllApplicationIds());
- namespace.evaluate(ta);
- return namespace;
- }
-
- /**
* Returns true if <b>single</b> placement constraint with associated
* allocationTags and scope is satisfied by a specific scheduler Node.
*
@@ -104,13 +74,10 @@ public final class PlacementConstraintsUtil {
ApplicationId targetApplicationId, SingleConstraint sc,
TargetExpression te, SchedulerNode node, AllocationTagsManager tm)
throws InvalidAllocationTagsQueryException {
- // Parse the allocation tag's namespace from the given target key,
- // then evaluate the namespace and get its scope,
- // which is represented by one or more application IDs.
- AllocationTagNamespace namespace = getAllocationTagNamespace(
- targetApplicationId, te.getTargetKey(), tm);
- AllocationTags allocationTags = AllocationTags
- .newAllocationTags(namespace, te.getTargetValues());
+ // Creates AllocationTags that will be further consumed by allocation
+ // tags manager for cardinality check.
+ AllocationTags allocationTags = AllocationTags.createAllocationTags(
+ targetApplicationId, te.getTargetKey(), te.getTargetValues());
long minScopeCardinality = 0;
long maxScopeCardinality = 0;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
index 0de7c9e..ab1bd9b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
@@ -18,34 +18,77 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.yarn.api.records.ApplicationId;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
/**
* This class is used by
- * {@link AllocationTagNamespace#evaluate(TargetApplications)} to evaluate
+ * {@link TargetApplicationsNamespace#evaluate(TargetApplications)} to evaluate
* a namespace.
*/
public class TargetApplications {
private ApplicationId currentAppId;
- private Set<ApplicationId> allAppIds;
+ private Map<ApplicationId, Set<String>> allApps;
public TargetApplications(ApplicationId currentApplicationId,
Set<ApplicationId> allApplicationIds) {
this.currentAppId = currentApplicationId;
- this.allAppIds = allApplicationIds;
+ allApps = new HashMap<>();
+ if (allApplicationIds != null) {
+ allApplicationIds.forEach(appId ->
+ allApps.put(appId, ImmutableSet.of()));
+ }
+ }
+
+ public TargetApplications(ApplicationId currentApplicationId,
+ Map<ApplicationId, Set<String>> allApplicationIds) {
+ this.currentAppId = currentApplicationId;
+ this.allApps = allApplicationIds;
}
public ApplicationId getCurrentApplicationId() {
return this.currentAppId;
}
+ public Set<ApplicationId> getAllApplicationIds() {
+ return this.allApps == null ?
+ ImmutableSet.of() : allApps.keySet();
+ }
+
public Set<ApplicationId> getOtherApplicationIds() {
- return allAppIds == null ? null : allAppIds.stream().filter(appId ->
- !appId.equals(getCurrentApplicationId()))
+ if (getAllApplicationIds() == null
+ || getAllApplicationIds().isEmpty()) {
+ return ImmutableSet.of();
+ }
+ return getAllApplicationIds()
+ .stream()
+ .filter(appId -> !appId.equals(getCurrentApplicationId()))
.collect(Collectors.toSet());
}
+
+ public Set<ApplicationId> getApplicationIdsByTag(String applicationTag) {
+ Set<ApplicationId> result = new HashSet<>();
+ if (Strings.isNullOrEmpty(applicationTag)
+ || this.allApps == null) {
+ return result;
+ }
+
+ for (Map.Entry<ApplicationId, Set<String>> app
+ : this.allApps.entrySet()) {
+ if (app.getValue() != null
+ && app.getValue().contains(applicationTag)) {
+ result.add(app.getKey());
+ }
+ }
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
new file mode 100644
index 0000000..a82a03f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
@@ -0,0 +1,326 @@
+/**
+ * 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.constraint;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
+import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.SELF;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.NOT_SELF;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_TAG;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_ID;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.ALL;
+
+/**
+ * Class to describe the namespace of allocation tags, used by
+ * {@link AllocationTags}. Each namespace can be evaluated against
+ * a target set applications, represented by {@link TargetApplications}.
+ * After evaluation, the namespace is interpreted to be a set of
+ * applications based on the namespace type.
+ */
+public abstract class TargetApplicationsNamespace implements
+ Evaluable<TargetApplications> {
+
+ public final static String NAMESPACE_DELIMITER = "/";
+
+ private AllocationTagNamespaceType nsType;
+ // Namespace scope value will be delay binding by eval method.
+ private Set<ApplicationId> nsScope;
+
+ public TargetApplicationsNamespace(AllocationTagNamespaceType
+ allocationTagNamespaceType) {
+ this.nsType = allocationTagNamespaceType;
+ }
+
+ protected void setScopeIfNotNull(Set<ApplicationId> appIds) {
+ if (appIds != null) {
+ this.nsScope = appIds;
+ }
+ }
+
+ /**
+ * Get the type of the namespace.
+ * @return namespace type.
+ */
+ public AllocationTagNamespaceType getNamespaceType() {
+ return nsType;
+ }
+
+ /**
+ * Get the scope of the namespace, in form of a set of applications.
+ *
+ * @return a set of applications.
+ */
+ public Set<ApplicationId> getNamespaceScope() {
+ if (this.nsScope == null) {
+ throw new IllegalStateException("Invalid namespace scope,"
+ + " it is not initialized. Evaluate must be called before"
+ + " a namespace can be consumed.");
+ }
+ return this.nsScope;
+ }
+
+ /**
+ * Evaluate the namespace against given target applications
+ * if it is necessary. Only self/not-self/app-label namespace types
+ * require this evaluation step, because they are not binding to a
+ * specific scope during initiating. So we do lazy binding for them
+ * in this method.
+ *
+ * @param target a generic type target that impacts this evaluation.
+ * @throws InvalidAllocationTagsQueryException
+ */
+ @Override
+ public void evaluate(TargetApplications target)
+ throws InvalidAllocationTagsQueryException {
+ // Sub-class needs to override this when it requires the eval step.
+ }
+
+ @Override
+ public String toString() {
+ return this.nsType.toString();
+ }
+
+ /**
+ * Namespace within application itself.
+ */
+ public static class Self extends TargetApplicationsNamespace {
+
+ public Self() {
+ super(SELF);
+ }
+
+ @Override
+ public void evaluate(TargetApplications target)
+ throws InvalidAllocationTagsQueryException {
+ if (target == null || target.getCurrentApplicationId() == null) {
+ throw new InvalidAllocationTagsQueryException("Namespace Self must"
+ + " be evaluated against a single application ID.");
+ }
+ ApplicationId applicationId = target.getCurrentApplicationId();
+ setScopeIfNotNull(ImmutableSet.of(applicationId));
+ }
+ }
+
+ /**
+ * Namespace to all applications except itself.
+ */
+ public static class NotSelf extends TargetApplicationsNamespace {
+
+ private ApplicationId applicationId;
+
+ public NotSelf() {
+ super(NOT_SELF);
+ }
+
+ /**
+ * The scope of self namespace is to an application itself,
+ * the application ID can be delay binding to the namespace.
+ *
+ * @param appId application ID.
+ */
+ public void setApplicationId(ApplicationId appId) {
+ this.applicationId = appId;
+ }
+
+ public ApplicationId getApplicationId() {
+ return this.applicationId;
+ }
+
+ @Override
+ public void evaluate(TargetApplications target) {
+ Set<ApplicationId> otherAppIds = target.getOtherApplicationIds();
+ setScopeIfNotNull(otherAppIds);
+ }
+ }
+
+ /**
+ * Namespace to all applications in the cluster.
+ */
+ public static class All extends TargetApplicationsNamespace {
+
+ public All() {
+ super(ALL);
+ }
+ }
+
+ /**
+ * Namespace to applications that attached with a certain application tag.
+ */
+ public static class AppTag extends TargetApplicationsNamespace {
+
+ private String applicationTag;
+
+ public AppTag(String appTag) {
+ super(APP_TAG);
+ this.applicationTag = appTag;
+ }
+
+ @Override
+ public void evaluate(TargetApplications target) {
+ setScopeIfNotNull(target.getApplicationIdsByTag(applicationTag));
+ }
+
+ @Override
+ public String toString() {
+ return APP_TAG.toString() + NAMESPACE_DELIMITER + this.applicationTag;
+ }
+ }
+
+ /**
+ * Namespace defined by a certain application ID.
+ */
+ public static class AppID extends TargetApplicationsNamespace {
+
+ private ApplicationId targetAppId;
+ // app-id namespace requires an extra value of an application id.
+ public AppID(ApplicationId applicationId) {
+ super(APP_ID);
+ this.targetAppId = applicationId;
+ setScopeIfNotNull(ImmutableSet.of(targetAppId));
+ }
+
+ @Override
+ public String toString() {
+ return APP_ID.toString() + NAMESPACE_DELIMITER + this.targetAppId;
+ }
+ }
+
+ /**
+ * Parse namespace from a string. The string must be in legal format
+ * defined by each {@link AllocationTagNamespaceType}.
+ *
+ * @param namespaceStr namespace string.
+ * @return an instance of {@link TargetApplicationsNamespace}.
+ * @throws InvalidAllocationTagsQueryException
+ * if given string is not in valid format
+ */
+ public static TargetApplicationsNamespace parse(String namespaceStr)
+ throws InvalidAllocationTagsQueryException {
+ // Return the default namespace if no valid string is given.
+ if (Strings.isNullOrEmpty(namespaceStr)) {
+ return new Self();
+ }
+
+ // Normalize the input, escape additional chars.
+ List<String> nsValues = normalize(namespaceStr);
+ // The first string should be the prefix.
+ String nsPrefix = nsValues.get(0);
+ AllocationTagNamespaceType allocationTagNamespaceType =
+ fromString(nsPrefix);
+ switch (allocationTagNamespaceType) {
+ case SELF:
+ return new Self();
+ case NOT_SELF:
+ return new NotSelf();
+ case ALL:
+ return new All();
+ case APP_ID:
+ if (nsValues.size() != 2) {
+ throw new InvalidAllocationTagsQueryException(
+ "Missing the application ID in the namespace string: "
+ + namespaceStr);
+ }
+ String appIDStr = nsValues.get(1);
+ return parseAppID(appIDStr);
+ case APP_TAG:
+ if (nsValues.size() != 2) {
+ throw new InvalidAllocationTagsQueryException(
+ "Missing the application tag in the namespace string: "
+ + namespaceStr);
+ }
+ return new AppTag(nsValues.get(1));
+ default:
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid namespace string " + namespaceStr);
+ }
+ }
+
+ private static AllocationTagNamespaceType fromString(String prefix) throws
+ InvalidAllocationTagsQueryException {
+ for (AllocationTagNamespaceType type :
+ AllocationTagNamespaceType.values()) {
+ if(type.getTypeKeyword().equals(prefix)) {
+ return type;
+ }
+ }
+
+ Set<String> values = Arrays.stream(AllocationTagNamespaceType.values())
+ .map(AllocationTagNamespaceType::toString)
+ .collect(Collectors.toSet());
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid namespace prefix: " + prefix
+ + ", valid values are: " + String.join(",", values));
+ }
+
+ private static TargetApplicationsNamespace parseAppID(String appIDStr)
+ throws InvalidAllocationTagsQueryException {
+ try {
+ ApplicationId applicationId = ApplicationId.fromString(appIDStr);
+ return new AppID(applicationId);
+ } catch (IllegalArgumentException e) {
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid application ID for "
+ + APP_ID.getTypeKeyword() + ": " + appIDStr);
+ }
+ }
+
+ /**
+ * Valid given namespace string and parse it to a list of sub-strings
+ * that can be consumed by the parser according to the type of the
+ * namespace. Currently the size of return list should be either 1 or 2.
+ * Extra slash is escaped during the normalization.
+ *
+ * @param namespaceStr namespace string.
+ * @return a list of parsed strings.
+ * @throws InvalidAllocationTagsQueryException
+ * if namespace format is unexpected.
+ */
+ private static List<String> normalize(String namespaceStr)
+ throws InvalidAllocationTagsQueryException {
+ List<String> result = new ArrayList<>();
+ if (namespaceStr == null) {
+ return result;
+ }
+
+ String[] nsValues = namespaceStr.split(NAMESPACE_DELIMITER);
+ for (String str : nsValues) {
+ if (!Strings.isNullOrEmpty(str)) {
+ result.add(str);
+ }
+ }
+
+ // Currently we only allow 1 or 2 values for a namespace string
+ if (result.size() == 0 || result.size() > 2) {
+ throw new InvalidAllocationTagsQueryException("Invalid namespace string: "
+ + namespaceStr + ", the syntax is <namespace_prefix> or"
+ + " <namespace_prefix>/<namespace_value>");
+ }
+
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
index 9004110..1fc6bad 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
@@ -23,8 +23,6 @@ import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.AllocationTagNamespace;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
@@ -334,25 +332,6 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
targetAllocationTags = new HashSet<>(
targetExpression.getTargetValues());
-
- try {
- AllocationTagNamespace tagNS =
- AllocationTagNamespace.parse(targetExpression.getTargetKey());
- if (AllocationTagNamespaceType.APP_LABEL
- .equals(tagNS.getNamespaceType())) {
- throwExceptionWithMetaInfo(
- "As of now, allocation tag namespace ["
- + AllocationTagNamespaceType.APP_LABEL.toString()
- + "] is not supported. Please make changes to placement "
- + "constraints accordingly. If this is null, it will be "
- + "set to "
- + AllocationTagNamespaceType.SELF.toString()
- + " by default.");
- }
- } catch (InvalidAllocationTagsQueryException e) {
- throwExceptionWithMetaInfo(
- "Invalid allocation tag namespace, message: " + e.getMessage());
- }
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
index c399368..664fae2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
@@ -68,6 +68,7 @@ public class MockRMApp implements RMApp {
RMAppAttempt attempt;
int maxAppAttempts = 1;
List<ResourceRequest> amReqs;
+ private Set<String> applicationTags = null;
public MockRMApp(int newid, long time, RMAppState newState) {
finish = time;
@@ -82,6 +83,12 @@ public class MockRMApp implements RMApp {
user = userName;
}
+ public MockRMApp(int newid, long time, RMAppState newState,
+ String userName, Set<String> appTags) {
+ this(newid, time, newState, userName);
+ this.applicationTags = appTags;
+ }
+
public MockRMApp(int newid, long time, RMAppState newState, String userName, String diag) {
this(newid, time, newState, userName);
this.diagnostics = new StringBuilder(diag);
@@ -248,7 +255,7 @@ public class MockRMApp implements RMApp {
@Override
public Set<String> getApplicationTags() {
- return null;
+ return this.applicationTags;
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
index d7124bb..13247a7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
@@ -20,7 +20,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.AllocationTagNamespace;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.TargetApplicationsNamespace;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
@@ -294,7 +294,8 @@ public class TestSchedulingRequestContainerAllocation {
// App2 asks for 3 containers that anti-affinity with any mapper,
// since 3 out of 4 nodes already have mapper containers, all 3
// containers will be allocated on the other node.
- AllocationTagNamespace.All allNs = new AllocationTagNamespace.All();
+ TargetApplicationsNamespace.All allNs =
+ new TargetApplicationsNamespace.All();
am2.allocateAppAntiAffinity(
ResourceSizing.newInstance(3, Resource.newInstance(1024, 1)),
Priority.newInstance(1), 1L, allNs.toString(),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
index cbf5968..3f2aaed 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
@@ -27,13 +27,19 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
/**
* Test functionality of AllocationTagsManager.
@@ -468,15 +474,27 @@ public class TestAllocationTagsManager {
@Test
public void testNodeAllocationTagsAggregation()
throws InvalidAllocationTagsQueryException {
+ RMContext mockContext = Mockito.spy(rmContext);
- AllocationTagsManager atm = new AllocationTagsManager(rmContext);
ApplicationId app1 = TestUtils.getMockApplicationId(1);
ApplicationId app2 = TestUtils.getMockApplicationId(2);
ApplicationId app3 = TestUtils.getMockApplicationId(3);
+
NodeId host1 = NodeId.fromString("host1:123");
NodeId host2 = NodeId.fromString("host2:123");
NodeId host3 = NodeId.fromString("host3:123");
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(app1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(app2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(app3, new MockRMApp(125, 1002,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ Mockito.when(mockContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager atm = new AllocationTagsManager(mockContext);
+
/**
* Node1 (rack0)
* app1/A(2)
@@ -561,7 +579,7 @@ public class TestAllocationTagsManager {
*
*/
tags = AllocationTags.createOtherAppAllocationTags(app1,
- ImmutableSet.of(app1, app2, app3), ImmutableSet.of("A", "B"));
+ ImmutableSet.of("A", "B"));
Assert.assertEquals(4, atm.getNodeCardinalityByOp(host1, tags, Long::max));
Assert.assertEquals(0, atm.getNodeCardinalityByOp(host1, tags, Long::min));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
index d1ef331..27a121a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
@@ -16,50 +16,67 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint; /**
* limitations under the License.
*/
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.junit.Assert;
import org.junit.Test;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
/**
- * Test class for {@link AllocationTagNamespace}.
+ * Test class for {@link TargetApplicationsNamespace}.
*/
public class TestAllocationTagsNamespace {
@Test
public void testNamespaceParse() throws InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace;
+ TargetApplicationsNamespace namespace;
String namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.SELF,
namespace.getNamespaceType());
namespaceStr = "not-self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.NOT_SELF,
namespace.getNamespaceType());
namespaceStr = "all";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.ALL,
namespace.getNamespaceType());
- namespaceStr = "app-label";
- namespace = AllocationTagNamespace.parse(namespaceStr);
- Assert.assertEquals(AllocationTagNamespaceType.APP_LABEL,
+ namespaceStr = "app-tag/spark-jobs";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ Assert.assertEquals(AllocationTagNamespaceType.APP_TAG,
namespace.getNamespaceType());
+ // Invalid app-tag namespace syntax
+ try {
+ namespaceStr = "app-tag/tag123/tag234";
+ TargetApplicationsNamespace.parse(namespaceStr);
+ Assert.fail("Parsing should fail as the given namespace is invalid");
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
+ Assert.assertTrue(e.getMessage().startsWith(
+ "Invalid namespace string"));
+ }
+
ApplicationId applicationId = ApplicationId.newInstance(12345, 1);
namespaceStr = "app-id/" + applicationId.toString();
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.APP_ID,
namespace.getNamespaceType());
// Invalid app-id namespace syntax, invalid app ID.
try {
namespaceStr = "app-id/apppppp_12345_99999";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the given app ID is invalid");
} catch (Exception e) {
Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
@@ -70,7 +87,7 @@ public class TestAllocationTagsNamespace {
// Invalid app-id namespace syntax, missing app ID.
try {
namespaceStr = "app-id";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the given namespace"
+ " is missing application ID");
} catch (Exception e) {
@@ -82,7 +99,7 @@ public class TestAllocationTagsNamespace {
// Invalid namespace type.
try {
namespaceStr = "non_exist_ns";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the giving type is not supported.");
} catch (Exception e) {
Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
@@ -94,7 +111,7 @@ public class TestAllocationTagsNamespace {
@Test
public void testNamespaceEvaluation() throws
InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace;
+ TargetApplicationsNamespace namespace;
TargetApplications targetApplications;
ApplicationId app1 = ApplicationId.newInstance(10000, 1);
ApplicationId app2 = ApplicationId.newInstance(10000, 2);
@@ -104,7 +121,7 @@ public class TestAllocationTagsNamespace {
// Ensure eval is called before using the scope.
String namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
try {
namespace.getNamespaceScope();
Assert.fail("Call getNamespaceScope before evaluate is not allowed.");
@@ -115,14 +132,14 @@ public class TestAllocationTagsNamespace {
}
namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1, ImmutableSet.of(app1));
namespace.evaluate(targetApplications);
Assert.assertEquals(1, namespace.getNamespaceScope().size());
Assert.assertEquals(app1, namespace.getNamespaceScope().iterator().next());
namespaceStr = "not-self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1, ImmutableSet.of(app1));
namespace.evaluate(targetApplications);
Assert.assertEquals(0, namespace.getNamespaceScope().size());
@@ -134,16 +151,52 @@ public class TestAllocationTagsNamespace {
Assert.assertFalse(namespace.getNamespaceScope().contains(app1));
namespaceStr = "all";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.ALL,
namespace.getNamespaceType());
namespaceStr = "app-id/" + app2.toString();
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1,
ImmutableSet.of(app1, app2, app3, app4, app5));
namespace.evaluate(targetApplications);
Assert.assertEquals(1, namespace.getNamespaceScope().size());
Assert.assertEquals(app2, namespace.getNamespaceScope().iterator().next());
+
+ /**
+ * App to Application Tags
+ * app1: A, B
+ * app2: A
+ * app3:
+ * app4: C
+ * app5: A, B, C
+ */
+ Map<ApplicationId, Set<String>> appsWithTags = new HashMap<>();
+ appsWithTags.put(app1, ImmutableSet.of("A", "B"));
+ appsWithTags.put(app2, ImmutableSet.of("A"));
+ appsWithTags.put(app3, ImmutableSet.of());
+ appsWithTags.put(app4, ImmutableSet.of("C"));
+ appsWithTags.put(app5, ImmutableSet.of("A", "B", "C"));
+
+ namespaceStr = "app-tag/A";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ targetApplications = new TargetApplications(app1, appsWithTags);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(3, namespace.getNamespaceScope().size());
+ Assert.assertTrue(Sets.difference(namespace.getNamespaceScope(),
+ ImmutableSet.of(app1, app2, app5)).isEmpty());
+
+ namespaceStr = "app-tag/B";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(2, namespace.getNamespaceScope().size());
+ Assert.assertTrue(Sets.difference(namespace.getNamespaceScope(),
+ ImmutableSet.of(app1, app5)).isEmpty());
+
+ // Not exist
+ namespaceStr = "app-tag/xyz";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(0, namespace.getNamespaceScope().size());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7853ec8d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
index 4814321..3248450 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
@@ -37,6 +37,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.concurrent.atomic.AtomicLong;
@@ -52,6 +54,9 @@ import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
@@ -63,6 +68,7 @@ import org.junit.Before;
import org.junit.Test;
import com.google.common.collect.ImmutableSet;
+import org.mockito.Mockito;
/**
* Test the PlacementConstraint Utility class functionality.
@@ -562,8 +568,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace namespaceAll =
- new AllocationTagNamespace.All();
+ TargetApplicationsNamespace namespaceAll =
+ new TargetApplicationsNamespace.All();
//***************************
// 1) all, anti-affinity
@@ -648,17 +654,28 @@ public class TestPlacementConstraintsUtil {
@Test
public void testNotSelfAppConstraints()
throws InvalidAllocationTagsQueryException {
- AllocationTagsManager tm = new AllocationTagsManager(rmContext);
- PlacementConstraintManagerService pcm =
- new MemoryPlacementConstraintManager();
- rmContext.setAllocationTagsManager(tm);
- rmContext.setPlacementConstraintManager(pcm);
-
long ts = System.currentTimeMillis();
ApplicationId application1 = BuilderUtils.newApplicationId(ts, 100);
ApplicationId application2 = BuilderUtils.newApplicationId(ts, 101);
ApplicationId application3 = BuilderUtils.newApplicationId(ts, 102);
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(application1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(application2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(application3, new MockRMApp(125, 1002,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+
+ RMContext mockedContext = Mockito.spy(rmContext);
+ when(mockedContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager tm = new AllocationTagsManager(mockedContext);
+ PlacementConstraintManagerService pcm =
+ new MemoryPlacementConstraintManager();
+ mockedContext.setAllocationTagsManager(tm);
+ mockedContext.setPlacementConstraintManager(pcm);
+
// Register App1 with anti-affinity constraint map.
RMNode n0r1 = rmNodes.get(0);
RMNode n1r1 = rmNodes.get(1);
@@ -696,8 +713,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace notSelf =
- new AllocationTagNamespace.NotSelf();
+ TargetApplicationsNamespace notSelf =
+ new TargetApplicationsNamespace.NotSelf();
//***************************
// 1) not-self, app1
@@ -800,8 +817,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 =newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace namespace =
- new AllocationTagNamespace.AppID(application1);
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppID(application1);
Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>();
PlacementConstraint constraint2 = PlacementConstraints
.targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(),
@@ -832,7 +849,7 @@ public class TestPlacementConstraintsUtil {
// Intra-app constraint
// Test with default and empty namespace
- AllocationTagNamespace self = new AllocationTagNamespace.Self();
+ TargetApplicationsNamespace self = new TargetApplicationsNamespace.Self();
PlacementConstraint constraint3 = PlacementConstraints
.targetNotIn(NODE, allocationTagWithNamespace(self.toString(),
"hbase-m"))
@@ -873,6 +890,88 @@ public class TestPlacementConstraintsUtil {
}
@Test
+ public void testInterAppConstriantsByAppTag()
+ throws InvalidAllocationTagsQueryException {
+
+ ApplicationId application1 = BuilderUtils.newApplicationId(1000, 123);
+ ApplicationId application2 = BuilderUtils.newApplicationId(1001, 124);
+
+ // app1: test-tag
+ // app2: N/A
+ RMContext mockedContext = Mockito.spy(rmContext);
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(application1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("test-tag")));
+ allApps.put(application2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ when(mockedContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager tm = new AllocationTagsManager(mockedContext);
+ PlacementConstraintManagerService pcm =
+ new MemoryPlacementConstraintManager();
+ mockedContext.setAllocationTagsManager(tm);
+ mockedContext.setPlacementConstraintManager(pcm);
+
+ // Register App1 with anti-affinity constraint map.
+ RMNode n0r1 = rmNodes.get(0);
+ RMNode n1r1 = rmNodes.get(1);
+ RMNode n2r2 = rmNodes.get(2);
+ RMNode n3r2 = rmNodes.get(3);
+
+ /**
+ * Place container:
+ * n0: app1/hbase-m(1)
+ * n1: ""
+ * n2: app1/hbase-m(1)
+ * n3: ""
+ */
+ tm.addContainer(n0r1.getNodeID(),
+ newContainerId(application1), ImmutableSet.of("hbase-m"));
+ tm.addContainer(n2r2.getNodeID(),
+ newContainerId(application1), ImmutableSet.of("hbase-m"));
+
+ SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(),
+ n0r1.getRackName(), n0r1.getNodeID());
+ SchedulerNode schedulerNode1 = newSchedulerNode(n1r1.getHostName(),
+ n1r1.getRackName(), n1r1.getNodeID());
+ SchedulerNode schedulerNode2 = newSchedulerNode(n2r2.getHostName(),
+ n2r2.getRackName(), n2r2.getNodeID());
+ SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
+ n3r2.getRackName(), n3r2.getNodeID());
+
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppTag("test-tag");
+ Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>();
+ PlacementConstraint constraint2 = PlacementConstraints
+ .targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(),
+ "hbase-m"))
+ .build();
+ Set<String> srcTags2 = ImmutableSet.of("app2");
+ constraintMap.put(srcTags2, constraint2);
+
+ pcm.registerApplication(application2, constraintMap);
+
+ // Anti-affinity with app-tag/test-tag/hbase-m,
+ // app1 has tag "test-tag" so the constraint is equally to work on app1
+ // onto n1 and n3 as they don't have "hbase-m" from app1.
+ Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode0, pcm, tm));
+ Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode1, pcm, tm));
+ Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode2, pcm, tm));
+ Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode3, pcm, tm));
+
+ pcm.unregisterApplication(application1);
+ pcm.unregisterApplication(application2);
+ }
+
+ @Test
public void testInvalidAllocationTagNamespace() {
AllocationTagsManager tm = new AllocationTagsManager(rmContext);
PlacementConstraintManagerService pcm =
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/2] hadoop git commit: YARN-8013. Support application tags when
defining application namespaces for placement constraints. Contributed by
Weiwei Yang.
Posted by kk...@apache.org.
YARN-8013. Support application tags when defining application namespaces for placement constraints. Contributed by Weiwei Yang.
(cherry picked from commit 7853ec8d2fb8731b7f7c28fd87491a0a2d47967e)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/fa464c12
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/fa464c12
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/fa464c12
Branch: refs/heads/branch-3.1
Commit: fa464c12ad67228ec6dfd20f9217c0630c9adf23
Parents: 4db13cb
Author: Konstantinos Karanasos <kk...@apache.org>
Authored: Wed Apr 4 10:51:58 2018 -0700
Committer: Konstantinos Karanasos <kk...@apache.org>
Committed: Wed Apr 4 10:53:33 2018 -0700
----------------------------------------------------------------------
.../api/records/AllocationTagNamespaceType.java | 2 +-
.../constraint/AllocationTagNamespace.java | 312 ------------------
.../scheduler/constraint/AllocationTags.java | 44 ++-
.../constraint/AllocationTagsManager.java | 47 ++-
.../constraint/PlacementConstraintsUtil.java | 41 +--
.../constraint/TargetApplications.java | 53 ++-
.../constraint/TargetApplicationsNamespace.java | 326 +++++++++++++++++++
.../SingleConstraintAppPlacementAllocator.java | 21 --
.../server/resourcemanager/rmapp/MockRMApp.java | 9 +-
...estSchedulingRequestContainerAllocation.java | 5 +-
.../constraint/TestAllocationTagsManager.java | 22 +-
.../constraint/TestAllocationTagsNamespace.java | 89 ++++-
.../TestPlacementConstraintsUtil.java | 125 ++++++-
13 files changed, 654 insertions(+), 442 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
index de5492e..f304600 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/AllocationTagNamespaceType.java
@@ -26,7 +26,7 @@ public enum AllocationTagNamespaceType {
SELF("self"),
NOT_SELF("not-self"),
APP_ID("app-id"),
- APP_LABEL("app-label"),
+ APP_TAG("app-tag"),
ALL("all");
private String typeKeyword;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
deleted file mode 100644
index 7b9f3be..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagNamespace.java
+++ /dev/null
@@ -1,312 +0,0 @@
-/**
- * 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.constraint;
-
-import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableSet;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.SELF;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.NOT_SELF;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_LABEL;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_ID;
-import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.ALL;
-
-/**
- * Class to describe the namespace of an allocation tag.
- * Each namespace can be evaluated against a set of applications.
- * After evaluation, the namespace should have an implicit set of
- * applications which defines its scope.
- */
-public abstract class AllocationTagNamespace implements
- Evaluable<TargetApplications> {
-
- public final static String NAMESPACE_DELIMITER = "/";
-
- private AllocationTagNamespaceType nsType;
- // Namespace scope value will be delay binding by eval method.
- private Set<ApplicationId> nsScope;
-
- public AllocationTagNamespace(AllocationTagNamespaceType
- allocationTagNamespaceType) {
- this.nsType = allocationTagNamespaceType;
- }
-
- protected void setScopeIfNotNull(Set<ApplicationId> appIds) {
- if (appIds != null) {
- this.nsScope = appIds;
- }
- }
-
- /**
- * Get the type of the namespace.
- * @return namespace type.
- */
- public AllocationTagNamespaceType getNamespaceType() {
- return nsType;
- }
-
- /**
- * Get the scope of the namespace, in form of a set of applications.
- *
- * @return a set of applications.
- */
- public Set<ApplicationId> getNamespaceScope() {
- if (this.nsScope == null) {
- throw new IllegalStateException("Invalid namespace scope,"
- + " it is not initialized. Evaluate must be called before"
- + " a namespace can be consumed.");
- }
- return this.nsScope;
- }
-
- /**
- * Evaluate the namespace against given target applications
- * if it is necessary. Only self/not-self/app-label namespace types
- * require this evaluation step, because they are not binding to a
- * specific scope during initiating. So we do lazy binding for them
- * in this method.
- *
- * @param target a generic type target that impacts this evaluation.
- * @throws InvalidAllocationTagsQueryException
- */
- @Override
- public void evaluate(TargetApplications target)
- throws InvalidAllocationTagsQueryException {
- // Sub-class needs to override this when it requires the eval step.
- }
-
- @Override
- public String toString() {
- return this.nsType.toString();
- }
-
- /**
- * Namespace within application itself.
- */
- public static class Self extends AllocationTagNamespace {
-
- public Self() {
- super(SELF);
- }
-
- @Override
- public void evaluate(TargetApplications target)
- throws InvalidAllocationTagsQueryException {
- if (target == null || target.getCurrentApplicationId() == null) {
- throw new InvalidAllocationTagsQueryException("Namespace Self must"
- + " be evaluated against a single application ID.");
- }
- ApplicationId applicationId = target.getCurrentApplicationId();
- setScopeIfNotNull(ImmutableSet.of(applicationId));
- }
- }
-
- /**
- * Namespace to all applications except itself.
- */
- public static class NotSelf extends AllocationTagNamespace {
-
- private ApplicationId applicationId;
-
- public NotSelf() {
- super(NOT_SELF);
- }
-
- /**
- * The scope of self namespace is to an application itself,
- * the application ID can be delay binding to the namespace.
- *
- * @param appId application ID.
- */
- public void setApplicationId(ApplicationId appId) {
- this.applicationId = appId;
- }
-
- public ApplicationId getApplicationId() {
- return this.applicationId;
- }
-
- @Override
- public void evaluate(TargetApplications target) {
- Set<ApplicationId> otherAppIds = target.getOtherApplicationIds();
- setScopeIfNotNull(otherAppIds);
- }
- }
-
- /**
- * Namespace to all applications in the cluster.
- */
- public static class All extends AllocationTagNamespace {
-
- public All() {
- super(ALL);
- }
- }
-
- /**
- * Namespace to all applications in the cluster.
- */
- public static class AppLabel extends AllocationTagNamespace {
-
- public AppLabel() {
- super(APP_LABEL);
- }
-
- @Override
- public void evaluate(TargetApplications target) {
- // TODO Implement app-label namespace evaluation
- }
- }
-
- /**
- * Namespace defined by a certain application ID.
- */
- public static class AppID extends AllocationTagNamespace {
-
- private ApplicationId targetAppId;
- // app-id namespace requires an extra value of an application id.
- public AppID(ApplicationId applicationId) {
- super(APP_ID);
- this.targetAppId = applicationId;
- setScopeIfNotNull(ImmutableSet.of(targetAppId));
- }
-
- @Override
- public String toString() {
- return APP_ID.toString() + NAMESPACE_DELIMITER + this.targetAppId;
- }
- }
-
- /**
- * Parse namespace from a string. The string must be in legal format
- * defined by each {@link AllocationTagNamespaceType}.
- *
- * @param namespaceStr namespace string.
- * @return an instance of {@link AllocationTagNamespace}.
- * @throws InvalidAllocationTagsQueryException
- * if given string is not in valid format
- */
- public static AllocationTagNamespace parse(String namespaceStr)
- throws InvalidAllocationTagsQueryException {
- // Return the default namespace if no valid string is given.
- if (Strings.isNullOrEmpty(namespaceStr)) {
- return new Self();
- }
-
- // Normalize the input, escape additional chars.
- List<String> nsValues = normalize(namespaceStr);
- // The first string should be the prefix.
- String nsPrefix = nsValues.get(0);
- AllocationTagNamespaceType allocationTagNamespaceType =
- fromString(nsPrefix);
- switch (allocationTagNamespaceType) {
- case SELF:
- return new Self();
- case NOT_SELF:
- return new NotSelf();
- case ALL:
- return new All();
- case APP_ID:
- if (nsValues.size() != 2) {
- throw new InvalidAllocationTagsQueryException(
- "Missing the application ID in the namespace string: "
- + namespaceStr);
- }
- String appIDStr = nsValues.get(1);
- return parseAppID(appIDStr);
- case APP_LABEL:
- return new AppLabel();
- default:
- throw new InvalidAllocationTagsQueryException(
- "Invalid namespace string " + namespaceStr);
- }
- }
-
- private static AllocationTagNamespaceType fromString(String prefix) throws
- InvalidAllocationTagsQueryException {
- for (AllocationTagNamespaceType type :
- AllocationTagNamespaceType.values()) {
- if(type.getTypeKeyword().equals(prefix)) {
- return type;
- }
- }
-
- Set<String> values = Arrays.stream(AllocationTagNamespaceType.values())
- .map(AllocationTagNamespaceType::toString)
- .collect(Collectors.toSet());
- throw new InvalidAllocationTagsQueryException(
- "Invalid namespace prefix: " + prefix
- + ", valid values are: " + String.join(",", values));
- }
-
- private static AllocationTagNamespace parseAppID(String appIDStr)
- throws InvalidAllocationTagsQueryException {
- try {
- ApplicationId applicationId = ApplicationId.fromString(appIDStr);
- return new AppID(applicationId);
- } catch (IllegalArgumentException e) {
- throw new InvalidAllocationTagsQueryException(
- "Invalid application ID for "
- + APP_ID.getTypeKeyword() + ": " + appIDStr);
- }
- }
-
- /**
- * Valid given namespace string and parse it to a list of sub-strings
- * that can be consumed by the parser according to the type of the
- * namespace. Currently the size of return list should be either 1 or 2.
- * Extra slash is escaped during the normalization.
- *
- * @param namespaceStr namespace string.
- * @return a list of parsed strings.
- * @throws InvalidAllocationTagsQueryException
- * if namespace format is unexpected.
- */
- private static List<String> normalize(String namespaceStr)
- throws InvalidAllocationTagsQueryException {
- List<String> result = new ArrayList<>();
- if (namespaceStr == null) {
- return result;
- }
-
- String[] nsValues = namespaceStr.split(NAMESPACE_DELIMITER);
- for (String str : nsValues) {
- if (!Strings.isNullOrEmpty(str)) {
- result.add(str);
- }
- }
-
- // Currently we only allow 1 or 2 values for a namespace string
- if (result.size() == 0 || result.size() > 2) {
- throw new InvalidAllocationTagsQueryException("Invalid namespace string: "
- + namespaceStr + ", the syntax is <namespace_prefix> or"
- + " <namespace_prefix>/<namespace_value>");
- }
-
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
index dc0237e..5cae92c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTags.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import java.util.Set;
@@ -29,22 +28,34 @@ import java.util.Set;
*/
public final class AllocationTags {
- private AllocationTagNamespace ns;
+ private TargetApplicationsNamespace ns;
private Set<String> tags;
+ private ApplicationId applicationId;
- private AllocationTags(AllocationTagNamespace namespace,
+ private AllocationTags(TargetApplicationsNamespace namespace,
Set<String> allocationTags) {
this.ns = namespace;
this.tags = allocationTags;
}
+ private AllocationTags(TargetApplicationsNamespace namespace,
+ Set<String> allocationTags, ApplicationId currentAppId) {
+ this.ns = namespace;
+ this.tags = allocationTags;
+ this.applicationId = currentAppId;
+ }
+
/**
* @return the namespace of these tags.
*/
- public AllocationTagNamespace getNamespace() {
+ public TargetApplicationsNamespace getNamespace() {
return this.ns;
}
+ public ApplicationId getCurrentApplicationId() {
+ return this.applicationId;
+ }
+
/**
* @return the allocation tags.
*/
@@ -55,28 +66,31 @@ public final class AllocationTags {
@VisibleForTesting
public static AllocationTags createSingleAppAllocationTags(
ApplicationId appId, Set<String> tags) {
- AllocationTagNamespace namespace = new AllocationTagNamespace.AppID(appId);
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppID(appId);
return new AllocationTags(namespace, tags);
}
@VisibleForTesting
public static AllocationTags createGlobalAllocationTags(Set<String> tags) {
- AllocationTagNamespace namespace = new AllocationTagNamespace.All();
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.All();
return new AllocationTags(namespace, tags);
}
@VisibleForTesting
public static AllocationTags createOtherAppAllocationTags(
- ApplicationId currentApp, Set<ApplicationId> allIds, Set<String> tags)
- throws InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace = new AllocationTagNamespace.NotSelf();
- TargetApplications ta = new TargetApplications(currentApp, allIds);
- namespace.evaluate(ta);
- return new AllocationTags(namespace, tags);
+ ApplicationId currentApp, Set<String> tags) {
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.NotSelf();
+ return new AllocationTags(namespace, tags, currentApp);
}
- public static AllocationTags newAllocationTags(
- AllocationTagNamespace namespace, Set<String> tags) {
- return new AllocationTags(namespace, tags);
+ public static AllocationTags createAllocationTags(
+ ApplicationId currentApplicationId, String namespaceString,
+ Set<String> tags) throws InvalidAllocationTagsQueryException {
+ TargetApplicationsNamespace namespace = TargetApplicationsNamespace
+ .parse(namespaceString);
+ return new AllocationTags(namespace, tags, currentApplicationId);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
index 830566a..4fc2bab 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/AllocationTagsManager.java
@@ -21,7 +21,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -32,12 +31,14 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.log4j.Logger;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.LongBinaryOperator;
@@ -292,13 +293,21 @@ public class AllocationTagsManager {
/**
* Aggregates multiple {@link TypeToCountedTags} to a single one based on
- * a given set of application IDs, the values are properly merged.
+ * the scope defined in the allocation tags, the values are properly merged.
*
- * @param appIds a set of application IDs.
+ * @param allocationTags {@link AllocationTags}.
* @return an aggregated {@link TypeToCountedTags}.
*/
- private TypeToCountedTags aggregateAllocationTags(Set<ApplicationId> appIds,
- Map<ApplicationId, TypeToCountedTags> mapping) {
+ private TypeToCountedTags aggregateAllocationTags(
+ AllocationTags allocationTags,
+ Map<ApplicationId, TypeToCountedTags> mapping)
+ throws InvalidAllocationTagsQueryException {
+ // Based on the namespace type of the given allocation tags
+ TargetApplicationsNamespace namespace = allocationTags.getNamespace();
+ TargetApplications ta = new TargetApplications(
+ allocationTags.getCurrentApplicationId(), getApplicationIdToTags());
+ namespace.evaluate(ta);
+ Set<ApplicationId> appIds = namespace.getNamespaceScope();
TypeToCountedTags result = new TypeToCountedTags();
if (appIds != null) {
if (appIds.size() == 1) {
@@ -571,9 +580,7 @@ public class AllocationTagsManager {
mapping = globalNodeMapping;
} else {
// Aggregate app tags cardinality by applications.
- mapping = aggregateAllocationTags(
- tags.getNamespace().getNamespaceScope(),
- perAppNodeMappings);
+ mapping = aggregateAllocationTags(tags, perAppNodeMappings);
}
return mapping == null ? 0 :
@@ -618,9 +625,7 @@ public class AllocationTagsManager {
mapping = globalRackMapping;
} else {
// Aggregates cardinality by rack.
- mapping = aggregateAllocationTags(
- tags.getNamespace().getNamespaceScope(),
- perAppRackMappings);
+ mapping = aggregateAllocationTags(tags, perAppRackMappings);
}
return mapping == null ? 0 :
@@ -642,10 +647,22 @@ public class AllocationTagsManager {
}
/**
- * @return all application IDs in a set that currently visible by
- * the allocation tags manager.
+ * @return all applications that is known to the
+ * {@link AllocationTagsManager}, along with their application tags.
+ * The result is a map, where key is an application ID, and value is the
+ * application-tags attached to this application. If there is no
+ * application-tag exists for the application, the value is an empty set.
*/
- public Set<ApplicationId> getAllApplicationIds() {
- return ImmutableSet.copyOf(perAppNodeMappings.keySet());
+ private Map<ApplicationId, Set<String>> getApplicationIdToTags() {
+ Map<ApplicationId, Set<String>> result = new HashMap<>();
+ ConcurrentMap<ApplicationId, RMApp> allApps = rmContext.getRMApps();
+ if (allApps != null) {
+ for (Map.Entry<ApplicationId, RMApp> app : allApps.entrySet()) {
+ if (perAppNodeMappings.containsKey(app.getKey())) {
+ result.put(app.getKey(), app.getValue().getApplicationTags());
+ }
+ }
+ }
+ return result;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
index 389fc5c..efa7b65 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/PlacementConstraintsUtil.java
@@ -24,7 +24,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.api.resource.PlacementConstraint;
@@ -58,35 +57,6 @@ public final class PlacementConstraintsUtil {
}
/**
- * Try to the namespace of the allocation tags from the given target key.
- *
- * @param targetKey
- * @return allocation tag namespace.
- * @throws InvalidAllocationTagsQueryException
- * if fail to parse the target key to a valid namespace.
- */
- private static AllocationTagNamespace getAllocationTagNamespace(
- ApplicationId currentAppId, String targetKey, AllocationTagsManager atm)
- throws InvalidAllocationTagsQueryException {
- // Parse to a valid namespace.
- AllocationTagNamespace namespace = AllocationTagNamespace.parse(targetKey);
-
- // TODO Complete remove this check once we support app-label.
- if (AllocationTagNamespaceType.APP_LABEL
- .equals(namespace.getNamespaceType())) {
- throw new InvalidAllocationTagsQueryException(
- namespace.toString() + " is not supported yet!");
- }
-
- // Evaluate the namespace according to the given target
- // before it can be consumed.
- TargetApplications ta =
- new TargetApplications(currentAppId, atm.getAllApplicationIds());
- namespace.evaluate(ta);
- return namespace;
- }
-
- /**
* Returns true if <b>single</b> placement constraint with associated
* allocationTags and scope is satisfied by a specific scheduler Node.
*
@@ -104,13 +74,10 @@ public final class PlacementConstraintsUtil {
ApplicationId targetApplicationId, SingleConstraint sc,
TargetExpression te, SchedulerNode node, AllocationTagsManager tm)
throws InvalidAllocationTagsQueryException {
- // Parse the allocation tag's namespace from the given target key,
- // then evaluate the namespace and get its scope,
- // which is represented by one or more application IDs.
- AllocationTagNamespace namespace = getAllocationTagNamespace(
- targetApplicationId, te.getTargetKey(), tm);
- AllocationTags allocationTags = AllocationTags
- .newAllocationTags(namespace, te.getTargetValues());
+ // Creates AllocationTags that will be further consumed by allocation
+ // tags manager for cardinality check.
+ AllocationTags allocationTags = AllocationTags.createAllocationTags(
+ targetApplicationId, te.getTargetKey(), te.getTargetValues());
long minScopeCardinality = 0;
long maxScopeCardinality = 0;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
index 0de7c9e..ab1bd9b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplications.java
@@ -18,34 +18,77 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.yarn.api.records.ApplicationId;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
/**
* This class is used by
- * {@link AllocationTagNamespace#evaluate(TargetApplications)} to evaluate
+ * {@link TargetApplicationsNamespace#evaluate(TargetApplications)} to evaluate
* a namespace.
*/
public class TargetApplications {
private ApplicationId currentAppId;
- private Set<ApplicationId> allAppIds;
+ private Map<ApplicationId, Set<String>> allApps;
public TargetApplications(ApplicationId currentApplicationId,
Set<ApplicationId> allApplicationIds) {
this.currentAppId = currentApplicationId;
- this.allAppIds = allApplicationIds;
+ allApps = new HashMap<>();
+ if (allApplicationIds != null) {
+ allApplicationIds.forEach(appId ->
+ allApps.put(appId, ImmutableSet.of()));
+ }
+ }
+
+ public TargetApplications(ApplicationId currentApplicationId,
+ Map<ApplicationId, Set<String>> allApplicationIds) {
+ this.currentAppId = currentApplicationId;
+ this.allApps = allApplicationIds;
}
public ApplicationId getCurrentApplicationId() {
return this.currentAppId;
}
+ public Set<ApplicationId> getAllApplicationIds() {
+ return this.allApps == null ?
+ ImmutableSet.of() : allApps.keySet();
+ }
+
public Set<ApplicationId> getOtherApplicationIds() {
- return allAppIds == null ? null : allAppIds.stream().filter(appId ->
- !appId.equals(getCurrentApplicationId()))
+ if (getAllApplicationIds() == null
+ || getAllApplicationIds().isEmpty()) {
+ return ImmutableSet.of();
+ }
+ return getAllApplicationIds()
+ .stream()
+ .filter(appId -> !appId.equals(getCurrentApplicationId()))
.collect(Collectors.toSet());
}
+
+ public Set<ApplicationId> getApplicationIdsByTag(String applicationTag) {
+ Set<ApplicationId> result = new HashSet<>();
+ if (Strings.isNullOrEmpty(applicationTag)
+ || this.allApps == null) {
+ return result;
+ }
+
+ for (Map.Entry<ApplicationId, Set<String>> app
+ : this.allApps.entrySet()) {
+ if (app.getValue() != null
+ && app.getValue().contains(applicationTag)) {
+ result.add(app.getKey());
+ }
+ }
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
new file mode 100644
index 0000000..a82a03f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TargetApplicationsNamespace.java
@@ -0,0 +1,326 @@
+/**
+ * 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.constraint;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
+import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.SELF;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.NOT_SELF;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_TAG;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.APP_ID;
+import static org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType.ALL;
+
+/**
+ * Class to describe the namespace of allocation tags, used by
+ * {@link AllocationTags}. Each namespace can be evaluated against
+ * a target set applications, represented by {@link TargetApplications}.
+ * After evaluation, the namespace is interpreted to be a set of
+ * applications based on the namespace type.
+ */
+public abstract class TargetApplicationsNamespace implements
+ Evaluable<TargetApplications> {
+
+ public final static String NAMESPACE_DELIMITER = "/";
+
+ private AllocationTagNamespaceType nsType;
+ // Namespace scope value will be delay binding by eval method.
+ private Set<ApplicationId> nsScope;
+
+ public TargetApplicationsNamespace(AllocationTagNamespaceType
+ allocationTagNamespaceType) {
+ this.nsType = allocationTagNamespaceType;
+ }
+
+ protected void setScopeIfNotNull(Set<ApplicationId> appIds) {
+ if (appIds != null) {
+ this.nsScope = appIds;
+ }
+ }
+
+ /**
+ * Get the type of the namespace.
+ * @return namespace type.
+ */
+ public AllocationTagNamespaceType getNamespaceType() {
+ return nsType;
+ }
+
+ /**
+ * Get the scope of the namespace, in form of a set of applications.
+ *
+ * @return a set of applications.
+ */
+ public Set<ApplicationId> getNamespaceScope() {
+ if (this.nsScope == null) {
+ throw new IllegalStateException("Invalid namespace scope,"
+ + " it is not initialized. Evaluate must be called before"
+ + " a namespace can be consumed.");
+ }
+ return this.nsScope;
+ }
+
+ /**
+ * Evaluate the namespace against given target applications
+ * if it is necessary. Only self/not-self/app-label namespace types
+ * require this evaluation step, because they are not binding to a
+ * specific scope during initiating. So we do lazy binding for them
+ * in this method.
+ *
+ * @param target a generic type target that impacts this evaluation.
+ * @throws InvalidAllocationTagsQueryException
+ */
+ @Override
+ public void evaluate(TargetApplications target)
+ throws InvalidAllocationTagsQueryException {
+ // Sub-class needs to override this when it requires the eval step.
+ }
+
+ @Override
+ public String toString() {
+ return this.nsType.toString();
+ }
+
+ /**
+ * Namespace within application itself.
+ */
+ public static class Self extends TargetApplicationsNamespace {
+
+ public Self() {
+ super(SELF);
+ }
+
+ @Override
+ public void evaluate(TargetApplications target)
+ throws InvalidAllocationTagsQueryException {
+ if (target == null || target.getCurrentApplicationId() == null) {
+ throw new InvalidAllocationTagsQueryException("Namespace Self must"
+ + " be evaluated against a single application ID.");
+ }
+ ApplicationId applicationId = target.getCurrentApplicationId();
+ setScopeIfNotNull(ImmutableSet.of(applicationId));
+ }
+ }
+
+ /**
+ * Namespace to all applications except itself.
+ */
+ public static class NotSelf extends TargetApplicationsNamespace {
+
+ private ApplicationId applicationId;
+
+ public NotSelf() {
+ super(NOT_SELF);
+ }
+
+ /**
+ * The scope of self namespace is to an application itself,
+ * the application ID can be delay binding to the namespace.
+ *
+ * @param appId application ID.
+ */
+ public void setApplicationId(ApplicationId appId) {
+ this.applicationId = appId;
+ }
+
+ public ApplicationId getApplicationId() {
+ return this.applicationId;
+ }
+
+ @Override
+ public void evaluate(TargetApplications target) {
+ Set<ApplicationId> otherAppIds = target.getOtherApplicationIds();
+ setScopeIfNotNull(otherAppIds);
+ }
+ }
+
+ /**
+ * Namespace to all applications in the cluster.
+ */
+ public static class All extends TargetApplicationsNamespace {
+
+ public All() {
+ super(ALL);
+ }
+ }
+
+ /**
+ * Namespace to applications that attached with a certain application tag.
+ */
+ public static class AppTag extends TargetApplicationsNamespace {
+
+ private String applicationTag;
+
+ public AppTag(String appTag) {
+ super(APP_TAG);
+ this.applicationTag = appTag;
+ }
+
+ @Override
+ public void evaluate(TargetApplications target) {
+ setScopeIfNotNull(target.getApplicationIdsByTag(applicationTag));
+ }
+
+ @Override
+ public String toString() {
+ return APP_TAG.toString() + NAMESPACE_DELIMITER + this.applicationTag;
+ }
+ }
+
+ /**
+ * Namespace defined by a certain application ID.
+ */
+ public static class AppID extends TargetApplicationsNamespace {
+
+ private ApplicationId targetAppId;
+ // app-id namespace requires an extra value of an application id.
+ public AppID(ApplicationId applicationId) {
+ super(APP_ID);
+ this.targetAppId = applicationId;
+ setScopeIfNotNull(ImmutableSet.of(targetAppId));
+ }
+
+ @Override
+ public String toString() {
+ return APP_ID.toString() + NAMESPACE_DELIMITER + this.targetAppId;
+ }
+ }
+
+ /**
+ * Parse namespace from a string. The string must be in legal format
+ * defined by each {@link AllocationTagNamespaceType}.
+ *
+ * @param namespaceStr namespace string.
+ * @return an instance of {@link TargetApplicationsNamespace}.
+ * @throws InvalidAllocationTagsQueryException
+ * if given string is not in valid format
+ */
+ public static TargetApplicationsNamespace parse(String namespaceStr)
+ throws InvalidAllocationTagsQueryException {
+ // Return the default namespace if no valid string is given.
+ if (Strings.isNullOrEmpty(namespaceStr)) {
+ return new Self();
+ }
+
+ // Normalize the input, escape additional chars.
+ List<String> nsValues = normalize(namespaceStr);
+ // The first string should be the prefix.
+ String nsPrefix = nsValues.get(0);
+ AllocationTagNamespaceType allocationTagNamespaceType =
+ fromString(nsPrefix);
+ switch (allocationTagNamespaceType) {
+ case SELF:
+ return new Self();
+ case NOT_SELF:
+ return new NotSelf();
+ case ALL:
+ return new All();
+ case APP_ID:
+ if (nsValues.size() != 2) {
+ throw new InvalidAllocationTagsQueryException(
+ "Missing the application ID in the namespace string: "
+ + namespaceStr);
+ }
+ String appIDStr = nsValues.get(1);
+ return parseAppID(appIDStr);
+ case APP_TAG:
+ if (nsValues.size() != 2) {
+ throw new InvalidAllocationTagsQueryException(
+ "Missing the application tag in the namespace string: "
+ + namespaceStr);
+ }
+ return new AppTag(nsValues.get(1));
+ default:
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid namespace string " + namespaceStr);
+ }
+ }
+
+ private static AllocationTagNamespaceType fromString(String prefix) throws
+ InvalidAllocationTagsQueryException {
+ for (AllocationTagNamespaceType type :
+ AllocationTagNamespaceType.values()) {
+ if(type.getTypeKeyword().equals(prefix)) {
+ return type;
+ }
+ }
+
+ Set<String> values = Arrays.stream(AllocationTagNamespaceType.values())
+ .map(AllocationTagNamespaceType::toString)
+ .collect(Collectors.toSet());
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid namespace prefix: " + prefix
+ + ", valid values are: " + String.join(",", values));
+ }
+
+ private static TargetApplicationsNamespace parseAppID(String appIDStr)
+ throws InvalidAllocationTagsQueryException {
+ try {
+ ApplicationId applicationId = ApplicationId.fromString(appIDStr);
+ return new AppID(applicationId);
+ } catch (IllegalArgumentException e) {
+ throw new InvalidAllocationTagsQueryException(
+ "Invalid application ID for "
+ + APP_ID.getTypeKeyword() + ": " + appIDStr);
+ }
+ }
+
+ /**
+ * Valid given namespace string and parse it to a list of sub-strings
+ * that can be consumed by the parser according to the type of the
+ * namespace. Currently the size of return list should be either 1 or 2.
+ * Extra slash is escaped during the normalization.
+ *
+ * @param namespaceStr namespace string.
+ * @return a list of parsed strings.
+ * @throws InvalidAllocationTagsQueryException
+ * if namespace format is unexpected.
+ */
+ private static List<String> normalize(String namespaceStr)
+ throws InvalidAllocationTagsQueryException {
+ List<String> result = new ArrayList<>();
+ if (namespaceStr == null) {
+ return result;
+ }
+
+ String[] nsValues = namespaceStr.split(NAMESPACE_DELIMITER);
+ for (String str : nsValues) {
+ if (!Strings.isNullOrEmpty(str)) {
+ result.add(str);
+ }
+ }
+
+ // Currently we only allow 1 or 2 values for a namespace string
+ if (result.size() == 0 || result.size() > 2) {
+ throw new InvalidAllocationTagsQueryException("Invalid namespace string: "
+ + namespaceStr + ", the syntax is <namespace_prefix> or"
+ + " <namespace_prefix>/<namespace_value>");
+ }
+
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
index 9004110..1fc6bad 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SingleConstraintAppPlacementAllocator.java
@@ -23,8 +23,6 @@ import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.AllocationTagNamespace;
-import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
@@ -334,25 +332,6 @@ public class SingleConstraintAppPlacementAllocator<N extends SchedulerNode>
targetAllocationTags = new HashSet<>(
targetExpression.getTargetValues());
-
- try {
- AllocationTagNamespace tagNS =
- AllocationTagNamespace.parse(targetExpression.getTargetKey());
- if (AllocationTagNamespaceType.APP_LABEL
- .equals(tagNS.getNamespaceType())) {
- throwExceptionWithMetaInfo(
- "As of now, allocation tag namespace ["
- + AllocationTagNamespaceType.APP_LABEL.toString()
- + "] is not supported. Please make changes to placement "
- + "constraints accordingly. If this is null, it will be "
- + "set to "
- + AllocationTagNamespaceType.SELF.toString()
- + " by default.");
- }
- } catch (InvalidAllocationTagsQueryException e) {
- throwExceptionWithMetaInfo(
- "Invalid allocation tag namespace, message: " + e.getMessage());
- }
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
index c399368..664fae2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
@@ -68,6 +68,7 @@ public class MockRMApp implements RMApp {
RMAppAttempt attempt;
int maxAppAttempts = 1;
List<ResourceRequest> amReqs;
+ private Set<String> applicationTags = null;
public MockRMApp(int newid, long time, RMAppState newState) {
finish = time;
@@ -82,6 +83,12 @@ public class MockRMApp implements RMApp {
user = userName;
}
+ public MockRMApp(int newid, long time, RMAppState newState,
+ String userName, Set<String> appTags) {
+ this(newid, time, newState, userName);
+ this.applicationTags = appTags;
+ }
+
public MockRMApp(int newid, long time, RMAppState newState, String userName, String diag) {
this(newid, time, newState, userName);
this.diagnostics = new StringBuilder(diag);
@@ -248,7 +255,7 @@ public class MockRMApp implements RMApp {
@Override
public Set<String> getApplicationTags() {
- return null;
+ return this.applicationTags;
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
index d7124bb..13247a7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestSchedulingRequestContainerAllocation.java
@@ -20,7 +20,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.AllocationTagNamespace;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.TargetApplicationsNamespace;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
@@ -294,7 +294,8 @@ public class TestSchedulingRequestContainerAllocation {
// App2 asks for 3 containers that anti-affinity with any mapper,
// since 3 out of 4 nodes already have mapper containers, all 3
// containers will be allocated on the other node.
- AllocationTagNamespace.All allNs = new AllocationTagNamespace.All();
+ TargetApplicationsNamespace.All allNs =
+ new TargetApplicationsNamespace.All();
am2.allocateAppAntiAffinity(
ResourceSizing.newInstance(3, Resource.newInstance(1024, 1)),
Priority.newInstance(1), 1L, allNs.toString(),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
index cbf5968..3f2aaed 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsManager.java
@@ -27,13 +27,19 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
/**
* Test functionality of AllocationTagsManager.
@@ -468,15 +474,27 @@ public class TestAllocationTagsManager {
@Test
public void testNodeAllocationTagsAggregation()
throws InvalidAllocationTagsQueryException {
+ RMContext mockContext = Mockito.spy(rmContext);
- AllocationTagsManager atm = new AllocationTagsManager(rmContext);
ApplicationId app1 = TestUtils.getMockApplicationId(1);
ApplicationId app2 = TestUtils.getMockApplicationId(2);
ApplicationId app3 = TestUtils.getMockApplicationId(3);
+
NodeId host1 = NodeId.fromString("host1:123");
NodeId host2 = NodeId.fromString("host2:123");
NodeId host3 = NodeId.fromString("host3:123");
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(app1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(app2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(app3, new MockRMApp(125, 1002,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ Mockito.when(mockContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager atm = new AllocationTagsManager(mockContext);
+
/**
* Node1 (rack0)
* app1/A(2)
@@ -561,7 +579,7 @@ public class TestAllocationTagsManager {
*
*/
tags = AllocationTags.createOtherAppAllocationTags(app1,
- ImmutableSet.of(app1, app2, app3), ImmutableSet.of("A", "B"));
+ ImmutableSet.of("A", "B"));
Assert.assertEquals(4, atm.getNodeCardinalityByOp(host1, tags, Long::max));
Assert.assertEquals(0, atm.getNodeCardinalityByOp(host1, tags, Long::min));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
index d1ef331..27a121a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestAllocationTagsNamespace.java
@@ -16,50 +16,67 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint; /**
* limitations under the License.
*/
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
import org.apache.hadoop.yarn.api.records.AllocationTagNamespaceType;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.junit.Assert;
import org.junit.Test;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
/**
- * Test class for {@link AllocationTagNamespace}.
+ * Test class for {@link TargetApplicationsNamespace}.
*/
public class TestAllocationTagsNamespace {
@Test
public void testNamespaceParse() throws InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace;
+ TargetApplicationsNamespace namespace;
String namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.SELF,
namespace.getNamespaceType());
namespaceStr = "not-self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.NOT_SELF,
namespace.getNamespaceType());
namespaceStr = "all";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.ALL,
namespace.getNamespaceType());
- namespaceStr = "app-label";
- namespace = AllocationTagNamespace.parse(namespaceStr);
- Assert.assertEquals(AllocationTagNamespaceType.APP_LABEL,
+ namespaceStr = "app-tag/spark-jobs";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ Assert.assertEquals(AllocationTagNamespaceType.APP_TAG,
namespace.getNamespaceType());
+ // Invalid app-tag namespace syntax
+ try {
+ namespaceStr = "app-tag/tag123/tag234";
+ TargetApplicationsNamespace.parse(namespaceStr);
+ Assert.fail("Parsing should fail as the given namespace is invalid");
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
+ Assert.assertTrue(e.getMessage().startsWith(
+ "Invalid namespace string"));
+ }
+
ApplicationId applicationId = ApplicationId.newInstance(12345, 1);
namespaceStr = "app-id/" + applicationId.toString();
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.APP_ID,
namespace.getNamespaceType());
// Invalid app-id namespace syntax, invalid app ID.
try {
namespaceStr = "app-id/apppppp_12345_99999";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the given app ID is invalid");
} catch (Exception e) {
Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
@@ -70,7 +87,7 @@ public class TestAllocationTagsNamespace {
// Invalid app-id namespace syntax, missing app ID.
try {
namespaceStr = "app-id";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the given namespace"
+ " is missing application ID");
} catch (Exception e) {
@@ -82,7 +99,7 @@ public class TestAllocationTagsNamespace {
// Invalid namespace type.
try {
namespaceStr = "non_exist_ns";
- AllocationTagNamespace.parse(namespaceStr);
+ TargetApplicationsNamespace.parse(namespaceStr);
Assert.fail("Parsing should fail as the giving type is not supported.");
} catch (Exception e) {
Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException);
@@ -94,7 +111,7 @@ public class TestAllocationTagsNamespace {
@Test
public void testNamespaceEvaluation() throws
InvalidAllocationTagsQueryException {
- AllocationTagNamespace namespace;
+ TargetApplicationsNamespace namespace;
TargetApplications targetApplications;
ApplicationId app1 = ApplicationId.newInstance(10000, 1);
ApplicationId app2 = ApplicationId.newInstance(10000, 2);
@@ -104,7 +121,7 @@ public class TestAllocationTagsNamespace {
// Ensure eval is called before using the scope.
String namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
try {
namespace.getNamespaceScope();
Assert.fail("Call getNamespaceScope before evaluate is not allowed.");
@@ -115,14 +132,14 @@ public class TestAllocationTagsNamespace {
}
namespaceStr = "self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1, ImmutableSet.of(app1));
namespace.evaluate(targetApplications);
Assert.assertEquals(1, namespace.getNamespaceScope().size());
Assert.assertEquals(app1, namespace.getNamespaceScope().iterator().next());
namespaceStr = "not-self";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1, ImmutableSet.of(app1));
namespace.evaluate(targetApplications);
Assert.assertEquals(0, namespace.getNamespaceScope().size());
@@ -134,16 +151,52 @@ public class TestAllocationTagsNamespace {
Assert.assertFalse(namespace.getNamespaceScope().contains(app1));
namespaceStr = "all";
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
Assert.assertEquals(AllocationTagNamespaceType.ALL,
namespace.getNamespaceType());
namespaceStr = "app-id/" + app2.toString();
- namespace = AllocationTagNamespace.parse(namespaceStr);
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
targetApplications = new TargetApplications(app1,
ImmutableSet.of(app1, app2, app3, app4, app5));
namespace.evaluate(targetApplications);
Assert.assertEquals(1, namespace.getNamespaceScope().size());
Assert.assertEquals(app2, namespace.getNamespaceScope().iterator().next());
+
+ /**
+ * App to Application Tags
+ * app1: A, B
+ * app2: A
+ * app3:
+ * app4: C
+ * app5: A, B, C
+ */
+ Map<ApplicationId, Set<String>> appsWithTags = new HashMap<>();
+ appsWithTags.put(app1, ImmutableSet.of("A", "B"));
+ appsWithTags.put(app2, ImmutableSet.of("A"));
+ appsWithTags.put(app3, ImmutableSet.of());
+ appsWithTags.put(app4, ImmutableSet.of("C"));
+ appsWithTags.put(app5, ImmutableSet.of("A", "B", "C"));
+
+ namespaceStr = "app-tag/A";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ targetApplications = new TargetApplications(app1, appsWithTags);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(3, namespace.getNamespaceScope().size());
+ Assert.assertTrue(Sets.difference(namespace.getNamespaceScope(),
+ ImmutableSet.of(app1, app2, app5)).isEmpty());
+
+ namespaceStr = "app-tag/B";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(2, namespace.getNamespaceScope().size());
+ Assert.assertTrue(Sets.difference(namespace.getNamespaceScope(),
+ ImmutableSet.of(app1, app5)).isEmpty());
+
+ // Not exist
+ namespaceStr = "app-tag/xyz";
+ namespace = TargetApplicationsNamespace.parse(namespaceStr);
+ namespace.evaluate(targetApplications);
+ Assert.assertEquals(0, namespace.getNamespaceScope().size());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa464c12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
index 4814321..3248450 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/constraint/TestPlacementConstraintsUtil.java
@@ -37,6 +37,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.concurrent.atomic.AtomicLong;
@@ -52,6 +54,9 @@ import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
@@ -63,6 +68,7 @@ import org.junit.Before;
import org.junit.Test;
import com.google.common.collect.ImmutableSet;
+import org.mockito.Mockito;
/**
* Test the PlacementConstraint Utility class functionality.
@@ -562,8 +568,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace namespaceAll =
- new AllocationTagNamespace.All();
+ TargetApplicationsNamespace namespaceAll =
+ new TargetApplicationsNamespace.All();
//***************************
// 1) all, anti-affinity
@@ -648,17 +654,28 @@ public class TestPlacementConstraintsUtil {
@Test
public void testNotSelfAppConstraints()
throws InvalidAllocationTagsQueryException {
- AllocationTagsManager tm = new AllocationTagsManager(rmContext);
- PlacementConstraintManagerService pcm =
- new MemoryPlacementConstraintManager();
- rmContext.setAllocationTagsManager(tm);
- rmContext.setPlacementConstraintManager(pcm);
-
long ts = System.currentTimeMillis();
ApplicationId application1 = BuilderUtils.newApplicationId(ts, 100);
ApplicationId application2 = BuilderUtils.newApplicationId(ts, 101);
ApplicationId application3 = BuilderUtils.newApplicationId(ts, 102);
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(application1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(application2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ allApps.put(application3, new MockRMApp(125, 1002,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+
+ RMContext mockedContext = Mockito.spy(rmContext);
+ when(mockedContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager tm = new AllocationTagsManager(mockedContext);
+ PlacementConstraintManagerService pcm =
+ new MemoryPlacementConstraintManager();
+ mockedContext.setAllocationTagsManager(tm);
+ mockedContext.setPlacementConstraintManager(pcm);
+
// Register App1 with anti-affinity constraint map.
RMNode n0r1 = rmNodes.get(0);
RMNode n1r1 = rmNodes.get(1);
@@ -696,8 +713,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace notSelf =
- new AllocationTagNamespace.NotSelf();
+ TargetApplicationsNamespace notSelf =
+ new TargetApplicationsNamespace.NotSelf();
//***************************
// 1) not-self, app1
@@ -800,8 +817,8 @@ public class TestPlacementConstraintsUtil {
SchedulerNode schedulerNode3 =newSchedulerNode(n3r2.getHostName(),
n3r2.getRackName(), n3r2.getNodeID());
- AllocationTagNamespace namespace =
- new AllocationTagNamespace.AppID(application1);
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppID(application1);
Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>();
PlacementConstraint constraint2 = PlacementConstraints
.targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(),
@@ -832,7 +849,7 @@ public class TestPlacementConstraintsUtil {
// Intra-app constraint
// Test with default and empty namespace
- AllocationTagNamespace self = new AllocationTagNamespace.Self();
+ TargetApplicationsNamespace self = new TargetApplicationsNamespace.Self();
PlacementConstraint constraint3 = PlacementConstraints
.targetNotIn(NODE, allocationTagWithNamespace(self.toString(),
"hbase-m"))
@@ -873,6 +890,88 @@ public class TestPlacementConstraintsUtil {
}
@Test
+ public void testInterAppConstriantsByAppTag()
+ throws InvalidAllocationTagsQueryException {
+
+ ApplicationId application1 = BuilderUtils.newApplicationId(1000, 123);
+ ApplicationId application2 = BuilderUtils.newApplicationId(1001, 124);
+
+ // app1: test-tag
+ // app2: N/A
+ RMContext mockedContext = Mockito.spy(rmContext);
+ ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>();
+ allApps.put(application1, new MockRMApp(123, 1000,
+ RMAppState.NEW, "userA", ImmutableSet.of("test-tag")));
+ allApps.put(application2, new MockRMApp(124, 1001,
+ RMAppState.NEW, "userA", ImmutableSet.of("")));
+ when(mockedContext.getRMApps()).thenReturn(allApps);
+
+ AllocationTagsManager tm = new AllocationTagsManager(mockedContext);
+ PlacementConstraintManagerService pcm =
+ new MemoryPlacementConstraintManager();
+ mockedContext.setAllocationTagsManager(tm);
+ mockedContext.setPlacementConstraintManager(pcm);
+
+ // Register App1 with anti-affinity constraint map.
+ RMNode n0r1 = rmNodes.get(0);
+ RMNode n1r1 = rmNodes.get(1);
+ RMNode n2r2 = rmNodes.get(2);
+ RMNode n3r2 = rmNodes.get(3);
+
+ /**
+ * Place container:
+ * n0: app1/hbase-m(1)
+ * n1: ""
+ * n2: app1/hbase-m(1)
+ * n3: ""
+ */
+ tm.addContainer(n0r1.getNodeID(),
+ newContainerId(application1), ImmutableSet.of("hbase-m"));
+ tm.addContainer(n2r2.getNodeID(),
+ newContainerId(application1), ImmutableSet.of("hbase-m"));
+
+ SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(),
+ n0r1.getRackName(), n0r1.getNodeID());
+ SchedulerNode schedulerNode1 = newSchedulerNode(n1r1.getHostName(),
+ n1r1.getRackName(), n1r1.getNodeID());
+ SchedulerNode schedulerNode2 = newSchedulerNode(n2r2.getHostName(),
+ n2r2.getRackName(), n2r2.getNodeID());
+ SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(),
+ n3r2.getRackName(), n3r2.getNodeID());
+
+ TargetApplicationsNamespace namespace =
+ new TargetApplicationsNamespace.AppTag("test-tag");
+ Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>();
+ PlacementConstraint constraint2 = PlacementConstraints
+ .targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(),
+ "hbase-m"))
+ .build();
+ Set<String> srcTags2 = ImmutableSet.of("app2");
+ constraintMap.put(srcTags2, constraint2);
+
+ pcm.registerApplication(application2, constraintMap);
+
+ // Anti-affinity with app-tag/test-tag/hbase-m,
+ // app1 has tag "test-tag" so the constraint is equally to work on app1
+ // onto n1 and n3 as they don't have "hbase-m" from app1.
+ Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode0, pcm, tm));
+ Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode1, pcm, tm));
+ Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode2, pcm, tm));
+ Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(
+ application2, createSchedulingRequest(srcTags2),
+ schedulerNode3, pcm, tm));
+
+ pcm.unregisterApplication(application1);
+ pcm.unregisterApplication(application2);
+ }
+
+ @Test
public void testInvalidAllocationTagNamespace() {
AllocationTagsManager tm = new AllocationTagsManager(rmContext);
PlacementConstraintManagerService pcm =
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org