YARN-9865. Capacity scheduler: add support for combined %user + %secondary_group mapping. Contributed by Manikandan R
This commit is contained in:
parent
516377bfa6
commit
30b93f914b
@ -157,6 +157,21 @@ public class UserGroupMappingPlacementRule extends PlacementRule {
|
|||||||
this.groups = groups;
|
this.groups = groups;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private String getSecondaryGroup(String user) throws IOException {
|
||||||
|
List<String> groupsList = groups.getGroups(user);
|
||||||
|
String secondaryGroup = null;
|
||||||
|
// Traverse all secondary groups (as there could be more than one
|
||||||
|
// and position is not guaranteed) and ensure there is queue with
|
||||||
|
// the same name
|
||||||
|
for (int i = 1; i < groupsList.size(); i++) {
|
||||||
|
if (this.queueManager.getQueue(groupsList.get(i)) != null) {
|
||||||
|
secondaryGroup = groupsList.get(i);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return secondaryGroup;
|
||||||
|
}
|
||||||
|
|
||||||
private ApplicationPlacementContext getPlacementForUser(String user)
|
private ApplicationPlacementContext getPlacementForUser(String user)
|
||||||
throws IOException {
|
throws IOException {
|
||||||
for (QueueMapping mapping : mappings) {
|
for (QueueMapping mapping : mappings) {
|
||||||
@ -169,22 +184,27 @@ public class UserGroupMappingPlacementRule extends PlacementRule {
|
|||||||
new QueueMapping(mapping.getType(), mapping.getSource(),
|
new QueueMapping(mapping.getType(), mapping.getSource(),
|
||||||
CURRENT_USER_MAPPING, groups.getGroups(user).get(0)),
|
CURRENT_USER_MAPPING, groups.getGroups(user).get(0)),
|
||||||
user);
|
user);
|
||||||
|
} else if (mapping.getParentQueue() != null
|
||||||
|
&& mapping.getParentQueue().equals(SECONDARY_GROUP_MAPPING)
|
||||||
|
&& mapping.getQueue().equals(CURRENT_USER_MAPPING)) {
|
||||||
|
String secondaryGroup = getSecondaryGroup(user);
|
||||||
|
if (secondaryGroup != null) {
|
||||||
|
return getPlacementContext(new QueueMapping(mapping.getType(),
|
||||||
|
mapping.getSource(), CURRENT_USER_MAPPING, secondaryGroup),
|
||||||
|
user);
|
||||||
|
} else {
|
||||||
|
if (LOG.isDebugEnabled()) {
|
||||||
|
LOG.debug("User {} is not associated with any Secondary Group. "
|
||||||
|
+ "Hence it may use the 'default' queue", user);
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
} else if (mapping.queue.equals(CURRENT_USER_MAPPING)) {
|
} else if (mapping.queue.equals(CURRENT_USER_MAPPING)) {
|
||||||
return getPlacementContext(mapping, user);
|
return getPlacementContext(mapping, user);
|
||||||
} else if (mapping.queue.equals(PRIMARY_GROUP_MAPPING)) {
|
} else if (mapping.queue.equals(PRIMARY_GROUP_MAPPING)) {
|
||||||
return getPlacementContext(mapping, groups.getGroups(user).get(0));
|
return getPlacementContext(mapping, groups.getGroups(user).get(0));
|
||||||
} else if (mapping.queue.equals(SECONDARY_GROUP_MAPPING)) {
|
} else if (mapping.queue.equals(SECONDARY_GROUP_MAPPING)) {
|
||||||
List<String> groupsList = groups.getGroups(user);
|
String secondaryGroup = getSecondaryGroup(user);
|
||||||
String secondaryGroup = null;
|
|
||||||
// Traverse all secondary groups (as there could be more than one
|
|
||||||
// and position is not guaranteed) and ensure there is queue with
|
|
||||||
// the same name
|
|
||||||
for (int i = 1; i < groupsList.size(); i++) {
|
|
||||||
if (this.queueManager.getQueue(groupsList.get(i)) != null) {
|
|
||||||
secondaryGroup = groupsList.get(i);
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (secondaryGroup != null) {
|
if (secondaryGroup != null) {
|
||||||
return getPlacementContext(mapping, secondaryGroup);
|
return getPlacementContext(mapping, secondaryGroup);
|
||||||
} else {
|
} else {
|
||||||
@ -383,7 +403,8 @@ public class UserGroupMappingPlacementRule extends PlacementRule {
|
|||||||
CapacitySchedulerQueueManager queueManager, QueueMapping mapping,
|
CapacitySchedulerQueueManager queueManager, QueueMapping mapping,
|
||||||
QueuePath queuePath) throws IOException {
|
QueuePath queuePath) throws IOException {
|
||||||
if (queuePath.hasParentQueue()
|
if (queuePath.hasParentQueue()
|
||||||
&& queuePath.getParentQueue().equals(PRIMARY_GROUP_MAPPING)) {
|
&& (queuePath.getParentQueue().equals(PRIMARY_GROUP_MAPPING)
|
||||||
|
|| queuePath.getParentQueue().equals(SECONDARY_GROUP_MAPPING))) {
|
||||||
// dynamic parent queue
|
// dynamic parent queue
|
||||||
return new QueueMapping(mapping.getType(), mapping.getSource(),
|
return new QueueMapping(mapping.getType(), mapping.getSource(),
|
||||||
queuePath.getLeafQueue(), queuePath.getParentQueue());
|
queuePath.getLeafQueue(), queuePath.getParentQueue());
|
||||||
|
@ -114,6 +114,10 @@ public class TestUserGroupMappingPlacementRule {
|
|||||||
verifyQueueMapping(
|
verifyQueueMapping(
|
||||||
new QueueMapping(MappingType.USER, "%user", "%user", "%primary_group"),
|
new QueueMapping(MappingType.USER, "%user", "%user", "%primary_group"),
|
||||||
"a", YarnConfiguration.DEFAULT_QUEUE_NAME, "a", false, "agroup");
|
"a", YarnConfiguration.DEFAULT_QUEUE_NAME, "a", false, "agroup");
|
||||||
|
verifyQueueMapping(
|
||||||
|
new QueueMapping(MappingType.USER, "%user", "%user",
|
||||||
|
"%secondary_group"),
|
||||||
|
"a", YarnConfiguration.DEFAULT_QUEUE_NAME, "a", false, "asubgroup2");
|
||||||
verifyQueueMapping(new QueueMapping(MappingType.GROUP, "asubgroup1", "q1"),
|
verifyQueueMapping(new QueueMapping(MappingType.GROUP, "asubgroup1", "q1"),
|
||||||
"a", "q1");
|
"a", "q1");
|
||||||
|
|
||||||
|
@ -329,7 +329,7 @@ public class TestCapacitySchedulerAutoCreatedQueueBase {
|
|||||||
// Define top-level queues
|
// Define top-level queues
|
||||||
// Set childQueue for root
|
// Set childQueue for root
|
||||||
conf.setQueues(ROOT,
|
conf.setQueues(ROOT,
|
||||||
new String[] { "a", "b", "c", "d" });
|
new String[] { "a", "b", "c", "d", "asubgroup1", "asubgroup2" });
|
||||||
|
|
||||||
conf.setCapacity(A, A_CAPACITY);
|
conf.setCapacity(A, A_CAPACITY);
|
||||||
conf.setCapacity(B, B_CAPACITY);
|
conf.setCapacity(B, B_CAPACITY);
|
||||||
|
@ -132,61 +132,6 @@ public class TestCapacitySchedulerQueueMappingFactory {
|
|||||||
assertThat(placementRuleNames, hasItems(QUEUE_MAPPING_RULE_APP_NAME));
|
assertThat(placementRuleNames, hasItems(QUEUE_MAPPING_RULE_APP_NAME));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
|
||||||
public void testNestedUserQueueWithDynamicParentQueue() throws Exception {
|
|
||||||
|
|
||||||
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
|
|
||||||
setupQueueConfiguration(conf);
|
|
||||||
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
|
|
||||||
ResourceScheduler.class);
|
|
||||||
conf.setClass(CommonConfigurationKeys.HADOOP_SECURITY_GROUP_MAPPING,
|
|
||||||
SimpleGroupsMapping.class, GroupMappingServiceProvider.class);
|
|
||||||
|
|
||||||
List<String> queuePlacementRules = new ArrayList<>();
|
|
||||||
queuePlacementRules.add(QUEUE_MAPPING_RULE_USER_GROUP);
|
|
||||||
conf.setQueuePlacementRules(queuePlacementRules);
|
|
||||||
|
|
||||||
List<UserGroupMappingPlacementRule.QueueMapping> existingMappingsForUG =
|
|
||||||
conf.getQueueMappings();
|
|
||||||
|
|
||||||
// set queue mapping
|
|
||||||
List<UserGroupMappingPlacementRule.QueueMapping> queueMappingsForUG =
|
|
||||||
new ArrayList<>();
|
|
||||||
|
|
||||||
// u:%user:%primary_group.%user
|
|
||||||
UserGroupMappingPlacementRule.QueueMapping userQueueMapping =
|
|
||||||
new UserGroupMappingPlacementRule.QueueMapping(
|
|
||||||
UserGroupMappingPlacementRule.QueueMapping.MappingType.USER,
|
|
||||||
"%user", getQueueMapping("%primary_group", "%user"));
|
|
||||||
queueMappingsForUG.add(userQueueMapping);
|
|
||||||
|
|
||||||
existingMappingsForUG.addAll(queueMappingsForUG);
|
|
||||||
conf.setQueueMappings(existingMappingsForUG);
|
|
||||||
|
|
||||||
// override with queue mappings
|
|
||||||
conf.setOverrideWithQueueMappings(true);
|
|
||||||
|
|
||||||
mockRM = new MockRM(conf);
|
|
||||||
CapacityScheduler cs = (CapacityScheduler) mockRM.getResourceScheduler();
|
|
||||||
cs.updatePlacementRules();
|
|
||||||
mockRM.start();
|
|
||||||
cs.start();
|
|
||||||
|
|
||||||
ApplicationSubmissionContext asc =
|
|
||||||
Records.newRecord(ApplicationSubmissionContext.class);
|
|
||||||
asc.setQueue("default");
|
|
||||||
String inputUser = "a";
|
|
||||||
|
|
||||||
List<PlacementRule> rules =
|
|
||||||
cs.getRMContext().getQueuePlacementManager().getPlacementRules();
|
|
||||||
|
|
||||||
UserGroupMappingPlacementRule r =
|
|
||||||
(UserGroupMappingPlacementRule) rules.get(0);
|
|
||||||
ApplicationPlacementContext ctx = r.getPlacementForApp(asc, inputUser);
|
|
||||||
assertEquals("Queue", "a", ctx.getQueue());
|
|
||||||
assertEquals("Group", "agroup", ctx.getParentQueue());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void testNestedUserQueueWithStaticParentQueue() throws Exception {
|
public void testNestedUserQueueWithStaticParentQueue() throws Exception {
|
||||||
|
|
||||||
@ -250,4 +195,120 @@ public class TestCapacitySchedulerQueueMappingFactory {
|
|||||||
assertEquals("Queue", "user2", ctx2.getQueue());
|
assertEquals("Queue", "user2", ctx2.getQueue());
|
||||||
assertEquals("Queue", "c", ctx2.getParentQueue());
|
assertEquals("Queue", "c", ctx2.getParentQueue());
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
@Test
|
||||||
|
public void testNestedUserQueueWithPrimaryGroupAsDynamicParentQueue()
|
||||||
|
throws Exception {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Mapping order: 1. u:%user:%primary_group.%user 2.
|
||||||
|
* u:%user:%secondary_group.%user
|
||||||
|
*
|
||||||
|
* Expected parent queue is primary group of the user
|
||||||
|
*/
|
||||||
|
|
||||||
|
// set queue mapping
|
||||||
|
List<UserGroupMappingPlacementRule.QueueMapping> queueMappingsForUG =
|
||||||
|
new ArrayList<>();
|
||||||
|
|
||||||
|
// u:%user:%primary_group.%user
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping userQueueMapping1 =
|
||||||
|
new UserGroupMappingPlacementRule.QueueMapping(
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping.MappingType.USER,
|
||||||
|
"%user", getQueueMapping("%primary_group", "%user"));
|
||||||
|
|
||||||
|
// u:%user:%secondary_group.%user
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping userQueueMapping2 =
|
||||||
|
new UserGroupMappingPlacementRule.QueueMapping(
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping.MappingType.USER,
|
||||||
|
"%user", getQueueMapping("%secondary_group", "%user"));
|
||||||
|
|
||||||
|
queueMappingsForUG.add(userQueueMapping1);
|
||||||
|
queueMappingsForUG.add(userQueueMapping2);
|
||||||
|
|
||||||
|
testNestedUserQueueWithDynamicParentQueue(queueMappingsForUG, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testNestedUserQueueWithSecondaryGroupAsDynamicParentQueue()
|
||||||
|
throws Exception {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Mapping order: 1. u:%user:%secondary_group.%user 2.
|
||||||
|
* u:%user:%primary_group.%user
|
||||||
|
*
|
||||||
|
* Expected parent queue is secondary group of the user
|
||||||
|
*/
|
||||||
|
|
||||||
|
// set queue mapping
|
||||||
|
List<UserGroupMappingPlacementRule.QueueMapping> queueMappingsForUG =
|
||||||
|
new ArrayList<>();
|
||||||
|
|
||||||
|
// u:%user:%primary_group.%user
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping userQueueMapping1 =
|
||||||
|
new UserGroupMappingPlacementRule.QueueMapping(
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping.MappingType.USER,
|
||||||
|
"%user", getQueueMapping("%primary_group", "%user"));
|
||||||
|
|
||||||
|
// u:%user:%secondary_group.%user
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping userQueueMapping2 =
|
||||||
|
new UserGroupMappingPlacementRule.QueueMapping(
|
||||||
|
UserGroupMappingPlacementRule.QueueMapping.MappingType.USER,
|
||||||
|
"%user", getQueueMapping("%secondary_group", "%user"));
|
||||||
|
|
||||||
|
queueMappingsForUG.add(userQueueMapping2);
|
||||||
|
queueMappingsForUG.add(userQueueMapping1);
|
||||||
|
|
||||||
|
testNestedUserQueueWithDynamicParentQueue(queueMappingsForUG, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void testNestedUserQueueWithDynamicParentQueue(
|
||||||
|
List<UserGroupMappingPlacementRule.QueueMapping> mapping,
|
||||||
|
boolean primary)
|
||||||
|
throws Exception {
|
||||||
|
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
|
||||||
|
setupQueueConfiguration(conf);
|
||||||
|
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
|
||||||
|
ResourceScheduler.class);
|
||||||
|
conf.setClass(CommonConfigurationKeys.HADOOP_SECURITY_GROUP_MAPPING,
|
||||||
|
SimpleGroupsMapping.class, GroupMappingServiceProvider.class);
|
||||||
|
|
||||||
|
List<String> queuePlacementRules = new ArrayList<>();
|
||||||
|
queuePlacementRules.add(QUEUE_MAPPING_RULE_USER_GROUP);
|
||||||
|
conf.setQueuePlacementRules(queuePlacementRules);
|
||||||
|
|
||||||
|
List<UserGroupMappingPlacementRule.QueueMapping> existingMappingsForUG =
|
||||||
|
conf.getQueueMappings();
|
||||||
|
|
||||||
|
existingMappingsForUG.addAll(mapping);
|
||||||
|
conf.setQueueMappings(existingMappingsForUG);
|
||||||
|
|
||||||
|
// override with queue mappings
|
||||||
|
conf.setOverrideWithQueueMappings(true);
|
||||||
|
|
||||||
|
mockRM = new MockRM(conf);
|
||||||
|
CapacityScheduler cs = (CapacityScheduler) mockRM.getResourceScheduler();
|
||||||
|
cs.updatePlacementRules();
|
||||||
|
mockRM.start();
|
||||||
|
cs.start();
|
||||||
|
|
||||||
|
ApplicationSubmissionContext asc =
|
||||||
|
Records.newRecord(ApplicationSubmissionContext.class);
|
||||||
|
asc.setQueue("default");
|
||||||
|
|
||||||
|
List<PlacementRule> rules =
|
||||||
|
cs.getRMContext().getQueuePlacementManager().getPlacementRules();
|
||||||
|
|
||||||
|
UserGroupMappingPlacementRule r =
|
||||||
|
(UserGroupMappingPlacementRule) rules.get(0);
|
||||||
|
ApplicationPlacementContext ctx = r.getPlacementForApp(asc, "a");
|
||||||
|
assertEquals("Queue", "a", ctx.getQueue());
|
||||||
|
|
||||||
|
if (primary) {
|
||||||
|
assertEquals("Primary Group", "agroup", ctx.getParentQueue());
|
||||||
|
} else {
|
||||||
|
assertEquals("Secondary Group", "asubgroup1", ctx.getParentQueue());
|
||||||
|
}
|
||||||
|
mockRM.close();
|
||||||
|
}
|
||||||
|
}
|
@ -170,13 +170,15 @@ Example:
|
|||||||
```
|
```
|
||||||
<property>
|
<property>
|
||||||
<name>yarn.scheduler.capacity.queue-mappings</name>
|
<name>yarn.scheduler.capacity.queue-mappings</name>
|
||||||
<value>u:user1:queue1,g:group1:queue2,u:%user:%user,u:user2:%primary_group,u:user3:%secondary_group,u:%user:%primary_group.%user</value>
|
<value>u:user1:queue1,g:group1:queue2,u:%user:%user,u:user2:%primary_group,u:user3:%secondary_group,u:%user:%primary_group.%user,u:%user:%secondary_group.%user</value>
|
||||||
<description>
|
<description>
|
||||||
Here, <user1> is mapped to <queue1>, <group1> is mapped to <queue2>,
|
Here, <user1> is mapped to <queue1>, <group1> is mapped to <queue2>,
|
||||||
maps users to queues with the same name as user, <user2> is mapped
|
maps users to queues with the same name as user, <user2> is mapped
|
||||||
to queue name same as <primary group>, maps users to queue with the
|
to queue name same as <primary group>, maps users to queue with the
|
||||||
same name as user but parent queue name should be same as <primary group>
|
same name as user but parent queue name should be same as <primary group>
|
||||||
of the user respectively. The mappings will be evaluated from left to
|
of the user, maps users to queue with the same name as user but parent
|
||||||
|
queue name should be same as any <secondary group> of the user
|
||||||
|
respectively. The mappings will be evaluated from left to
|
||||||
right, and the first valid mapping will be used.
|
right, and the first valid mapping will be used.
|
||||||
</description>
|
</description>
|
||||||
</property>
|
</property>
|
||||||
|
Loading…
x
Reference in New Issue
Block a user