YARN-8361. Change App Name Placement Rule to use App Name instead of App Id for configuration. (Zian Chen via wangda)
Change-Id: I17e5021f8f611a9c5e3bd4b38f25e08585afc6b1
This commit is contained in:
Wangda Tan
parent
752dcce5f4
commit
a2e49f41a8
|
|
@ -66,7 +66,7 @@ public boolean initialize(CapacitySchedulerContext schedulerContext)
|
|||
CapacitySchedulerConfiguration conf = schedulerContext.getConfiguration();
|
||||
boolean overrideWithQueueMappings = conf.getOverrideWithQueueMappings();
|
||||
LOG.info(
|
||||
"Initialized queue mappings, override: " + overrideWithQueueMappings);
|
||||
"Initialized App Name queue mappings, override: " + overrideWithQueueMappings);
|
||||
|
||||
List<QueueMappingEntity> queueMappings =
|
||||
conf.getQueueMappingEntity(QUEUE_MAPPING_NAME);
|
||||
|
|
@ -139,6 +139,8 @@ public boolean initialize(CapacitySchedulerContext schedulerContext)
|
|||
if (newMappings.size() > 0) {
|
||||
this.mappings = newMappings;
|
||||
this.overrideWithQueueMappings = overrideWithQueueMappings;
|
||||
LOG.info("get valid queue mapping from app name config: " +
|
||||
newMappings.toString() + ", override: " + overrideWithQueueMappings);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
|
|
@ -149,16 +151,16 @@ private static boolean ifQueueDoesNotExist(CSQueue queue) {
|
|||
}
|
||||
|
||||
private ApplicationPlacementContext getAppPlacementContext(String user,
|
||||
ApplicationId applicationId) throws IOException {
|
||||
String applicationName) throws IOException {
|
||||
for (QueueMappingEntity mapping : mappings) {
|
||||
if (mapping.getSource().equals(CURRENT_APP_MAPPING)) {
|
||||
if (mapping.getQueue().equals(CURRENT_APP_MAPPING)) {
|
||||
return getPlacementContext(mapping, String.valueOf(applicationId));
|
||||
return getPlacementContext(mapping, applicationName);
|
||||
} else {
|
||||
return getPlacementContext(mapping);
|
||||
}
|
||||
}
|
||||
if (mapping.getSource().equals(applicationId.toString())) {
|
||||
if (mapping.getSource().equals(applicationName)) {
|
||||
return getPlacementContext(mapping);
|
||||
}
|
||||
}
|
||||
|
|
@ -169,25 +171,25 @@ private ApplicationPlacementContext getAppPlacementContext(String user,
|
|||
public ApplicationPlacementContext getPlacementForApp(
|
||||
ApplicationSubmissionContext asc, String user) throws YarnException {
|
||||
String queueName = asc.getQueue();
|
||||
ApplicationId applicationId = asc.getApplicationId();
|
||||
String applicationName = asc.getApplicationName();
|
||||
if (mappings != null && mappings.size() > 0) {
|
||||
try {
|
||||
ApplicationPlacementContext mappedQueue = getAppPlacementContext(user,
|
||||
applicationId);
|
||||
applicationName);
|
||||
if (mappedQueue != null) {
|
||||
// We have a mapping, should we use it?
|
||||
if (queueName.equals(YarnConfiguration.DEFAULT_QUEUE_NAME)
|
||||
//queueName will be same as mapped queue name in case of recovery
|
||||
|| queueName.equals(mappedQueue.getQueue())
|
||||
|| overrideWithQueueMappings) {
|
||||
LOG.info("Application " + applicationId
|
||||
LOG.info("Application " + applicationName
|
||||
+ " mapping [" + queueName + "] to [" + mappedQueue
|
||||
+ "] override " + overrideWithQueueMappings);
|
||||
return mappedQueue;
|
||||
}
|
||||
}
|
||||
} catch (IOException ioex) {
|
||||
String message = "Failed to submit application " + applicationId +
|
||||
String message = "Failed to submit application " + applicationName +
|
||||
" reason: " + ioex.getMessage();
|
||||
throw new YarnException(message);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@
|
|||
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
|
||||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SimpleGroupsMapping;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
import org.junit.Assert;
|
||||
|
|
@ -33,13 +34,7 @@
|
|||
import java.util.Arrays;
|
||||
|
||||
public class TestAppNameMappingPlacementRule {
|
||||
|
||||
private static final long CLUSTER_TIMESTAMP = System.currentTimeMillis();
|
||||
public static final String APPIDSTRPREFIX = "application";
|
||||
private static final String APPLICATION_ID_PREFIX = APPIDSTRPREFIX + '_';
|
||||
private static final String APPLICATION_ID_SUFFIX = '_' + "0001";
|
||||
private static final String CLUSTER_APP_ID = APPLICATION_ID_PREFIX +
|
||||
CLUSTER_TIMESTAMP + APPLICATION_ID_SUFFIX;
|
||||
private static final String APP_NAME = "DistributedShell";
|
||||
|
||||
private YarnConfiguration conf = new YarnConfiguration();
|
||||
|
||||
|
|
@ -50,24 +45,29 @@ public void setup() {
|
|||
}
|
||||
|
||||
private void verifyQueueMapping(QueueMappingEntity queueMapping,
|
||||
String inputAppId, String expectedQueue) throws YarnException {
|
||||
verifyQueueMapping(queueMapping, inputAppId,
|
||||
YarnConfiguration.DEFAULT_QUEUE_NAME, expectedQueue, false);
|
||||
String user, String expectedQueue) throws YarnException {
|
||||
verifyQueueMapping(queueMapping, user,
|
||||
queueMapping.getQueue(), expectedQueue, false);
|
||||
}
|
||||
|
||||
private void verifyQueueMapping(QueueMappingEntity queueMapping,
|
||||
String inputAppId, String inputQueue, String expectedQueue,
|
||||
String user, String inputQueue, String expectedQueue,
|
||||
boolean overwrite) throws YarnException {
|
||||
AppNameMappingPlacementRule rule = new AppNameMappingPlacementRule(
|
||||
overwrite, Arrays.asList(queueMapping));
|
||||
ApplicationSubmissionContext asc = Records.newRecord(
|
||||
ApplicationSubmissionContext.class);
|
||||
if (inputQueue.equals("%application")) {
|
||||
inputQueue = APP_NAME;
|
||||
}
|
||||
asc.setQueue(inputQueue);
|
||||
ApplicationId appId = ApplicationId.newInstance(CLUSTER_TIMESTAMP,
|
||||
Integer.parseInt(inputAppId));
|
||||
asc.setApplicationId(appId);
|
||||
String appName = queueMapping.getSource();
|
||||
if (appName.equals("%application")) {
|
||||
appName = APP_NAME;
|
||||
}
|
||||
asc.setApplicationName(appName);
|
||||
ApplicationPlacementContext ctx = rule.getPlacementForApp(asc,
|
||||
queueMapping.getSource());
|
||||
user);
|
||||
Assert.assertEquals(expectedQueue,
|
||||
ctx != null ? ctx.getQueue() : inputQueue);
|
||||
}
|
||||
|
|
@ -75,19 +75,20 @@ private void verifyQueueMapping(QueueMappingEntity queueMapping,
|
|||
@Test
|
||||
public void testMapping() throws YarnException {
|
||||
// simple base case for mapping user to queue
|
||||
verifyQueueMapping(new QueueMappingEntity(CLUSTER_APP_ID,
|
||||
"q1"), "1", "q1");
|
||||
verifyQueueMapping(new QueueMappingEntity("%application", "q2"), "1", "q2");
|
||||
verifyQueueMapping(new QueueMappingEntity(APP_NAME,
|
||||
"q1"), "user_1", "q1");
|
||||
verifyQueueMapping(new QueueMappingEntity("%application", "q2"), "user_1",
|
||||
"q2");
|
||||
verifyQueueMapping(new QueueMappingEntity("%application", "%application"),
|
||||
"1", CLUSTER_APP_ID);
|
||||
"user_1", APP_NAME);
|
||||
|
||||
// specify overwritten, and see if user specified a queue, and it will be
|
||||
// overridden
|
||||
verifyQueueMapping(new QueueMappingEntity(CLUSTER_APP_ID,
|
||||
verifyQueueMapping(new QueueMappingEntity(APP_NAME,
|
||||
"q1"), "1", "q2", "q1", true);
|
||||
|
||||
// if overwritten not specified, it should be which user specified
|
||||
verifyQueueMapping(new QueueMappingEntity(CLUSTER_APP_ID,
|
||||
verifyQueueMapping(new QueueMappingEntity(APP_NAME,
|
||||
"q1"), "1", "q2", "q2", false);
|
||||
}
|
||||
}
|
||||
|
|
@ -39,6 +39,7 @@
|
|||
public class TestPlacementManager {
|
||||
|
||||
public static final String USER = "user_";
|
||||
public static final String APP_NAME = "DistributedShell";
|
||||
public static final String APP_ID1 = "1";
|
||||
public static final String USER1 = USER + APP_ID1;
|
||||
public static final String APP_ID2 = "2";
|
||||
|
|
@ -82,9 +83,7 @@ public void testPlaceApplicationWithPlacementRuleChain() throws Exception {
|
|||
|
||||
ApplicationSubmissionContext asc = Records.newRecord(
|
||||
ApplicationSubmissionContext.class);
|
||||
ApplicationId appId = ApplicationId.newInstance(CLUSTER_TIMESTAMP,
|
||||
Integer.parseInt(APP_ID1));
|
||||
asc.setApplicationId(appId);
|
||||
asc.setApplicationName(APP_NAME);
|
||||
|
||||
boolean caughtException = false;
|
||||
try{
|
||||
|
|
@ -94,7 +93,7 @@ public void testPlaceApplicationWithPlacementRuleChain() throws Exception {
|
|||
}
|
||||
Assert.assertTrue(caughtException);
|
||||
|
||||
QueueMappingEntity queueMappingEntity = new QueueMappingEntity(APP_ID1,
|
||||
QueueMappingEntity queueMappingEntity = new QueueMappingEntity(APP_NAME,
|
||||
USER1, PARENT_QUEUE);
|
||||
|
||||
AppNameMappingPlacementRule anRule = new AppNameMappingPlacementRule(false,
|
||||
|
|
|
|||
|
|
@ -162,7 +162,7 @@ Configuration
|
|||
| Property | Description |
|
||||
|:---- |:---- |
|
||||
| `yarn.scheduler.capacity.queue-mappings` | This configuration specifies the mapping of user or group to a specific queue. You can map a single user or a list of users to queues. Syntax: `[u or g]:[name]:[queue_name][,next_mapping]*`. Here, *u or g* indicates whether the mapping is for a user or group. The value is *u* for user and *g* for group. *name* indicates the user name or group name. To specify the user who has submitted the application, %user can be used. *queue_name* indicates the queue name for which the application has to be mapped. To specify queue name same as user name, *%user* can be used. To specify queue name same as the name of the primary group for which the user belongs to, *%primary_group* can be used.|
|
||||
| `yarn.scheduler.queue-placement-rules.app-name` | This configuration specifies the mapping of application_id to a specific queue. You can map a single application or a list of applications to queues. Syntax: `[app_id]:[queue_name][,next_mapping]*`. Here, *app_id* indicates the application id you want to do the mapping. To specify the current application's id as the app_id, %application can be used. *queue_name* indicates the queue name for which the application has to be mapped. To specify queue name same as application id, *%application* can be used.|
|
||||
| `yarn.scheduler.queue-placement-rules.app-name` | This configuration specifies the mapping of application_name to a specific queue. You can map a single application or a list of applications to queues. Syntax: `[app_name]:[queue_name][,next_mapping]*`. Here, *app_name* indicates the application name you want to do the mapping. *queue_name* indicates the queue name for which the application has to be mapped. To specify the current application's name as the app_name, %application can be used.|
|
||||
| `yarn.scheduler.capacity.queue-mappings-override.enable` | This function is used to specify whether the user specified queues can be overridden. This is a Boolean value and the default value is *false*. |
|
||||
|
||||
Example:
|
||||
|
|
@ -181,9 +181,9 @@ Example:
|
|||
|
||||
<property>
|
||||
<name>yarn.scheduler.queue-placement-rules.app-name</name>
|
||||
<value>appId1:queue1,%application:%application</value>
|
||||
<value>appName1:queue1,%application:%application</value>
|
||||
<description>
|
||||
Here, <appId1> is mapped to <queue1>, maps applications to queues with
|
||||
Here, <appName1> is mapped to <queue1>, maps applications to queues with
|
||||
the same name as application respectively. The mappings will be
|
||||
evaluated from left to right, and the first valid mapping will be used.
|
||||
</description>
|
||||
|
|
|
|||
Loading…
Reference in New Issue