diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/CommonNodeLabelsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/CommonNodeLabelsManager.java
index fe38164..1300a54 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/CommonNodeLabelsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/CommonNodeLabelsManager.java
@@ -98,6 +98,9 @@
protected NodeLabelsStore store;
private boolean nodeLabelsEnabled = false;
+ @VisibleForTesting
+ boolean isDistributedNodeLabelConfiguration = false;
+
/**
* A Host can have multiple Nodes
*/
@@ -221,6 +224,12 @@ protected void serviceInit(Configuration conf) throws Exception {
nodeLabelsEnabled =
conf.getBoolean(YarnConfiguration.NODE_LABELS_ENABLED,
YarnConfiguration.DEFAULT_NODE_LABELS_ENABLED);
+
+ isDistributedNodeLabelConfiguration =
+ YarnConfiguration.DISTRIBUTED_NODELABEL_CONFIGURATION_TYPE.equals(conf
+ .get(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE,
+ YarnConfiguration.DEFAULT_NODELABEL_CONFIGURATION_TYPE));
+
if (nodeLabelsEnabled) {
initNodeLabelStore(conf);
}
@@ -228,10 +237,14 @@ protected void serviceInit(Configuration conf) throws Exception {
labelCollections.put(NO_LABEL, new RMNodeLabel(NO_LABEL));
}
+ public boolean isDistributedNodeLabelConfiguration() {
+ return isDistributedNodeLabelConfiguration;
+ }
+
protected void initNodeLabelStore(Configuration conf) throws Exception {
this.store = new FileSystemNodeLabelsStore(this);
this.store.init(conf);
- this.store.recover();
+ this.store.recover(isDistributedNodeLabelConfiguration);
}
// for UT purpose
@@ -610,7 +623,7 @@ protected void internalUpdateLabelsOnNodes(
}
}
- if (null != dispatcher) {
+ if (null != dispatcher && !isDistributedNodeLabelConfiguration) {
dispatcher.getEventHandler().handle(
new UpdateNodeToLabelsMappingsEvent(newNMToLabels));
}
@@ -843,7 +856,6 @@ public boolean isExclusiveNodeLabel(String nodeLabel) throws IOException {
readLock.unlock();
}
}
-
private void checkAndThrowLabelName(String label) throws IOException {
if (label == null || label.isEmpty() || label.length() > MAX_LABEL_LENGTH) {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/FileSystemNodeLabelsStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/FileSystemNodeLabelsStore.java
index 0c7192f..afa15fd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/FileSystemNodeLabelsStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/FileSystemNodeLabelsStore.java
@@ -168,7 +168,8 @@ public void updateNodeLabels(List updatedNodeLabels)
}
@Override
- public void recover() throws YarnException, IOException {
+ public void recover(boolean ignoreNodeToLabelsMappings) throws YarnException,
+ IOException {
/*
* Steps of recover
* 1) Read from last mirror (from mirror or mirror.old)
@@ -234,7 +235,9 @@ public void recover() throws YarnException, IOException {
new ReplaceLabelsOnNodeRequestPBImpl(
ReplaceLabelsOnNodeRequestProto.parseDelimitedFrom(is))
.getNodeToLabels();
- mgr.replaceLabelsOnNode(map);
+ if (!ignoreNodeToLabelsMappings) {
+ mgr.replaceLabelsOnNode(map);
+ }
break;
}
case UPDATE_NODE_LABELS: {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/NodeLabelsStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/NodeLabelsStore.java
index 6bd90db..9805d24 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/NodeLabelsStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/nodelabels/NodeLabelsStore.java
@@ -64,7 +64,8 @@ public abstract void updateNodeLabels(
/**
* Recover labels and node to labels mappings from store
*/
- public abstract void recover() throws IOException, YarnException;
+ public abstract void recover(boolean ignoreNodeToLabelsMappings)
+ throws IOException, YarnException;
public void init(Configuration conf) throws Exception {}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestCommonNodeLabelsManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestCommonNodeLabelsManager.java
index 1e2326b..e0f62ed 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestCommonNodeLabelsManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestCommonNodeLabelsManager.java
@@ -564,4 +564,20 @@ public void testUpdateNodeLabels() throws Exception {
Assert.assertFalse(mgr.isExclusiveNodeLabel("p1"));
Assert.assertTrue(mgr.isExclusiveNodeLabel("p2"));
}
+
+ @Test(timeout = 5000)
+ public void testReplaceLabelsOnNodeInDistributedMode() throws Exception {
+ mgr.isDistributedNodeLabelConfiguration = true;
+
+ mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3"));
+ mgr.replaceLabelsOnNode(ImmutableMap.of(toNodeId("n1"), toSet("p1")));
+ Set labelsByNode = mgr.getLabelsByNode(toNodeId("n1"));
+
+ Assert.assertNull(
+ "Labels are not expected to be written to the NodeLabelStore",
+ mgr.lastUpdatedNodeLabels);
+ Assert.assertNotNull("Updated labels should be available from the Mgr",
+ labelsByNode);
+ Assert.assertTrue(labelsByNode.contains("p1"));
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestFileSystemNodeLabelsStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestFileSystemNodeLabelsStore.java
index 6694290..64e04cf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestFileSystemNodeLabelsStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/nodelabels/TestFileSystemNodeLabelsStore.java
@@ -143,6 +143,40 @@ public void testRecoverWithMirror() throws Exception {
"p2", toSet(toNodeId("n2"))));
mgr.stop();
}
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Test(timeout = 10000)
+ public void testRecoverWithDistributedNodeLabels() throws Exception {
+ mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3"));
+ mgr.addToCluserNodeLabels(toSet("p4"));
+ mgr.addToCluserNodeLabels(toSet("p5", "p6"));
+ mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n1"), toSet("p1"),
+ toNodeId("n2"), toSet("p2")));
+ mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"),
+ toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"),
+ toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6")));
+
+ mgr.removeFromClusterNodeLabels(toSet("p1"));
+ mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5"));
+ mgr.stop();
+
+ mgr = new MockNodeLabelManager();
+ Configuration cf = new Configuration(conf);
+ cf.set(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE,
+ YarnConfiguration.DISTRIBUTED_NODELABEL_CONFIGURATION_TYPE);
+ mgr.init(cf);
+
+ // check variables
+ Assert.assertEquals(3, mgr.getClusterNodeLabels().size());
+ Assert.assertTrue(mgr.getClusterNodeLabels().containsAll(
+ Arrays.asList("p2", "p4", "p6")));
+
+ Assert.assertTrue("During recovery in distributed node-labels setup, "
+ + "node to labels mapping should not be recovered ", mgr
+ .getNodeLabels().size() == 0);
+
+ mgr.stop();
+ }
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test(timeout = 10000)
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java
index 12714de..13ec187 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java
@@ -111,6 +111,9 @@
private final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null);
+ @VisibleForTesting
+ boolean isDistributedNodeLabelConfiguration = false;
+
public AdminService(ResourceManager rm, RMContext rmContext) {
super(AdminService.class.getName());
this.rm = rm;
@@ -140,6 +143,11 @@ public void serviceInit(Configuration conf) throws Exception {
YarnConfiguration.DEFAULT_YARN_ADMIN_ACL)), UserGroupInformation
.getCurrentUser());
rmId = conf.get(YarnConfiguration.RM_HA_ID);
+
+ isDistributedNodeLabelConfiguration =
+ rm.getRMContext().getNodeLabelManager()
+ .isDistributedNodeLabelConfiguration();
+
super.serviceInit(conf);
}
@@ -626,32 +634,35 @@ public AddToClusterNodeLabelsResponse addToClusterNodeLabels(AddToClusterNodeLab
@Override
public RemoveFromClusterNodeLabelsResponse removeFromClusterNodeLabels(
RemoveFromClusterNodeLabelsRequest request) throws YarnException, IOException {
- String argName = "removeFromClusterNodeLabels";
+ String operation = "removeFromClusterNodeLabels";
final String msg = "remove labels.";
- UserGroupInformation user = checkAcls(argName);
- checkRMStatus(user.getShortUserName(), argName, msg);
+ checkAndThrowIfDistributedNodeLabelConfEnabled(operation);
+ UserGroupInformation user = checkAcls(operation);
+
+ checkRMStatus(user.getShortUserName(), operation, msg);
RemoveFromClusterNodeLabelsResponse response =
recordFactory.newRecordInstance(RemoveFromClusterNodeLabelsResponse.class);
try {
rmContext.getNodeLabelManager().removeFromClusterNodeLabels(request.getNodeLabels());
RMAuditLogger
- .logSuccess(user.getShortUserName(), argName, "AdminService");
+ .logSuccess(user.getShortUserName(), operation, "AdminService");
return response;
} catch (IOException ioe) {
- throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
+ throw logAndWrapException(ioe, user.getShortUserName(), operation, msg);
}
}
@Override
public ReplaceLabelsOnNodeResponse replaceLabelsOnNode(
ReplaceLabelsOnNodeRequest request) throws YarnException, IOException {
- String argName = "replaceLabelsOnNode";
+ String operation = "replaceLabelsOnNode";
final String msg = "set node to labels.";
- UserGroupInformation user = checkAcls(argName);
+ checkAndThrowIfDistributedNodeLabelConfEnabled(operation);
+ UserGroupInformation user = checkAcls(operation);
- checkRMStatus(user.getShortUserName(), argName, msg);
+ checkRMStatus(user.getShortUserName(), operation, msg);
ReplaceLabelsOnNodeResponse response =
recordFactory.newRecordInstance(ReplaceLabelsOnNodeResponse.class);
@@ -659,10 +670,10 @@ public ReplaceLabelsOnNodeResponse replaceLabelsOnNode(
rmContext.getNodeLabelManager().replaceLabelsOnNode(
request.getNodeToLabels());
RMAuditLogger
- .logSuccess(user.getShortUserName(), argName, "AdminService");
+ .logSuccess(user.getShortUserName(), operation, "AdminService");
return response;
} catch (IOException ioe) {
- throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
+ throw logAndWrapException(ioe, user.getShortUserName(), operation, msg);
}
}
@@ -688,20 +699,31 @@ public UpdateNodeLabelsResponse updateNodeLabels(
}
}
- private void checkRMStatus(String user, String argName, String msg)
+ private void checkRMStatus(String user, String operation, String msg)
throws StandbyException {
if (!isRMActive()) {
- RMAuditLogger.logFailure(user, argName, "",
+ RMAuditLogger.logFailure(user, operation, "",
"AdminService", "ResourceManager is not active. Can not " + msg);
throwStandbyException();
}
}
private YarnException logAndWrapException(Exception exception, String user,
- String argName, String msg) throws YarnException {
+ String operation, String msg) throws YarnException {
LOG.warn("Exception " + msg, exception);
- RMAuditLogger.logFailure(user, argName, "",
+ RMAuditLogger.logFailure(user, operation, "",
"AdminService", "Exception " + msg);
return RPCUtil.getRemoteException(exception);
}
+
+ private void checkAndThrowIfDistributedNodeLabelConfEnabled(String operation)
+ throws YarnException {
+ if (isDistributedNodeLabelConfiguration) {
+ String msg =
+ String.format("Error when invoke method=%s because of "
+ + "distributed node label configuration enabled.", operation);
+ LOG.error(msg);
+ throw RPCUtil.getRemoteException(new IOException(msg));
+ }
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java
index 584da7d..aa2edb8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java
@@ -20,8 +20,8 @@
import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
-import java.security.AccessControlException;
import java.nio.ByteBuffer;
+import java.security.AccessControlException;
import java.security.Principal;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
@@ -64,25 +64,25 @@
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.hadoop.security.authentication.server.KerberosAuthenticationHandler;
import org.apache.hadoop.security.authorize.AuthorizationException;
+import org.apache.hadoop.security.token.SecretManager.InvalidToken;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticationHandler;
import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
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.security.token.SecretManager.InvalidToken;
-import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.MoveApplicationAcrossQueuesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.RenewDelegationTokenRequest;
import org.apache.hadoop.yarn.api.protocolrecords.RenewDelegationTokenResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.MoveApplicationAcrossQueuesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse;
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
@@ -116,12 +116,11 @@
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppAttemptInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppAttemptsInfo;
-import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NewApplication;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppInfo;
-import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppState;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppQueue;
-import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ApplicationSubmissionContextInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppState;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ApplicationStatisticsInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ApplicationSubmissionContextInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppsInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedulerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ClusterInfo;
@@ -131,14 +130,15 @@
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.FairSchedulerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.FifoSchedulerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.LocalResourceInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NewApplication;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeLabelsInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeToLabelsInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodesInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ResourceInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.SchedulerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.SchedulerTypeInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.StatisticsItemInfo;
-import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeLabelsInfo;
-import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodeToLabelsInfo;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.AdHocLogDumper;
import org.apache.hadoop.yarn.util.ConverterUtils;
@@ -146,6 +146,7 @@
import org.apache.hadoop.yarn.webapp.NotFoundException;
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
+import com.google.common.annotations.VisibleForTesting;
import com.google.inject.Inject;
import com.google.inject.Singleton;
@@ -162,6 +163,9 @@
private final Configuration conf;
private @Context HttpServletResponse response;
+ @VisibleForTesting
+ boolean isDistributedNodeLabelConfiguration = false;
+
public final static String DELEGATION_TOKEN_HEADER =
"Hadoop-YARN-RM-Delegation-Token";
@@ -169,6 +173,20 @@
public RMWebServices(final ResourceManager rm, Configuration conf) {
this.rm = rm;
this.conf = conf;
+ isDistributedNodeLabelConfiguration =
+ rm.getRMContext().getNodeLabelManager()
+ .isDistributedNodeLabelConfiguration();
+ }
+
+ private void checkAndThrowIfDistributedNodeLabelConfEnabled(String operation)
+ throws IOException {
+ if (isDistributedNodeLabelConfiguration) {
+ String msg =
+ String.format("Error when invoke method=%s because of "
+ + "distributed node label configuration enabled.", operation);
+ LOG.error(msg);
+ throw new IOException(msg);
+ }
}
RMWebServices(ResourceManager rm, Configuration conf,
@@ -790,7 +808,9 @@ public Response replaceLabelsOnNodes(
@Context HttpServletRequest hsr)
throws IOException {
init();
-
+
+ checkAndThrowIfDistributedNodeLabelConfEnabled("replaceLabelsOnNodes");
+
UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
if (callerUGI == null) {
String msg = "Unable to obtain user name, user not authenticated for"
@@ -908,7 +928,9 @@ public Response replaceLabelsOnNode(NodeLabelsInfo newNodeLabelsInfo,
@Context HttpServletRequest hsr, @PathParam("nodeId") String nodeId)
throws Exception {
init();
-
+
+ checkAndThrowIfDistributedNodeLabelConfEnabled("replaceLabelsOnNode");
+
UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
if (callerUGI == null) {
String msg = "Unable to obtain user name, user not authenticated for"
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAdminService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAdminService.java
index 1bd3dbf..7cfd215 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAdminService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMAdminService.java
@@ -43,6 +43,7 @@
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.security.authorize.ProxyUsers;
import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
+import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.conf.HAUtil;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -52,6 +53,8 @@
import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshServiceAclsRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshSuperUserGroupsConfigurationRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshUserToGroupsMappingsRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReplaceLabelsOnNodeRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.junit.After;
@@ -59,6 +62,8 @@
import org.junit.Before;
import org.junit.Test;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
public class TestRMAdminService {
@@ -751,6 +756,45 @@ public void testRMInitialsWithFileSystemBasedConfigurationProvider()
}
}
+ @Test
+ public void testModifyLabelsOnNodesWithDistributedConfigurationDisabled()
+ throws IOException, YarnException {
+ // create RM and set it's ACTIVE
+ MockRM rm = new MockRM();
+ ((RMContextImpl) rm.getRMContext())
+ .setHAServiceState(HAServiceState.ACTIVE);
+ RMNodeLabelsManager labelMgr = rm.rmContext.getNodeLabelManager();
+
+ // by default, distributed configuration for node label is disabled, this
+ // should pass
+ labelMgr.addToCluserNodeLabels(ImmutableSet.of("x", "y"));
+ rm.adminService.replaceLabelsOnNode(ReplaceLabelsOnNodeRequest
+ .newInstance(ImmutableMap.of(NodeId.newInstance("host", 0),
+ (Set) ImmutableSet.of("x"))));
+ rm.close();
+ }
+
+ @Test(expected = YarnException.class)
+ public void testModifyLabelsOnNodesWithDistributedConfigurationEnabled()
+ throws IOException, YarnException {
+ // create RM and set it's ACTIVE, and set distributed node label
+ // configuration to true
+ MockRM rm = new MockRM();
+ rm.adminService.isDistributedNodeLabelConfiguration = true;
+
+ ((RMContextImpl) rm.getRMContext())
+ .setHAServiceState(HAServiceState.ACTIVE);
+ RMNodeLabelsManager labelMgr = rm.rmContext.getNodeLabelManager();
+
+ // by default, distributed configuration for node label is disabled, this
+ // should pass
+ labelMgr.addToCluserNodeLabels(ImmutableSet.of("x", "y"));
+ rm.adminService.replaceLabelsOnNode(ReplaceLabelsOnNodeRequest
+ .newInstance(ImmutableMap.of(NodeId.newInstance("host", 0),
+ (Set) ImmutableSet.of("x"))));
+ rm.close();
+ }
+
private String writeConfigurationXML(Configuration conf, String confXMLName)
throws IOException {
DataOutputStream output = null;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodeLabels.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodeLabels.java
index 0cc576c..bdf8b2c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodeLabels.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodeLabels.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.yarn.server.resourcemanager.webapp;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
@@ -64,13 +65,13 @@
private String userName;
private String notUserName;
+ private RMWebServices rmWebService;
private Injector injector = Guice.createInjector(new ServletModule() {
+
@Override
protected void configureServlets() {
bind(JAXBContextResolver.class);
- bind(RMWebServices.class);
- bind(GenericExceptionHandler.class);
try {
userName = UserGroupInformation.getCurrentUser().getShortUserName();
} catch (IOException ioe) {
@@ -81,6 +82,9 @@ protected void configureServlets() {
conf = new YarnConfiguration();
conf.set(YarnConfiguration.YARN_ADMIN_ACL, userName);
rm = new MockRM(conf);
+ rmWebService = new RMWebServices(rm,conf);
+ bind(RMWebServices.class).toInstance(rmWebService);
+ bind(GenericExceptionHandler.class);
bind(ResourceManager.class).toInstance(rm);
filter("/*").through(
TestRMWebServicesAppsModification.TestRMCustomAuthFilter.class);
@@ -331,6 +335,74 @@ public void testNodeLabels() throws JSONException, Exception {
assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
String res = response.getEntity(String.class);
assertTrue(res.equals("null"));
+
+ // Following test cases are to test replace when distributed node label
+ // configuration is on
+ // Reset for testing : add cluster labels
+ response =
+ r.path("ws")
+ .path("v1")
+ .path("cluster")
+ .path("add-node-labels")
+ .queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON)
+ .entity("{\"nodeLabels\":[\"x\",\"y\"]}",
+ MediaType.APPLICATION_JSON).post(ClientResponse.class);
+ // Reset for testing : Add labels to a node
+ response =
+ r.path("ws").path("v1").path("cluster").path("nodes").path("nid:0")
+ .path("replace-labels").queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON)
+ .entity("{\"nodeLabels\": [\"y\"]}", MediaType.APPLICATION_JSON)
+ .post(ClientResponse.class);
+ LOG.info("posted node nodelabel");
+
+ rmWebService.isDistributedNodeLabelConfiguration = true;
+ // Case1 : Replace labels using node-to-labels
+ ntli = new NodeToLabelsInfo();
+ nli = new NodeLabelsInfo();
+ nli.getNodeLabels().add("x");
+ ntli.getNodeToLabels().put("nid:0", nli);
+ response =
+ r.path("ws")
+ .path("v1")
+ .path("cluster")
+ .path("replace-node-to-labels")
+ .queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON)
+ .entity(toJson(ntli, NodeToLabelsInfo.class),
+ MediaType.APPLICATION_JSON).post(ClientResponse.class);
+
+ // Verify, using node-to-labels that previous operation has failed
+ response =
+ r.path("ws").path("v1").path("cluster").path("get-node-to-labels")
+ .queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ ntli = response.getEntity(NodeToLabelsInfo.class);
+ nli = ntli.getNodeToLabels().get("nid:0");
+ assertEquals(1, nli.getNodeLabels().size());
+ assertFalse(nli.getNodeLabels().contains("x"));
+
+ // Case2 : failure to Replace labels using replace-labels
+ response =
+ r.path("ws").path("v1").path("cluster").path("nodes").path("nid:0")
+ .path("replace-labels").queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON)
+ .entity("{\"nodeLabels\": [\"x\"]}", MediaType.APPLICATION_JSON)
+ .post(ClientResponse.class);
+ LOG.info("posted node nodelabel");
+
+ // Verify, using node-to-labels that previous operation has failed
+ response =
+ r.path("ws").path("v1").path("cluster").path("get-node-to-labels")
+ .queryParam("user.name", userName)
+ .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ ntli = response.getEntity(NodeToLabelsInfo.class);
+ nli = ntli.getNodeToLabels().get("nid:0");
+ assertEquals(1, nli.getNodeLabels().size());
+ assertFalse(nli.getNodeLabels().contains("x"));
}
@SuppressWarnings("rawtypes")
@@ -341,13 +413,4 @@ private String toJson(Object nsli, Class klass) throws Exception {
jm.marshallToJSON(nsli, sw);
return sw.toString();
}
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private Object fromJson(String json, Class klass) throws Exception {
- StringReader sr = new StringReader(json);
- JSONJAXBContext ctx = new JSONJAXBContext(klass);
- JSONUnmarshaller jm = ctx.createJSONUnmarshaller();
- return jm.unmarshalFromJSON(sr, klass);
- }
-
}
--
1.9.5.msysgit.1