You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by at...@apache.org on 2011/11/02 06:35:03 UTC
svn commit: r1196458 [15/19] - in
/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project: ./ assembly/
bin/ conf/ dev-support/ hadoop-mapreduce-client/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/
hadoop-mapreduce-client/hadoop-mapreduce-cl...
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java Wed Nov 2 05:34:31 2011
@@ -21,16 +21,17 @@ package org.apache.hadoop.yarn.server.re
import java.io.IOException;
import java.net.InetSocketAddress;
-import org.apache.avro.ipc.Server;
+import org.apache.hadoop.ipc.Server;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.Groups;
-import org.apache.hadoop.security.SecurityInfo;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.security.authorize.ProxyUsers;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
@@ -38,7 +39,7 @@ import org.apache.hadoop.yarn.factories.
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.ipc.YarnRPC;
-import org.apache.hadoop.yarn.security.SchedulerSecurityInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
import org.apache.hadoop.yarn.server.resourcemanager.api.RMAdminProtocol;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshAdminAclsRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshAdminAclsResponse;
@@ -46,12 +47,14 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshNodesResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesResponse;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshUserToGroupsMappingsRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshUserToGroupsMappingsResponse;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
+import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
import org.apache.hadoop.yarn.service.AbstractService;
public class AdminService extends AbstractService implements RMAdminProtocol {
@@ -62,7 +65,11 @@ public class AdminService extends Abstra
private final ResourceScheduler scheduler;
private final RMContext rmContext;
private final NodesListManager nodesListManager;
-
+
+ private final ClientRMService clientRMService;
+ private final ApplicationMasterService applicationMasterService;
+ private final ResourceTrackerService resourceTrackerService;
+
private Server server;
private InetSocketAddress masterServiceAddress;
private AccessControlList adminAcl;
@@ -71,12 +78,18 @@ public class AdminService extends Abstra
RecordFactoryProvider.getRecordFactory(null);
public AdminService(Configuration conf, ResourceScheduler scheduler,
- RMContext rmContext, NodesListManager nodesListManager) {
+ RMContext rmContext, NodesListManager nodesListManager,
+ ClientRMService clientRMService,
+ ApplicationMasterService applicationMasterService,
+ ResourceTrackerService resourceTrackerService) {
super(AdminService.class.getName());
this.conf = conf;
this.scheduler = scheduler;
this.rmContext = rmContext;
this.nodesListManager = nodesListManager;
+ this.clientRMService = clientRMService;
+ this.applicationMasterService = applicationMasterService;
+ this.resourceTrackerService = resourceTrackerService;
}
@Override
@@ -85,10 +98,12 @@ public class AdminService extends Abstra
String bindAddress =
conf.get(YarnConfiguration.RM_ADMIN_ADDRESS,
YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS);
- masterServiceAddress = NetUtils.createSocketAddr(bindAddress);
- adminAcl =
- new AccessControlList(
- conf.get(YarnConfiguration.RM_ADMIN_ACL, YarnConfiguration.DEFAULT_RM_ADMIN_ACL));
+ masterServiceAddress = NetUtils.createSocketAddr(bindAddress,
+ YarnConfiguration.DEFAULT_RM_ADMIN_PORT,
+ YarnConfiguration.RM_ADMIN_ADDRESS);
+ adminAcl = new AccessControlList(conf.get(
+ YarnConfiguration.YARN_ADMIN_ACL,
+ YarnConfiguration.DEFAULT_YARN_ADMIN_ACL));
}
public void start() {
@@ -99,6 +114,14 @@ public class AdminService extends Abstra
conf, null,
conf.getInt(YarnConfiguration.RM_ADMIN_CLIENT_THREAD_COUNT,
YarnConfiguration.DEFAULT_RM_ADMIN_CLIENT_THREAD_COUNT));
+
+ // Enable service authorization?
+ if (conf.getBoolean(
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
+ false)) {
+ refreshServiceAcls(conf, new RMPolicyProvider());
+ }
+
this.server.start();
super.start();
}
@@ -106,7 +129,7 @@ public class AdminService extends Abstra
@Override
public void stop() {
if (this.server != null) {
- this.server.close();
+ this.server.stop();
}
super.stop();
}
@@ -214,12 +237,41 @@ public class AdminService extends Abstra
UserGroupInformation user = checkAcls("refreshAdminAcls");
Configuration conf = new Configuration();
- adminAcl =
- new AccessControlList(
- conf.get(YarnConfiguration.RM_ADMIN_ACL, YarnConfiguration.DEFAULT_RM_ADMIN_ACL));
+ adminAcl = new AccessControlList(conf.get(
+ YarnConfiguration.YARN_ADMIN_ACL,
+ YarnConfiguration.DEFAULT_YARN_ADMIN_ACL));
RMAuditLogger.logSuccess(user.getShortUserName(), "refreshAdminAcls",
"AdminService");
return recordFactory.newRecordInstance(RefreshAdminAclsResponse.class);
}
+
+ @Override
+ public RefreshServiceAclsResponse refreshServiceAcls(
+ RefreshServiceAclsRequest request) throws YarnRemoteException {
+ Configuration conf = new Configuration();
+ if (!conf.getBoolean(
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
+ false)) {
+ throw RPCUtil.getRemoteException(
+ new IOException("Service Authorization (" +
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION +
+ ") not enabled."));
+ }
+
+ PolicyProvider policyProvider = new RMPolicyProvider();
+
+ refreshServiceAcls(conf, policyProvider);
+ clientRMService.refreshServiceAcls(conf, policyProvider);
+ applicationMasterService.refreshServiceAcls(conf, policyProvider);
+ resourceTrackerService.refreshServiceAcls(conf, policyProvider);
+
+ return recordFactory.newRecordInstance(RefreshServiceAclsResponse.class);
+ }
+
+ void refreshServiceAcls(Configuration configuration,
+ PolicyProvider policyProvider) {
+ this.server.refreshServiceAcl(configuration, policyProvider);
+ }
+
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java Wed Nov 2 05:34:31 2011
@@ -18,18 +18,22 @@
package org.apache.hadoop.yarn.server.resourcemanager;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import org.apache.avro.ipc.Server;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.SecurityInfo;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.PolicyProvider;
+import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.AMRMProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
@@ -38,8 +42,8 @@ import org.apache.hadoop.yarn.api.protoc
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
import org.apache.hadoop.yarn.api.records.AMResponse;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -49,7 +53,6 @@ import org.apache.hadoop.yarn.factory.pr
import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.security.ApplicationTokenSecretManager;
-import org.apache.hadoop.yarn.security.SchedulerSecurityInfo;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
@@ -59,6 +62,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptUnregistrationEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
import org.apache.hadoop.yarn.service.AbstractService;
@Private
@@ -92,7 +96,9 @@ public class ApplicationMasterService ex
String bindAddress =
conf.get(YarnConfiguration.RM_SCHEDULER_ADDRESS,
YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS);
- masterServiceAddress = NetUtils.createSocketAddr(bindAddress);
+ masterServiceAddress = NetUtils.createSocketAddr(bindAddress,
+ YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT,
+ YarnConfiguration.RM_SCHEDULER_ADDRESS);
super.init(conf);
}
@@ -105,16 +111,55 @@ public class ApplicationMasterService ex
conf, this.appTokenManager,
conf.getInt(YarnConfiguration.RM_SCHEDULER_CLIENT_THREAD_COUNT,
YarnConfiguration.DEFAULT_RM_SCHEDULER_CLIENT_THREAD_COUNT));
+
+ // Enable service authorization?
+ if (conf.getBoolean(
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
+ false)) {
+ refreshServiceAcls(conf, new RMPolicyProvider());
+ }
+
this.server.start();
super.start();
}
+ private void authorizeRequest(ApplicationAttemptId appAttemptID)
+ throws YarnRemoteException {
+
+ if (!UserGroupInformation.isSecurityEnabled()) {
+ return;
+ }
+
+ String appAttemptIDStr = appAttemptID.toString();
+
+ UserGroupInformation remoteUgi;
+ try {
+ remoteUgi = UserGroupInformation.getCurrentUser();
+ } catch (IOException e) {
+ String msg = "Cannot obtain the user-name for ApplicationAttemptID: "
+ + appAttemptIDStr + ". Got exception: "
+ + StringUtils.stringifyException(e);
+ LOG.warn(msg);
+ throw RPCUtil.getRemoteException(msg);
+ }
+
+ if (!remoteUgi.getUserName().equals(appAttemptIDStr)) {
+ String msg = "Unauthorized request from ApplicationMaster. "
+ + "Expected ApplicationAttemptID: " + remoteUgi.getUserName()
+ + " Found: " + appAttemptIDStr;
+ LOG.warn(msg);
+ throw RPCUtil.getRemoteException(msg);
+ }
+ }
+
@Override
public RegisterApplicationMasterResponse registerApplicationMaster(
RegisterApplicationMasterRequest request) throws YarnRemoteException {
ApplicationAttemptId applicationAttemptId = request
.getApplicationAttemptId();
+ authorizeRequest(applicationAttemptId);
+
ApplicationId appID = applicationAttemptId.getApplicationId();
AMResponse lastResponse = responseMap.get(applicationAttemptId);
if (lastResponse == null) {
@@ -159,6 +204,8 @@ public class ApplicationMasterService ex
ApplicationAttemptId applicationAttemptId = request
.getApplicationAttemptId();
+ authorizeRequest(applicationAttemptId);
+
AMResponse lastResponse = responseMap.get(applicationAttemptId);
if (lastResponse == null) {
String message = "Application doesn't exist in cache "
@@ -188,6 +235,7 @@ public class ApplicationMasterService ex
throws YarnRemoteException {
ApplicationAttemptId appAttemptId = request.getApplicationAttemptId();
+ authorizeRequest(appAttemptId);
this.amLivelinessMonitor.receivedPing(appAttemptId);
@@ -244,6 +292,7 @@ public class ApplicationMasterService ex
public void registerAppAttempt(ApplicationAttemptId attemptId) {
AMResponse response = recordFactory.newRecordInstance(AMResponse.class);
response.setResponseId(0);
+ LOG.info("Registering " + attemptId);
responseMap.put(attemptId, response);
}
@@ -256,10 +305,15 @@ public class ApplicationMasterService ex
}
}
+ public void refreshServiceAcls(Configuration configuration,
+ PolicyProvider policyProvider) {
+ this.server.refreshServiceAcl(configuration, policyProvider);
+ }
+
@Override
public void stop() {
if (this.server != null) {
- this.server.close();
+ this.server.stop();
}
super.stop();
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java Wed Nov 2 05:34:31 2011
@@ -24,20 +24,17 @@ import java.security.AccessControlExcept
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.avro.ipc.Server;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.SecurityInfo;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.yarn.api.ClientRMProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
@@ -52,13 +49,17 @@ import org.apache.hadoop.yarn.api.protoc
import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse;
import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.QueueInfo;
+import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
@@ -66,13 +67,16 @@ import org.apache.hadoop.yarn.factories.
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.ipc.YarnRPC;
-import org.apache.hadoop.yarn.security.client.ClientRMSecurityInfo;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
import org.apache.hadoop.yarn.service.AbstractService;
@@ -96,15 +100,15 @@ public class ClientRMService extends Abs
private final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
InetSocketAddress clientBindAddress;
- private ApplicationACLsManager aclsManager;
- private Map<ApplicationACL, AccessControlList> applicationACLs;
+ private final ApplicationACLsManager applicationsACLsManager;
public ClientRMService(RMContext rmContext, YarnScheduler scheduler,
- RMAppManager rmAppManager) {
+ RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager) {
super(ClientRMService.class.getName());
this.scheduler = scheduler;
this.rmContext = rmContext;
this.rmAppManager = rmAppManager;
+ this.applicationsACLsManager = applicationACLsManager;
}
@Override
@@ -113,11 +117,9 @@ public class ClientRMService extends Abs
conf.get(YarnConfiguration.RM_ADDRESS,
YarnConfiguration.DEFAULT_RM_ADDRESS);
clientBindAddress =
- NetUtils.createSocketAddr(clientServiceBindAddress);
-
- this.aclsManager = new ApplicationACLsManager(conf);
- this.applicationACLs = aclsManager.constructApplicationACLs(conf);
-
+ NetUtils.createSocketAddr(clientServiceBindAddress,
+ YarnConfiguration.DEFAULT_RM_PORT,
+ YarnConfiguration.RM_ADDRESS);
super.init(conf);
}
@@ -133,27 +135,33 @@ public class ClientRMService extends Abs
conf, null,
conf.getInt(YarnConfiguration.RM_CLIENT_THREAD_COUNT,
YarnConfiguration.DEFAULT_RM_CLIENT_THREAD_COUNT));
+
+ // Enable service authorization?
+ if (conf.getBoolean(
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
+ false)) {
+ refreshServiceAcls(conf, new RMPolicyProvider());
+ }
+
this.server.start();
super.start();
}
/**
* check if the calling user has the access to application information.
- * @param appAttemptId
* @param callerUGI
* @param owner
- * @param appACL
+ * @param operationPerformed
+ * @param applicationId
* @return
*/
- private boolean checkAccess(UserGroupInformation callerUGI, String owner, ApplicationACL appACL) {
- if (!UserGroupInformation.isSecurityEnabled()) {
- return true;
- }
- AccessControlList applicationACL = applicationACLs.get(appACL);
- return aclsManager.checkAccess(callerUGI, appACL, owner, applicationACL);
+ private boolean checkAccess(UserGroupInformation callerUGI, String owner,
+ ApplicationAccessType operationPerformed, ApplicationId applicationId) {
+ return applicationsACLsManager.checkAccess(callerUGI, operationPerformed,
+ owner, applicationId);
}
- public ApplicationId getNewApplicationId() {
+ ApplicationId getNewApplicationId() {
ApplicationId applicationId = org.apache.hadoop.yarn.util.BuilderUtils
.newApplicationId(recordFactory, ResourceManager.clusterTimeStamp,
applicationCounter.incrementAndGet());
@@ -180,9 +188,29 @@ public class ClientRMService extends Abs
public GetApplicationReportResponse getApplicationReport(
GetApplicationReportRequest request) throws YarnRemoteException {
ApplicationId applicationId = request.getApplicationId();
- RMApp application = rmContext.getRMApps().get(applicationId);
- ApplicationReport report = (application == null) ? null : application
- .createAndGetApplicationReport();
+
+ UserGroupInformation callerUGI;
+ try {
+ callerUGI = UserGroupInformation.getCurrentUser();
+ } catch (IOException ie) {
+ LOG.info("Error getting UGI ", ie);
+ throw RPCUtil.getRemoteException(ie);
+ }
+
+ RMApp application = this.rmContext.getRMApps().get(applicationId);
+ if (application == null) {
+ throw RPCUtil.getRemoteException("Trying to get information for an "
+ + "absent application " + applicationId);
+ }
+
+ if (!checkAccess(callerUGI, application.getUser(),
+ ApplicationAccessType.VIEW_APP, applicationId)) {
+ throw RPCUtil.getRemoteException(new AccessControlException("User "
+ + callerUGI.getShortUserName() + " cannot perform operation "
+ + ApplicationAccessType.VIEW_APP.name() + " on " + applicationId));
+ }
+
+ ApplicationReport report = application.createAndGetApplicationReport();
GetApplicationReportResponse response = recordFactory
.newRecordInstance(GetApplicationReportResponse.class);
@@ -203,14 +231,15 @@ public class ClientRMService extends Abs
throw new IOException("Application with id " + applicationId
+ " is already present! Cannot add a duplicate!");
}
-
+
// Safety
submissionContext.setUser(user);
-
+
// This needs to be synchronous as the client can query
// immediately following the submission to get the application status.
// So call handle directly and do not send an event.
- rmAppManager.handle(new RMAppManagerSubmitEvent(submissionContext));
+ rmAppManager.handle(new RMAppManagerSubmitEvent(submissionContext, System
+ .currentTimeMillis()));
LOG.info("Application with id " + applicationId.getId() +
" submitted by user " + user + " with " + submissionContext);
@@ -248,16 +277,25 @@ public class ClientRMService extends Abs
}
RMApp application = this.rmContext.getRMApps().get(applicationId);
- // TODO: What if null
+ if (application == null) {
+ RMAuditLogger.logFailure(callerUGI.getUserName(),
+ AuditConstants.KILL_APP_REQUEST, "UNKNOWN", "ClientRMService",
+ "Trying to kill an absent application", applicationId);
+ throw RPCUtil
+ .getRemoteException("Trying to kill an absent application "
+ + applicationId);
+ }
+
if (!checkAccess(callerUGI, application.getUser(),
- ApplicationACL.MODIFY_APP)) {
- RMAuditLogger.logFailure(callerUGI.getShortUserName(),
- AuditConstants.KILL_APP_REQUEST,
- "User doesn't have MODIFY_APP permissions", "ClientRMService",
+ ApplicationAccessType.MODIFY_APP, applicationId)) {
+ RMAuditLogger.logFailure(callerUGI.getShortUserName(),
+ AuditConstants.KILL_APP_REQUEST,
+ "User doesn't have permissions to "
+ + ApplicationAccessType.MODIFY_APP.toString(), "ClientRMService",
AuditConstants.UNAUTHORIZED_USER, applicationId);
throw RPCUtil.getRemoteException(new AccessControlException("User "
+ callerUGI.getShortUserName() + " cannot perform operation "
- + ApplicationACL.MODIFY_APP.name() + " on " + applicationId));
+ + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId));
}
this.rmContext.getDispatcher().getEventHandler().handle(
@@ -286,9 +324,24 @@ public class ClientRMService extends Abs
public GetAllApplicationsResponse getAllApplications(
GetAllApplicationsRequest request) throws YarnRemoteException {
+ UserGroupInformation callerUGI;
+ try {
+ callerUGI = UserGroupInformation.getCurrentUser();
+ } catch (IOException ie) {
+ LOG.info("Error getting UGI ", ie);
+ throw RPCUtil.getRemoteException(ie);
+ }
+
List<ApplicationReport> reports = new ArrayList<ApplicationReport>();
for (RMApp application : this.rmContext.getRMApps().values()) {
- reports.add(application.createAndGetApplicationReport());
+ // Only give out the applications viewable by the user as
+ // ApplicationReport has confidential information like client-token, ACLs
+ // etc. Web UI displays all applications though as we filter and print
+ // only public information there.
+ if (checkAccess(callerUGI, application.getUser(),
+ ApplicationAccessType.VIEW_APP, application.getApplicationId())) {
+ reports.add(application.createAndGetApplicationReport());
+ }
}
GetAllApplicationsResponse response =
@@ -346,10 +399,18 @@ public class ClientRMService extends Abs
report.setRackName(rmNode.getRackName());
report.setCapability(rmNode.getTotalCapability());
report.setNodeHealthStatus(rmNode.getNodeHealthStatus());
- org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport schedulerNodeReport = scheduler
- .getNodeReport(rmNode.getNodeID());
- report.setUsed(schedulerNodeReport.getUsedResource());
- report.setNumContainers(schedulerNodeReport.getNumContainers());
+
+ SchedulerNodeReport schedulerNodeReport =
+ scheduler.getNodeReport(rmNode.getNodeID());
+ Resource used = Resources.none();
+ int numContainers = 0;
+ if (schedulerNodeReport != null) {
+ used = schedulerNodeReport.getUsedResource();
+ numContainers = schedulerNodeReport.getNumContainers();
+ }
+ report.setUsed(used);
+ report.setNumContainers(numContainers);
+
return report;
}
@@ -362,11 +423,17 @@ public class ClientRMService extends Abs
return response;
}
+ void refreshServiceAcls(Configuration configuration,
+ PolicyProvider policyProvider) {
+ this.server.refreshServiceAcl(configuration, policyProvider);
+ }
+
@Override
public void stop() {
if (this.server != null) {
- this.server.close();
+ this.server.stop();
}
super.stop();
}
+
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/NMLivelinessMonitor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/NMLivelinessMonitor.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/NMLivelinessMonitor.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/NMLivelinessMonitor.java Wed Nov 2 05:34:31 2011
@@ -39,11 +39,10 @@ public class NMLivelinessMonitor extends
public void init(Configuration conf) {
super.init(conf);
- setExpireInterval(conf.getInt(YarnConfiguration.RM_NM_EXPIRY_INTERVAL_MS,
- YarnConfiguration.DEFAULT_RM_NM_EXPIRY_INTERVAL_MS));
- setMonitorInterval(conf.getInt(
- YarnConfiguration.RM_NM_LIVENESS_MONITOR_INTERVAL_MS,
- YarnConfiguration.DEFAULT_RM_NM_LIVENESS_MONITOR_INTERVAL_MS));
+ int expireIntvl = conf.getInt(YarnConfiguration.RM_NM_EXPIRY_INTERVAL_MS,
+ YarnConfiguration.DEFAULT_RM_NM_EXPIRY_INTERVAL_MS);
+ setExpireInterval(expireIntvl);
+ setMonitorInterval(expireIntvl/3);
}
@Override
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java Wed Nov 2 05:34:31 2011
@@ -18,11 +18,14 @@
package org.apache.hadoop.yarn.server.resourcemanager;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.LinkedList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataInputByteBuffer;
+import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.StringUtils;
@@ -31,8 +34,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.ipc.RPCUtil;
-import org.apache.hadoop.yarn.security.ApplicationTokenIdentifier;
import org.apache.hadoop.yarn.security.client.ClientToAMSecretManager;
+import org.apache.hadoop.yarn.security.client.ClientTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.ApplicationsStore.ApplicationStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
@@ -42,6 +45,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
/**
* This class manages the list of applications for the resource manager.
@@ -57,15 +61,18 @@ public class RMAppManager implements Eve
private final ClientToAMSecretManager clientToAMSecretManager;
private final ApplicationMasterService masterService;
private final YarnScheduler scheduler;
+ private final ApplicationACLsManager applicationACLsManager;
private Configuration conf;
- public RMAppManager(RMContext context, ClientToAMSecretManager
- clientToAMSecretManager, YarnScheduler scheduler,
- ApplicationMasterService masterService, Configuration conf) {
+ public RMAppManager(RMContext context,
+ ClientToAMSecretManager clientToAMSecretManager,
+ YarnScheduler scheduler, ApplicationMasterService masterService,
+ ApplicationACLsManager applicationACLsManager, Configuration conf) {
this.rmContext = context;
this.scheduler = scheduler;
this.clientToAMSecretManager = clientToAMSecretManager;
this.masterService = masterService;
+ this.applicationACLsManager = applicationACLsManager;
this.conf = conf;
setCompletedAppsMax(conf.getInt(
YarnConfiguration.RM_MAX_COMPLETED_APPLICATIONS,
@@ -160,12 +167,17 @@ public class RMAppManager implements Eve
return this.completedApps.size();
}
- protected synchronized void addCompletedApp(ApplicationId appId) {
- if (appId == null) {
+ protected synchronized void finishApplication(ApplicationId applicationId) {
+ if (applicationId == null) {
LOG.error("RMAppManager received completed appId of null, skipping");
} else {
- completedApps.add(appId);
- writeAuditLog(appId);
+ // Inform the DelegationTokenRenewer
+ if (UserGroupInformation.isSecurityEnabled()) {
+ rmContext.getDelegationTokenRenewer().removeApplication(applicationId);
+ }
+
+ completedApps.add(applicationId);
+ writeAuditLog(applicationId);
}
}
@@ -208,21 +220,22 @@ public class RMAppManager implements Eve
LOG.info("Application should be expired, max # apps"
+ " met. Removing app: " + removeId);
rmContext.getRMApps().remove(removeId);
+ this.applicationACLsManager.removeApplication(removeId);
}
}
@SuppressWarnings("unchecked")
protected synchronized void submitApplication(
- ApplicationSubmissionContext submissionContext) {
+ ApplicationSubmissionContext submissionContext, long submitTime) {
ApplicationId applicationId = submissionContext.getApplicationId();
RMApp application = null;
try {
String clientTokenStr = null;
String user = UserGroupInformation.getCurrentUser().getShortUserName();
if (UserGroupInformation.isSecurityEnabled()) {
- Token<ApplicationTokenIdentifier> clientToken = new
- Token<ApplicationTokenIdentifier>(
- new ApplicationTokenIdentifier(applicationId),
+ Token<ClientTokenIdentifier> clientToken = new
+ Token<ClientTokenIdentifier>(
+ new ClientTokenIdentifier(applicationId),
this.clientToAMSecretManager);
clientTokenStr = clientToken.encodeToUrlString();
LOG.debug("Sending client token as " + clientTokenStr);
@@ -241,51 +254,82 @@ public class RMAppManager implements Eve
ApplicationStore appStore = rmContext.getApplicationsStore()
.createApplicationStore(submissionContext.getApplicationId(),
submissionContext);
-
+
// Create RMApp
application = new RMAppImpl(applicationId, rmContext,
this.conf, submissionContext.getApplicationName(), user,
submissionContext.getQueue(), submissionContext, clientTokenStr,
appStore, this.scheduler,
- this.masterService);
+ this.masterService, submitTime);
+ // Sanity check - duplicate?
if (rmContext.getRMApps().putIfAbsent(applicationId, application) !=
null) {
String message = "Application with id " + applicationId
+ " is already present! Cannot add a duplicate!";
LOG.info(message);
throw RPCUtil.getRemoteException(message);
- } else {
- this.rmContext.getDispatcher().getEventHandler().handle(
- new RMAppEvent(applicationId, RMAppEventType.START));
- }
+ }
+
+ // Inform the ACLs Manager
+ this.applicationACLsManager.addApplication(applicationId,
+ submissionContext.getAMContainerSpec().getApplicationACLs());
+
+ // Setup tokens for renewal
+ if (UserGroupInformation.isSecurityEnabled()) {
+ this.rmContext.getDelegationTokenRenewer().addApplication(
+ applicationId,parseCredentials(submissionContext)
+ );
+ }
+
+ // All done, start the RMApp
+ this.rmContext.getDispatcher().getEventHandler().handle(
+ new RMAppEvent(applicationId, RMAppEventType.START));
} catch (IOException ie) {
LOG.info("RMAppManager submit application exception", ie);
if (application != null) {
+ // Sending APP_REJECTED is fine, since we assume that the
+ // RMApp is in NEW state and thus we havne't yet informed the
+ // Scheduler about the existence of the application
this.rmContext.getDispatcher().getEventHandler().handle(
new RMAppRejectedEvent(applicationId, ie.getMessage()));
}
}
}
+
+ private Credentials parseCredentials(ApplicationSubmissionContext application)
+ throws IOException {
+ Credentials credentials = new Credentials();
+ DataInputByteBuffer dibb = new DataInputByteBuffer();
+ ByteBuffer tokens = application.getAMContainerSpec().getContainerTokens();
+ if (tokens != null) {
+ dibb.reset(tokens);
+ credentials.readTokenStorageStream(dibb);
+ tokens.rewind();
+ }
+ return credentials;
+ }
@Override
public void handle(RMAppManagerEvent event) {
- ApplicationId appID = event.getApplicationId();
+ ApplicationId applicationId = event.getApplicationId();
LOG.debug("RMAppManager processing event for "
- + appID + " of type " + event.getType());
+ + applicationId + " of type " + event.getType());
switch(event.getType()) {
case APP_COMPLETED:
{
- addCompletedApp(appID);
- ApplicationSummary.logAppSummary(rmContext.getRMApps().get(appID));
+ finishApplication(applicationId);
+ ApplicationSummary.logAppSummary(
+ rmContext.getRMApps().get(applicationId));
checkAppNumCompletedLimit();
}
break;
case APP_SUBMIT:
{
ApplicationSubmissionContext submissionContext =
- ((RMAppManagerSubmitEvent)event).getSubmissionContext();
- submitApplication(submissionContext);
+ ((RMAppManagerSubmitEvent)event).getSubmissionContext();
+ long submitTime = ((RMAppManagerSubmitEvent)event).getSubmitTime();
+ submitApplication(submissionContext, submitTime);
}
break;
default:
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManagerSubmitEvent.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManagerSubmitEvent.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManagerSubmitEvent.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManagerSubmitEvent.java Wed Nov 2 05:34:31 2011
@@ -23,13 +23,21 @@ import org.apache.hadoop.yarn.api.record
public class RMAppManagerSubmitEvent extends RMAppManagerEvent {
private final ApplicationSubmissionContext submissionContext;
+ private final long submitTime;
- public RMAppManagerSubmitEvent(ApplicationSubmissionContext submissionContext) {
- super(submissionContext.getApplicationId(), RMAppManagerEventType.APP_SUBMIT);
+ public RMAppManagerSubmitEvent(
+ ApplicationSubmissionContext submissionContext, long submitTime) {
+ super(submissionContext.getApplicationId(),
+ RMAppManagerEventType.APP_SUBMIT);
this.submissionContext = submissionContext;
+ this.submitTime = submitTime;
}
public ApplicationSubmissionContext getSubmissionContext() {
return this.submissionContext;
}
+
+ public long getSubmitTime() {
+ return this.submitTime;
+ }
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java Wed Nov 2 05:34:31 2011
@@ -29,7 +29,11 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
+/**
+ * Context of the ResourceManager.
+ */
public interface RMContext {
Dispatcher getDispatcher();
@@ -45,4 +49,6 @@ public interface RMContext {
AMLivelinessMonitor getAMLivelinessMonitor();
ContainerAllocationExpirer getContainerAllocationExpirer();
+
+ DelegationTokenRenewer getDelegationTokenRenewer();
}
\ No newline at end of file
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java Wed Nov 2 05:34:31 2011
@@ -31,6 +31,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
public class RMContextImpl implements RMContext {
@@ -45,14 +46,17 @@ public class RMContextImpl implements RM
private AMLivelinessMonitor amLivelinessMonitor;
private ContainerAllocationExpirer containerAllocationExpirer;
+ private final DelegationTokenRenewer tokenRenewer;
public RMContextImpl(Store store, Dispatcher rmDispatcher,
ContainerAllocationExpirer containerAllocationExpirer,
- AMLivelinessMonitor amLivelinessMonitor) {
+ AMLivelinessMonitor amLivelinessMonitor,
+ DelegationTokenRenewer tokenRenewer) {
this.store = store;
this.rmDispatcher = rmDispatcher;
this.containerAllocationExpirer = containerAllocationExpirer;
this.amLivelinessMonitor = amLivelinessMonitor;
+ this.tokenRenewer = tokenRenewer;
}
@Override
@@ -89,4 +93,9 @@ public class RMContextImpl implements RM
public AMLivelinessMonitor getAMLivelinessMonitor() {
return this.amLivelinessMonitor;
}
+
+ @Override
+ public DelegationTokenRenewer getDelegationTokenRenewer() {
+ return tokenRenewer;
+ }
}
\ No newline at end of file
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java Wed Nov 2 05:34:31 2011
@@ -62,13 +62,20 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.RMWebApp;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
import org.apache.hadoop.yarn.server.security.ContainerTokenSecretManager;
+import org.apache.hadoop.yarn.server.webproxy.AppReportFetcher;
+import org.apache.hadoop.yarn.server.webproxy.WebAppProxy;
+import org.apache.hadoop.yarn.server.webproxy.WebAppProxyServlet;
+import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils;
import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.service.Service;
import org.apache.hadoop.yarn.webapp.WebApp;
import org.apache.hadoop.yarn.webapp.WebApps;
+import org.apache.hadoop.yarn.webapp.WebApps.Builder;
/**
* The ResourceManager is the main class that is a set of components.
@@ -77,7 +84,6 @@ import org.apache.hadoop.yarn.webapp.Web
public class ResourceManager extends CompositeService implements Recoverable {
private static final Log LOG = LogFactory.getLog(ResourceManager.class);
public static final long clusterTimeStamp = System.currentTimeMillis();
- private YarnConfiguration conf;
protected ClientToAMSecretManager clientToAMSecretManager =
new ClientToAMSecretManager();
@@ -100,12 +106,15 @@ public class ResourceManager extends Com
protected NodesListManager nodesListManager;
private EventHandler<SchedulerEvent> schedulerDispatcher;
protected RMAppManager rmAppManager;
+ protected ApplicationACLsManager applicationACLsManager;
private WebApp webApp;
private RMContext rmContext;
private final Store store;
protected ResourceTrackerService resourceTracker;
-
+
+ private Configuration conf;
+
public ResourceManager(Store store) {
super("ResourceManager");
this.store = store;
@@ -119,6 +128,8 @@ public class ResourceManager extends Com
@Override
public synchronized void init(Configuration conf) {
+ this.conf = conf;
+
this.rmDispatcher = createDispatcher();
addIfService(this.rmDispatcher);
@@ -129,13 +140,14 @@ public class ResourceManager extends Com
AMLivelinessMonitor amLivelinessMonitor = createAMLivelinessMonitor();
addService(amLivelinessMonitor);
+ DelegationTokenRenewer tokenRenewer = createDelegationTokenRenewer();
+ addService(tokenRenewer);
+
this.rmContext = new RMContextImpl(this.store, this.rmDispatcher,
- this.containerAllocationExpirer, amLivelinessMonitor);
+ this.containerAllocationExpirer, amLivelinessMonitor, tokenRenewer);
addService(nodesListManager);
- // Initialize the config
- this.conf = new YarnConfiguration(conf);
// Initialize the scheduler
this.scheduler = createScheduler();
this.schedulerDispatcher = createSchedulerEventDispatcher();
@@ -166,7 +178,7 @@ public class ResourceManager extends Com
addService(resourceTracker);
try {
- this.scheduler.reinitialize(this.conf,
+ this.scheduler.reinitialize(conf,
this.containerTokenSecretManager, this.rmContext);
} catch (IOException ioe) {
throw new RuntimeException("Failed to initialize scheduler", ioe);
@@ -175,6 +187,8 @@ public class ResourceManager extends Com
masterService = createApplicationMasterService();
addService(masterService) ;
+ this.applicationACLsManager = new ApplicationACLsManager(conf);
+
this.rmAppManager = createRMAppManager();
// Register event handler for RMAppManagerEvents
this.rmDispatcher.register(RMAppManagerEventType.class,
@@ -183,7 +197,7 @@ public class ResourceManager extends Com
clientRM = createClientRMService();
addService(clientRM);
- adminService = createAdminService();
+ adminService = createAdminService(clientRM, masterService, resourceTracker);
addService(adminService);
this.applicationMasterLauncher = createAMLauncher();
@@ -210,11 +224,9 @@ public class ResourceManager extends Com
}
protected ResourceScheduler createScheduler() {
- return
- ReflectionUtils.newInstance(
- conf.getClass(YarnConfiguration.RM_SCHEDULER,
- FifoScheduler.class, ResourceScheduler.class),
- this.conf);
+ return ReflectionUtils.newInstance(this.conf.getClass(
+ YarnConfiguration.RM_SCHEDULER, FifoScheduler.class,
+ ResourceScheduler.class), this.conf);
}
protected ApplicationMasterLauncher createAMLauncher() {
@@ -231,10 +243,15 @@ public class ResourceManager extends Com
protected AMLivelinessMonitor createAMLivelinessMonitor() {
return new AMLivelinessMonitor(this.rmDispatcher);
}
+
+ protected DelegationTokenRenewer createDelegationTokenRenewer() {
+ return new DelegationTokenRenewer();
+ }
protected RMAppManager createRMAppManager() {
return new RMAppManager(this.rmContext, this.clientToAMSecretManager,
- this.scheduler, this.masterService, this.conf);
+ this.scheduler, this.masterService, this.applicationACLsManager,
+ this.conf);
}
@Private
@@ -250,6 +267,7 @@ public class ResourceManager extends Com
super(SchedulerEventDispatcher.class.getName());
this.scheduler = scheduler;
this.eventProcessor = new Thread(new EventProcessor());
+ this.eventProcessor.setName("ResourceManager Event Processor");
}
@Override
@@ -393,11 +411,18 @@ public class ResourceManager extends Com
}
protected void startWepApp() {
- webApp = WebApps.$for("cluster", masterService).at(
- conf.get(YarnConfiguration.RM_WEBAPP_ADDRESS,
- YarnConfiguration.DEFAULT_RM_WEBAPP_ADDRESS)).
- start(new RMWebApp(this));
-
+ Builder<ApplicationMasterService> builder =
+ WebApps.$for("cluster", masterService).at(
+ this.conf.get(YarnConfiguration.RM_WEBAPP_ADDRESS,
+ YarnConfiguration.DEFAULT_RM_WEBAPP_ADDRESS));
+ if(YarnConfiguration.getRMWebAppHostAndPort(conf).
+ equals(YarnConfiguration.getProxyHostAndPort(conf))) {
+ AppReportFetcher fetcher = new AppReportFetcher(conf, getClientRMService());
+ builder.withServlet(ProxyUriUtils.PROXY_SERVLET_NAME,
+ ProxyUriUtils.PROXY_PATH_SPEC, WebAppProxyServlet.class);
+ builder.withAttribute(WebAppProxy.FETCHER_ATTRIBUTE, fetcher);
+ }
+ webApp = builder.start(new RMWebApp(this));
}
@Override
@@ -425,7 +450,7 @@ public class ResourceManager extends Com
}
protected void doSecureLogin() throws IOException {
- SecurityUtil.login(conf, YarnConfiguration.RM_KEYTAB,
+ SecurityUtil.login(this.conf, YarnConfiguration.RM_KEYTAB,
YarnConfiguration.RM_PRINCIPAL);
}
@@ -451,7 +476,8 @@ public class ResourceManager extends Com
}
protected ClientRMService createClientRMService() {
- return new ClientRMService(this.rmContext, scheduler, this.rmAppManager);
+ return new ClientRMService(this.rmContext, scheduler, this.rmAppManager,
+ this.applicationACLsManager);
}
protected ApplicationMasterService createApplicationMasterService() {
@@ -460,8 +486,13 @@ public class ResourceManager extends Com
}
- protected AdminService createAdminService() {
- return new AdminService(conf, scheduler, rmContext, this.nodesListManager);
+ protected AdminService createAdminService(
+ ClientRMService clientRMService,
+ ApplicationMasterService applicationMasterService,
+ ResourceTrackerService resourceTrackerService) {
+ return new AdminService(this.conf, scheduler, rmContext,
+ this.nodesListManager, clientRMService, applicationMasterService,
+ resourceTrackerService);
}
@Private
@@ -492,6 +523,11 @@ public class ResourceManager extends Com
return this.masterService;
}
+ @Private
+ public ApplicationACLsManager getApplicationACLsManager() {
+ return this.applicationACLsManager;
+ }
+
@Override
public void recover(RMState state) throws Exception {
resourceTracker.recover(state);
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java Wed Nov 2 05:34:31 2011
@@ -17,34 +17,33 @@
*/
package org.apache.hadoop.yarn.server.resourcemanager;
-import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import javax.crypto.SecretKey;
-import org.apache.avro.ipc.Server;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.net.Node;
-import org.apache.hadoop.security.SecurityInfo;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.ipc.YarnRPC;
-import org.apache.hadoop.yarn.server.RMNMSecurityInfoClass;
import org.apache.hadoop.yarn.server.api.ResourceTracker;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerResponse;
import org.apache.hadoop.yarn.server.api.records.HeartbeatResponse;
+import org.apache.hadoop.yarn.server.api.records.NodeAction;
import org.apache.hadoop.yarn.server.api.records.NodeStatus;
import org.apache.hadoop.yarn.server.api.records.RegistrationResponse;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.Store.RMState;
@@ -53,6 +52,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeImpl;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeStatusEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
import org.apache.hadoop.yarn.server.security.ContainerTokenSecretManager;
import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.util.RackResolver;
@@ -75,11 +75,19 @@ public class ResourceTrackerService exte
private static final NodeHeartbeatResponse reboot = recordFactory
.newRecordInstance(NodeHeartbeatResponse.class);
+ private static final NodeHeartbeatResponse shutDown = recordFactory
+ .newRecordInstance(NodeHeartbeatResponse.class);
+
static {
HeartbeatResponse rebootResp = recordFactory
.newRecordInstance(HeartbeatResponse.class);
- rebootResp.setReboot(true);
+ rebootResp.setNodeAction(NodeAction.REBOOT);
reboot.setHeartbeatResponse(rebootResp);
+
+ HeartbeatResponse decommissionedResp = recordFactory
+ .newRecordInstance(HeartbeatResponse.class);
+ decommissionedResp.setNodeAction(NodeAction.SHUTDOWN);
+ shutDown.setHeartbeatResponse(decommissionedResp);
}
public ResourceTrackerService(RMContext rmContext,
@@ -98,7 +106,10 @@ public class ResourceTrackerService exte
String resourceTrackerBindAddress =
conf.get(YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS,
YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_ADDRESS);
- resourceTrackerAddress = NetUtils.createSocketAddr(resourceTrackerBindAddress);
+ resourceTrackerAddress = NetUtils.createSocketAddr(
+ resourceTrackerBindAddress,
+ YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_PORT,
+ YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS);
RackResolver.init(conf);
super.init(conf);
@@ -116,18 +127,26 @@ public class ResourceTrackerService exte
conf, null,
conf.getInt(YarnConfiguration.RM_RESOURCE_TRACKER_CLIENT_THREAD_COUNT,
YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_CLIENT_THREAD_COUNT));
- this.server.start();
+
+ // Enable service authorization?
+ if (conf.getBoolean(
+ CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
+ false)) {
+ refreshServiceAcls(conf, new RMPolicyProvider());
+ }
+ this.server.start();
}
@Override
public synchronized void stop() {
if (this.server != null) {
- this.server.close();
+ this.server.stop();
}
super.stop();
}
+ @SuppressWarnings("unchecked")
@Override
public RegisterNodeManagerResponse registerNodeManager(
RegisterNodeManagerRequest request) throws YarnRemoteException {
@@ -138,121 +157,125 @@ public class ResourceTrackerService exte
int httpPort = request.getHttpPort();
Resource capability = request.getResource();
- try {
- // Check if this node is a 'valid' node
- if (!this.nodesListManager.isValidNode(host)) {
- LOG.info("Disallowed NodeManager from " + host);
- throw new IOException("Disallowed NodeManager from " + host);
- }
-
- RMNode rmNode = new RMNodeImpl(nodeId, rmContext, host, cmPort,
- httpPort, resolve(host), capability);
-
- if (this.rmContext.getRMNodes().putIfAbsent(nodeId, rmNode) != null) {
- throw new IOException("Duplicate registration from the node!");
- }
-
- this.nmLivelinessMonitor.register(nodeId);
-
- LOG.info("NodeManager from node " + host +
- "(cmPort: " + cmPort + " httpPort: " + httpPort + ") "
- + "registered with capability: " + capability.getMemory()
- + ", assigned nodeId " + nodeId);
-
- RegistrationResponse regResponse = recordFactory.newRecordInstance(
- RegistrationResponse.class);
- SecretKey secretKey = this.containerTokenSecretManager
- .createAndGetSecretKey(nodeId.toString());
- regResponse.setSecretKey(ByteBuffer.wrap(secretKey.getEncoded()));
+ RegisterNodeManagerResponse response = recordFactory
+ .newRecordInstance(RegisterNodeManagerResponse.class);
+ RegistrationResponse regResponse = recordFactory
+ .newRecordInstance(RegistrationResponse.class);
+ SecretKey secretKey = this.containerTokenSecretManager
+ .createAndGetSecretKey(nodeId.toString());
+ regResponse.setSecretKey(ByteBuffer.wrap(secretKey.getEncoded()));
+
+ // Check if this node is a 'valid' node
+ if (!this.nodesListManager.isValidNode(host)) {
+ LOG.info("Disallowed NodeManager from " + host
+ + ", Sending SHUTDOWN signal to the NodeManager.");
+ regResponse.setNodeAction(NodeAction.SHUTDOWN);
+ response.setRegistrationResponse(regResponse);
+ return response;
+ }
- RegisterNodeManagerResponse response = recordFactory
- .newRecordInstance(RegisterNodeManagerResponse.class);
+ RMNode rmNode = new RMNodeImpl(nodeId, rmContext, host, cmPort, httpPort,
+ resolve(host), capability);
+
+ if (this.rmContext.getRMNodes().putIfAbsent(nodeId, rmNode) != null) {
+ LOG.info("Duplicate registration from the node at: " + host
+ + ", Sending SHUTDOWN Signal to the NodeManager");
+ regResponse.setNodeAction(NodeAction.SHUTDOWN);
response.setRegistrationResponse(regResponse);
return response;
- } catch (IOException ioe) {
- LOG.info("Exception in node registration from " + nodeId.getHost(), ioe);
- throw RPCUtil.getRemoteException(ioe);
}
+
+ this.rmContext.getDispatcher().getEventHandler().handle(
+ new RMNodeEvent(nodeId, RMNodeEventType.STARTED));
+
+ this.nmLivelinessMonitor.register(nodeId);
+
+ LOG.info("NodeManager from node " + host + "(cmPort: " + cmPort
+ + " httpPort: " + httpPort + ") " + "registered with capability: "
+ + capability.getMemory() + ", assigned nodeId " + nodeId);
+
+ regResponse.setNodeAction(NodeAction.NORMAL);
+ response.setRegistrationResponse(regResponse);
+ return response;
}
+ @SuppressWarnings("unchecked")
@Override
public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)
throws YarnRemoteException {
NodeStatus remoteNodeStatus = request.getNodeStatus();
- try {
- /**
- * Here is the node heartbeat sequence...
- * 1. Check if it's a registered node
- * 2. Check if it's a valid (i.e. not excluded) node
- * 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
- * 4. Send healthStatus to RMNode
- */
-
- NodeId nodeId = remoteNodeStatus.getNodeId();
+ /**
+ * Here is the node heartbeat sequence...
+ * 1. Check if it's a registered node
+ * 2. Check if it's a valid (i.e. not excluded) node
+ * 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
+ * 4. Send healthStatus to RMNode
+ */
+
+ NodeId nodeId = remoteNodeStatus.getNodeId();
+
+ // 1. Check if it's a registered node
+ RMNode rmNode = this.rmContext.getRMNodes().get(nodeId);
+ if (rmNode == null) {
+ /* node does not exist */
+ LOG.info("Node not found rebooting " + remoteNodeStatus.getNodeId());
- // 1. Check if it's a registered node
- RMNode rmNode = this.rmContext.getRMNodes().get(nodeId);
- if (rmNode == null) {
- /* node does not exist */
- LOG.info("Node not found rebooting " + remoteNodeStatus.getNodeId());
- return reboot;
- }
-
- // Send ping
- this.nmLivelinessMonitor.receivedPing(nodeId);
-
- // 2. Check if it's a valid (i.e. not excluded) node
- if (!this.nodesListManager.isValidNode(rmNode.getHostName())) {
- LOG.info("Disallowed NodeManager nodeId: " + nodeId +
- " hostname: " + rmNode.getNodeAddress());
- throw new IOException("Disallowed NodeManager nodeId: " +
- remoteNodeStatus.getNodeId());
- }
-
- NodeHeartbeatResponse nodeHeartBeatResponse = recordFactory
- .newRecordInstance(NodeHeartbeatResponse.class);
-
- // 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
- HeartbeatResponse lastHeartbeatResponse = rmNode
- .getLastHeartBeatResponse();
- if (remoteNodeStatus.getResponseId() + 1 == lastHeartbeatResponse
- .getResponseId()) {
- LOG.info("Received duplicate heartbeat from node " +
- rmNode.getNodeAddress());
- nodeHeartBeatResponse.setHeartbeatResponse(lastHeartbeatResponse);
- return nodeHeartBeatResponse;
- } else if (remoteNodeStatus.getResponseId() + 1 < lastHeartbeatResponse
- .getResponseId()) {
- LOG.info("Too far behind rm response id:" +
- lastHeartbeatResponse.getResponseId() + " nm response id:"
- + remoteNodeStatus.getResponseId());
- // TODO: Just sending reboot is not enough. Think more.
- this.rmContext.getDispatcher().getEventHandler().handle(
- new RMNodeEvent(nodeId, RMNodeEventType.REBOOTING));
- return reboot;
- }
-
- // Heartbeat response
- HeartbeatResponse latestResponse = recordFactory
- .newRecordInstance(HeartbeatResponse.class);
- latestResponse
- .setResponseId(lastHeartbeatResponse.getResponseId() + 1);
- latestResponse.addAllContainersToCleanup(rmNode.pullContainersToCleanUp());
- latestResponse.addAllApplicationsToCleanup(rmNode.pullAppsToCleanup());
+ // Updating the metrics directly as reboot event cannot be
+ // triggered on a null rmNode
+ ClusterMetrics.getMetrics().incrNumRebootedNMs();
+ return reboot;
+ }
- // 4. Send status to RMNode, saving the latest response.
+ // Send ping
+ this.nmLivelinessMonitor.receivedPing(nodeId);
+
+ // 2. Check if it's a valid (i.e. not excluded) node
+ if (!this.nodesListManager.isValidNode(rmNode.getHostName())) {
+ LOG.info("Disallowed NodeManager nodeId: " + nodeId + " hostname: "
+ + rmNode.getNodeAddress());
this.rmContext.getDispatcher().getEventHandler().handle(
- new RMNodeStatusEvent(nodeId, remoteNodeStatus.getNodeHealthStatus(),
- remoteNodeStatus.getContainersStatuses(), latestResponse));
+ new RMNodeEvent(nodeId, RMNodeEventType.DECOMMISSION));
+ return shutDown;
+ }
+
+ NodeHeartbeatResponse nodeHeartBeatResponse = recordFactory
+ .newRecordInstance(NodeHeartbeatResponse.class);
- nodeHeartBeatResponse.setHeartbeatResponse(latestResponse);
+ // 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
+ HeartbeatResponse lastHeartbeatResponse = rmNode.getLastHeartBeatResponse();
+ if (remoteNodeStatus.getResponseId() + 1 == lastHeartbeatResponse
+ .getResponseId()) {
+ LOG.info("Received duplicate heartbeat from node "
+ + rmNode.getNodeAddress());
+ nodeHeartBeatResponse.setHeartbeatResponse(lastHeartbeatResponse);
return nodeHeartBeatResponse;
- } catch (IOException ioe) {
- LOG.info("Exception in heartbeat from node " +
- request.getNodeStatus().getNodeId(), ioe);
- throw RPCUtil.getRemoteException(ioe);
+ } else if (remoteNodeStatus.getResponseId() + 1 < lastHeartbeatResponse
+ .getResponseId()) {
+ LOG.info("Too far behind rm response id:"
+ + lastHeartbeatResponse.getResponseId() + " nm response id:"
+ + remoteNodeStatus.getResponseId());
+ // TODO: Just sending reboot is not enough. Think more.
+ this.rmContext.getDispatcher().getEventHandler().handle(
+ new RMNodeEvent(nodeId, RMNodeEventType.REBOOTING));
+ return reboot;
}
+
+ // Heartbeat response
+ HeartbeatResponse latestResponse = recordFactory
+ .newRecordInstance(HeartbeatResponse.class);
+ latestResponse.setResponseId(lastHeartbeatResponse.getResponseId() + 1);
+ latestResponse.addAllContainersToCleanup(rmNode.pullContainersToCleanUp());
+ latestResponse.addAllApplicationsToCleanup(rmNode.pullAppsToCleanup());
+ latestResponse.setNodeAction(NodeAction.NORMAL);
+
+ // 4. Send status to RMNode, saving the latest response.
+ this.rmContext.getDispatcher().getEventHandler().handle(
+ new RMNodeStatusEvent(nodeId, remoteNodeStatus.getNodeHealthStatus(),
+ remoteNodeStatus.getContainersStatuses(), latestResponse));
+
+ nodeHeartBeatResponse.setHeartbeatResponse(latestResponse);
+ return nodeHeartBeatResponse;
}
public void recover(RMState state) {
@@ -286,4 +309,9 @@ public class ResourceTrackerService exte
return RackResolver.resolve(hostName);
}
+ void refreshServiceAcls(Configuration configuration,
+ PolicyProvider policyProvider) {
+ this.server.refreshServiceAcl(configuration, policyProvider);
+ }
+
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/AMLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/AMLauncher.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/AMLauncher.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/AMLauncher.java Wed Nov 2 05:34:31 2011
@@ -22,8 +22,6 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.security.PrivilegedAction;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
import javax.crypto.SecretKey;
@@ -37,7 +35,6 @@ import org.apache.hadoop.io.DataOutputBu
import org.apache.hadoop.io.Text;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.SecurityInfo;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.StringUtils;
@@ -45,7 +42,6 @@ import org.apache.hadoop.yarn.api.Applic
import org.apache.hadoop.yarn.api.ContainerManager;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -58,9 +54,9 @@ import org.apache.hadoop.yarn.factory.pr
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.security.ApplicationTokenIdentifier;
import org.apache.hadoop.yarn.security.ApplicationTokenSecretManager;
-import org.apache.hadoop.yarn.security.ContainerManagerSecurityInfo;
import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
import org.apache.hadoop.yarn.security.client.ClientToAMSecretManager;
+import org.apache.hadoop.yarn.security.client.ClientTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
@@ -83,6 +79,7 @@ public class AMLauncher implements Runna
private final ApplicationTokenSecretManager applicationTokenSecretManager;
private final ClientToAMSecretManager clientToAMSecretManager;
private final AMLauncherEventType eventType;
+ private final RMContext rmContext;
@SuppressWarnings("rawtypes")
private final EventHandler handler;
@@ -96,15 +93,14 @@ public class AMLauncher implements Runna
this.applicationTokenSecretManager = applicationTokenSecretManager;
this.clientToAMSecretManager = clientToAMSecretManager;
this.eventType = eventType;
+ this.rmContext = rmContext;
this.handler = rmContext.getDispatcher().getEventHandler();
}
private void connect() throws IOException {
ContainerId masterContainerID = application.getMasterContainer().getId();
- containerMgrProxy =
- getContainerMgrProxy(
- masterContainerID.getApplicationAttemptId().getApplicationId());
+ containerMgrProxy = getContainerMgrProxy(masterContainerID);
}
private void launch() throws IOException {
@@ -134,7 +130,7 @@ public class AMLauncher implements Runna
}
protected ContainerManager getContainerMgrProxy(
- final ApplicationId applicationID) throws IOException {
+ final ContainerId containerId) {
Container container = application.getMasterContainer();
@@ -142,8 +138,8 @@ public class AMLauncher implements Runna
final YarnRPC rpc = YarnRPC.create(conf); // TODO: Don't create again and again.
- UserGroupInformation currentUser =
- UserGroupInformation.createRemoteUser("yarn"); // TODO
+ UserGroupInformation currentUser = UserGroupInformation
+ .createRemoteUser(containerId.toString());
if (UserGroupInformation.isSecurityEnabled()) {
ContainerToken containerToken = container.getContainerToken();
Token<ContainerTokenIdentifier> token =
@@ -189,10 +185,25 @@ public class AMLauncher implements Runna
throws IOException {
Map<String, String> environment = container.getEnvironment();
- // Set the AppAttemptId to be consumable by the AM.
- environment.put(ApplicationConstants.APPLICATION_ATTEMPT_ID_ENV,
- application.getAppAttemptId().toString());
-
+ environment.put(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV,
+ application.getWebProxyBase());
+ // Set the AppAttemptId, containerId, NMHTTPAdress, AppSubmitTime to be
+ // consumable by the AM.
+ environment.put(ApplicationConstants.AM_CONTAINER_ID_ENV, container
+ .getContainerId().toString());
+ environment.put(ApplicationConstants.NM_HOST_ENV, application
+ .getMasterContainer().getNodeId().getHost());
+ environment.put(ApplicationConstants.NM_PORT_ENV,
+ String.valueOf(application.getMasterContainer().getNodeId().getPort()));
+ String parts[] =
+ application.getMasterContainer().getNodeHttpAddress().split(":");
+ environment.put(ApplicationConstants.NM_HTTP_PORT_ENV, parts[1]);
+ environment.put(
+ ApplicationConstants.APP_SUBMIT_TIME_ENV,
+ String.valueOf(rmContext.getRMApps()
+ .get(application.getAppAttemptId().getApplicationId())
+ .getSubmitTime()));
+
if (UserGroupInformation.isSecurityEnabled()) {
// TODO: Security enabled/disabled info should come from RM.
@@ -206,7 +217,7 @@ public class AMLauncher implements Runna
}
ApplicationTokenIdentifier id = new ApplicationTokenIdentifier(
- application.getAppAttemptId().getApplicationId());
+ application.getAppAttemptId());
Token<ApplicationTokenIdentifier> token =
new Token<ApplicationTokenIdentifier>(id,
this.applicationTokenSecretManager);
@@ -232,7 +243,7 @@ public class AMLauncher implements Runna
container.setContainerTokens(
ByteBuffer.wrap(dob.getData(), 0, dob.getLength()));
- ApplicationTokenIdentifier identifier = new ApplicationTokenIdentifier(
+ ClientTokenIdentifier identifier = new ClientTokenIdentifier(
application.getAppAttemptId().getApplicationId());
SecretKey clientSecretKey =
this.clientToAMSecretManager.getMasterKey(identifier);
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/ApplicationMasterLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/ApplicationMasterLauncher.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/ApplicationMasterLauncher.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/amlauncher/ApplicationMasterLauncher.java Wed Nov 2 05:34:31 2011
@@ -90,6 +90,11 @@ public class ApplicationMasterLauncher e
}
private class LauncherThread extends Thread {
+
+ public LauncherThread() {
+ super("ApplicationMaster Launcher");
+ }
+
@Override
public void run() {
while (!this.isInterrupted()) {
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/RMAdminProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/RMAdminProtocol.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/RMAdminProtocol.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/RMAdminProtocol.java Wed Nov 2 05:34:31 2011
@@ -25,6 +25,8 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshNodesResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesResponse;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshUserToGroupsMappingsRequest;
@@ -49,4 +51,8 @@ public interface RMAdminProtocol {
public RefreshAdminAclsResponse refreshAdminAcls(
RefreshAdminAclsRequest request)
throws YarnRemoteException;
+
+ public RefreshServiceAclsResponse refreshServiceAcls(
+ RefreshServiceAclsRequest request)
+ throws YarnRemoteException;
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/impl/pb/client/RMAdminProtocolPBClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/impl/pb/client/RMAdminProtocolPBClientImpl.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/impl/pb/client/RMAdminProtocolPBClientImpl.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/api/impl/pb/client/RMAdminProtocolPBClientImpl.java Wed Nov 2 05:34:31 2011
@@ -30,6 +30,7 @@ import org.apache.hadoop.yarn.proto.RMAd
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshAdminAclsRequestProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshNodesRequestProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshQueuesRequestProto;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshServiceAclsRequestProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshSuperUserGroupsConfigurationRequestProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshUserToGroupsMappingsRequestProto;
import org.apache.hadoop.yarn.server.resourcemanager.api.RMAdminProtocol;
@@ -39,6 +40,8 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshNodesResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshQueuesResponse;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshServiceAclsResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationRequest;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshSuperUserGroupsConfigurationResponse;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.RefreshUserToGroupsMappingsRequest;
@@ -49,6 +52,8 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshNodesResponsePBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshQueuesRequestPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshQueuesResponsePBImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshServiceAclsRequestPBImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshServiceAclsResponsePBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshSuperUserGroupsConfigurationRequestPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshSuperUserGroupsConfigurationResponsePBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.api.protocolrecords.impl.pb.RefreshUserToGroupsMappingsRequestPBImpl;
@@ -165,5 +170,24 @@ public class RMAdminProtocolPBClientImpl
}
}
+ @Override
+ public RefreshServiceAclsResponse refreshServiceAcls(
+ RefreshServiceAclsRequest request) throws YarnRemoteException {
+ RefreshServiceAclsRequestProto requestProto =
+ ((RefreshServiceAclsRequestPBImpl)request).getProto();
+ try {
+ return new RefreshServiceAclsResponsePBImpl(
+ proxy.refreshServiceAcls(null, requestProto));
+ } catch (ServiceException e) {
+ if (e.getCause() instanceof YarnRemoteException) {
+ throw (YarnRemoteException)e.getCause();
+ } else if (e.getCause() instanceof UndeclaredThrowableException) {
+ throw (UndeclaredThrowableException)e.getCause();
+ } else {
+ throw new UndeclaredThrowableException(e);
+ }
+ }
+ }
+
}