先看完整的capacity-scheduler.xml配置
<configuration>
<property>
<name>yarn.scheduler.capacity.resource-calculator</name>
<value>org.apache.hadoop.yarn.util.resource.DominantResourceCalculator</value>
</property>
<property>
<name>yarn.scheduler.capacity.global-queue-max-application</name>
<value>10000</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.queues</name>
<value>legend,default</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.acl_administer_queue</name>
<value>legend_super_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.acl_submit_applications</name>
<value>legend_super_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.default.capacity</name>
<value>100</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.default.accessible-node-labels</name>
<value> </value>
</property>
<property>
<name>yarn.scheduler.capacity.root.default.default-node-label-expression</name>
<value> </value>
</property>
<property>
<name>yarn.scheduler.capacity.root.accessible-node-labels</name>
<value>*</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.accessible-node-labels.bdspLabel.capacity</name>
<value>100</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.accessible-node-labels</name>
<value>bdspLabel</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.default-node-label-expression</name>
<value>bdspLabel</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.accessible-node-labels.bdspLabel.capacity</name>
<value>100</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.accessible-node-labels.bdspLabel.maximum-capacity</name>
<value>100</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.acl_submit_applications</name>
<value>legend_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.acl_administer_queue</name>
<value>legend_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.queue-mappings</name>
<value>u:legend_user:legend</value>
</property>
<property>
<name>yarn.scheduler.capacity.legend.state</name>
<value>RUNNING</value>
</property>
<property>
<name>yarn.scheduler.capacity.legend.maximum-am-resource-percent</name>
<value>0.1</value>
</property>
<property>
<name>yarn.scheduler.capacity.legend.maximum-applications</name>
<value>100000</value>
</property>
</configuration>
该配置添加了NodeLabel限制，特别注意参数：yarn.scheduler.capacity.global-queue-max-application，如该参数不设置，必须设置能够访问默认NodeLabel的百分比，如设置yarn.scheduler.capacity.root.legend.capacity，这是因为提交队列时，LeafQueue类的validateSubmitApplication方法会Check submission limits for queues
public void validateSubmitApplication(ApplicationId applicationId,
String userName, String queue) throws AccessControlException {
try {
writeLock.lock();
// Check if the queue is accepting jobs
if (getState() != QueueState.RUNNING) {
String msg = "Queue " + getQueuePath()
+ " is STOPPED. Cannot accept submission of application: "
+ applicationId;
LOG.info(msg);
throw new AccessControlException(msg);
}
// Check submission limits for queues
if (getNumApplications() >= getMaxApplications()) {
String msg =
"Queue " + getQueuePath() + " already has " + getNumApplications()
+ " applications,"
+ " cannot accept submission of application: " + applicationId
+ " getMaxApplications() is: " + getMaxApplications();
LOG.info(msg);
throw new AccessControlException(msg);
}
// Check submission limits for the user on this queue
User user = usersManager.getUserAndAddIfAbsent(userName);
if (user.getTotalApplications() >= getMaxApplicationsPerUser()) {
String msg = "Queue " + getQueuePath() + " already has " + user
.getTotalApplications() + " applications from user " + userName
+ " cannot accept submission of application: " + applicationId;
LOG.info(msg);
throw new AccessControlException(msg);
}
} finally {
writeLock.unlock();
}
try {
getParent().validateSubmitApplication(applicationId, userName, queue);
} catch (AccessControlException ace) {
LOG.info("Failed to submit application to parent-queue: " +
getParent().getQueuePath(), ace);
throw ace;
}
}
注意说明
限制用户提交队列权限属性是yarn.scheduler.capacity.root.legend.acl_submit_applications
管理队列权限数据是yarn.scheduler.capacity.root.legend.acl_administer_queue
<property>
<name>yarn.scheduler.capacity.root.legend.acl_submit_applications</name>
<value>legend_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.legend.acl_administer_queue</name>
<value>legend_user</value>
</property>
YARN的权限是先校验该队列用户是否有相关权限，如没有则查找该队列的父队列是否拥有权限，一层一层往上查找，所以需配置root队列用户权限，默认是*，不限制权限
<property>
<name>yarn.scheduler.capacity.root.acl_administer_queue</name>
<value>legend_super_user</value>
</property>
<property>
<name>yarn.scheduler.capacity.root.acl_submit_applications</name>
<value>legend_super_user</value>
</property>
笔者在按照以上配置后，踩了一个坑，发现权限控制不生效，所有用户都可以提交权限，通过分析查看源码流程，调用ClientRMService类的submitApplication方法进行提交任务到队列，部分代码
try {
// call RMAppManager to submit application directly
rmAppManager.submitApplication(submissionContext,
System.currentTimeMillis(), user);
LOG.info("Application with id " + applicationId.getId() +
" submitted by user " + user);
RMAuditLogger.logSuccess(user, AuditConstants.SUBMIT_APP_REQUEST,
"ClientRMService", applicationId, callerContext);
} catch (YarnException e) {
LOG.info("Exception in submitting " + applicationId, e);
RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST,
e.getMessage(), "ClientRMService",
"Exception in submitting application", applicationId, callerContext);
throw e;
}
接着调用mAppManager.submitApplication方法进行提交，即RMAppManager类的submitApplication方法
@SuppressWarnings("unchecked")
protected void submitApplication(
ApplicationSubmissionContext submissionContext, long submitTime,
String user) throws YarnException {
LOG.info("===RMAppManager submitApplication====");
ApplicationId applicationId = submissionContext.getApplicationId();
// Passing start time as -1. It will be eventually set in RMAppImpl
// constructor.
RMAppImpl application = createAndPopulateNewRMApp(
submissionContext, submitTime, user, false, -1);
LOG.info("===RMAppManager submitApplication====");
try {
if (UserGroupInformation.isSecurityEnabled()) {
this.rmContext.getDelegationTokenRenewer()
.addApplicationAsync(applicationId,
BuilderUtils.parseCredentials(submissionContext),
submissionContext.getCancelTokensWhenComplete(),
application.getUser(),
BuilderUtils.parseTokensConf(submissionContext));
} else {
// Dispatcher is not yet started at this time, so these START events
// enqueued should be guaranteed to be first processed when dispatcher
// gets started.
this.rmContext.getDispatcher().getEventHandler()
.handle(new RMAppEvent(applicationId, RMAppEventType.START));
}
} catch (Exception e) {
LOG.warn("Unable to parse credentials for " + applicationId, e);
// Sending APP_REJECTED is fine, since we assume that the
// RMApp is in NEW state and thus we haven't yet informed the
// scheduler about the existence of the application
this.rmContext.getDispatcher().getEventHandler()
.handle(new RMAppEvent(applicationId,
RMAppEventType.APP_REJECTED, e.getMessage()));
throw RPCUtil.getRemoteException(e);
}
}
会在createAndPopulateNewRMApp方法中进行队列相关权限的检查，部分代码
// Since FairScheduler queue mapping is done inside scheduler,
// if FairScheduler is used and the queue doesn't exist, we should not
// fail here because queue will be created inside FS. Ideally, FS queue
// mapping should be done outside scheduler too like CS.
// For now, exclude FS for the acl check.
LOG.info("===RMAppManager createAndPopulateNewRMApp====");
LOG.info("--!isRecovery--" + !isRecovery);
LOG.info("--YarnConfiguration.isAclEnabled(conf)--" + YarnConfiguration.isAclEnabled(conf));
if (!isRecovery && YarnConfiguration.isAclEnabled(conf)
&& scheduler instanceof CapacityScheduler) {
String queueName = submissionContext.getQueue();
String appName = submissionContext.getApplicationName();
CSQueue csqueue = ((CapacityScheduler) scheduler).getQueue(queueName);
if (csqueue == null && placementContext != null) {
//could be an auto created queue through queue mapping. Validate
// parent queue exists and has valid acls
String parentQueueName = placementContext.getParentQueue();
csqueue = ((CapacityScheduler) scheduler).getQueue(parentQueueName);
}
LOG.info("=====RMAppImpl createAndPopulateNewRMApp===");
LOG.info(csqueue != null);
if (csqueue != null) {
LOG.info(csqueue.getQueueName());
}
LOG.info("=====RMAppImpl createAndPopulateNewRMApp===");
if (csqueue != null
&& !authorizer.checkPermission(
new AccessRequest(csqueue.getPrivilegedEntity(), userUgi,
SchedulerUtils.toAccessType(QueueACL.SUBMIT_APPLICATIONS),
applicationId.toString(), appName, Server.getRemoteAddress(),
null))
&& !authorizer.checkPermission(
new AccessRequest(csqueue.getPrivilegedEntity(), userUgi,
SchedulerUtils.toAccessType(QueueACL.ADMINISTER_QUEUE),
applicationId.toString(), appName, Server.getRemoteAddress(),
null))) {
throw RPCUtil.getRemoteException(new AccessControlException(
"User " + user + " does not have permission to submit "
+ applicationId + " to queue " + submissionContext.getQueue()));
}
}
其中会进行是否开启了Yarn的权限参数配置校验：YarnConfiguration.isAclEnabled(conf)，默认为false不进行权限的检查，如要开启Yarn权限校验，需配置yarn-site.xml中设置yarn.acl.enable参数为true
网友评论