You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by vi...@apache.org on 2013/08/12 23:26:19 UTC
svn commit: r1513258 [8/10] - in
/hadoop/common/branches/YARN-321/hadoop-yarn-project: ./ hadoop-yarn/
hadoop-yarn/conf/ hadoop-yarn/dev-support/ hadoop-yarn/hadoop-yarn-api/
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/
hadoop-...
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java Mon Aug 12 21:25:49 2013
@@ -18,6 +18,7 @@
package org.apache.hadoop.yarn.server.resourcemanager.rmapp;
+import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
@@ -32,6 +33,7 @@ import java.util.concurrent.locks.Reentr
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.ExitUtil;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -47,6 +49,7 @@ import org.apache.hadoop.yarn.conf.YarnC
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.security.client.ClientToAMTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEvent;
@@ -60,6 +63,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
@@ -445,6 +449,7 @@ public class RMAppImpl implements RMApp,
FinalApplicationStatus finishState = getFinalApplicationStatus();
String diags = UNAVAILABLE;
float progress = 0.0f;
+ org.apache.hadoop.yarn.api.records.Token amrmToken = null;
if (allowAccess) {
if (this.currentAttempt != null) {
currentApplicationAttemptId = this.currentAttempt.getAppAttemptId();
@@ -466,6 +471,24 @@ public class RMAppImpl implements RMApp,
progress = currentAttempt.getProgress();
}
diags = this.diagnostics.toString();
+
+ if (currentAttempt != null &&
+ currentAttempt.getAppAttemptState() == RMAppAttemptState.LAUNCHED) {
+ try {
+ if (getApplicationSubmissionContext().getUnmanagedAM() &&
+ getUser().equals(UserGroupInformation.getCurrentUser().getUserName())) {
+ Token<AMRMTokenIdentifier> token = currentAttempt.getAMRMToken();
+ if (token != null) {
+ amrmToken = BuilderUtils.newAMRMToken(token.getIdentifier(),
+ token.getKind().toString(), token.getPassword(),
+ token.getService().toString());
+ }
+ }
+ } catch (IOException ex) {
+ LOG.warn("UserGroupInformation.getCurrentUser() error: " +
+ ex.toString(), ex);
+ }
+ }
}
if (currentApplicationAttemptId == null) {
@@ -479,7 +502,8 @@ public class RMAppImpl implements RMApp,
this.name, host, rpcPort, clientToAMToken,
createApplicationState(this.stateMachine.getCurrentState()), diags,
trackingUrl, this.startTime, this.finishTime, finishState,
- appUsageReport, origTrackingUrl, progress, this.applicationType);
+ appUsageReport, origTrackingUrl, progress, this.applicationType,
+ amrmToken);
} finally {
this.readLock.unlock();
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java Mon Aug 12 21:25:49 2013
@@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.re
import static org.apache.hadoop.yarn.util.StringHelper.pjoin;
-import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
@@ -41,7 +40,6 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.http.HttpConfig;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.ExitUtil;
@@ -62,7 +60,6 @@ import org.apache.hadoop.yarn.event.Even
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
-import org.apache.hadoop.yarn.security.AMRMTokenSelector;
import org.apache.hadoop.yarn.security.client.ClientToAMTokenIdentifier;
import org.apache.hadoop.yarn.security.client.ClientToAMTokenSelector;
import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
@@ -682,23 +679,20 @@ public class RMAppAttemptImpl implements
this.clientToAMToken =
clientToAMTokenSelector.selectToken(new Text(),
appAttemptTokens.getAllTokens());
-
- InetSocketAddress serviceAddr = conf.getSocketAddr(
- YarnConfiguration.RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
- AMRMTokenSelector appTokenSelector = new AMRMTokenSelector();
- this.amrmToken =
- appTokenSelector.selectToken(
- SecurityUtil.buildTokenService(serviceAddr),
- appAttemptTokens.getAllTokens());
-
- // For now, no need to populate tokens back to
- // AMRMTokenSecretManager, because running attempts are rebooted
- // Later in work-preserve restart, we'll create NEW->RUNNING transition
- // in which the restored tokens will be added to the secret manager
}
+
+ // Only one AMRMToken is stored per-attempt, so this should be fine. Can't
+ // use TokenSelector as service may change - think fail-over.
+ this.amrmToken =
+ (Token<AMRMTokenIdentifier>) appAttemptTokens
+ .getToken(RMStateStore.AM_RM_TOKEN_SERVICE);
+
+ // For now, no need to populate tokens back to AMRMTokenSecretManager,
+ // because running attempts are rebooted. Later in work-preserve restart,
+ // we'll create NEW->RUNNING transition in which the restored tokens will be
+ // added to the secret manager
}
+
private static class BaseTransition implements
SingleArcTransition<RMAppAttemptImpl, RMAppAttemptEvent> {
@@ -730,26 +724,15 @@ public class RMAppAttemptImpl implements
new Token<ClientToAMTokenIdentifier>(new ClientToAMTokenIdentifier(
appAttempt.applicationAttemptId),
appAttempt.rmContext.getClientToAMTokenSecretManager());
-
- // create application token
- AMRMTokenIdentifier id =
- new AMRMTokenIdentifier(appAttempt.applicationAttemptId);
- Token<AMRMTokenIdentifier> amRmToken =
- new Token<AMRMTokenIdentifier>(id,
- appAttempt.rmContext.getAMRMTokenSecretManager());
- InetSocketAddress serviceAddr =
- appAttempt.conf.getSocketAddr(
- YarnConfiguration.RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
- // normally the client should set the service after acquiring the
- // token, but this token is directly provided to the AMs
- SecurityUtil.setTokenService(amRmToken, serviceAddr);
-
- appAttempt.amrmToken = amRmToken;
-
}
+ // create AMRMToken
+ AMRMTokenIdentifier id =
+ new AMRMTokenIdentifier(appAttempt.applicationAttemptId);
+ appAttempt.amrmToken =
+ new Token<AMRMTokenIdentifier>(id,
+ appAttempt.rmContext.getAMRMTokenSecretManager());
+
// Add the application to the scheduler
appAttempt.eventHandler.handle(
new AppAddedSchedulerEvent(appAttempt.applicationAttemptId,
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java Mon Aug 12 21:25:49 2013
@@ -501,8 +501,13 @@ public class RMNodeImpl implements RMNod
public void transition(RMNodeImpl rmNode, RMNodeEvent event) {
// Inform the scheduler
rmNode.nodeUpdateQueue.clear();
- rmNode.context.getDispatcher().getEventHandler().handle(
- new NodeRemovedSchedulerEvent(rmNode));
+ // If the current state is NodeState.UNHEALTHY
+ // Then node is already been removed from the
+ // Scheduler
+ if (!rmNode.getState().equals(NodeState.UNHEALTHY)) {
+ rmNode.context.getDispatcher().getEventHandler()
+ .handle(new NodeRemovedSchedulerEvent(rmNode));
+ }
rmNode.context.getDispatcher().getEventHandler().handle(
new NodesListManagerEvent(
NodesListManagerEventType.NODE_UNUSABLE, rmNode));
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java Mon Aug 12 21:25:49 2013
@@ -126,18 +126,6 @@ public class QueueMetrics implements Met
enableUserMetrics, conf);
}
- // this method is here because we want to make sure these metrics show up on
- // queue registration.
- public void initMetrics() {
- appsSubmitted.incr(0);
- appsRunning.incr(0);
- appsPending.incr(0);
- appsCompleted.incr(0);
- appsKilled.incr(0);
- appsFailed.incr(0);
- reservedContainers.incr(0);
- }
-
/**
* Helper method to clear cache - used only for unit tests.
*/
@@ -168,7 +156,6 @@ public class QueueMetrics implements Met
ms.register(
sourceName(queueName).toString(),
"Metrics for queue: " + queueName, metrics);
- metrics.initMetrics();
}
queueMetrics.put(queueName, metrics);
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java Mon Aug 12 21:25:49 2013
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -28,6 +29,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.exceptions.InvalidResourceBlacklistRequestException;
+import org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
@@ -155,6 +158,9 @@ public class SchedulerUtils {
/**
* Utility method to validate a resource request, by insuring that the
* requested memory/vcore is non-negative and not greater than max
+ *
+ * @throws <code>InvalidResourceRequestException</code> when there is invalid
+ * request
*/
public static void validateResourceRequest(ResourceRequest resReq,
Resource maximumResource) throws InvalidResourceRequestException {
@@ -177,26 +183,4 @@ public class SchedulerUtils {
+ ", maxVirtualCores=" + maximumResource.getVirtualCores());
}
}
-
- /**
- * Utility method to validate a list resource requests, by insuring that the
- * requested memory/vcore is non-negative and not greater than max
- */
- public static void validateResourceRequests(List<ResourceRequest> ask,
- Resource maximumResource) throws InvalidResourceRequestException {
- for (ResourceRequest resReq : ask) {
- validateResourceRequest(resReq, maximumResource);
- }
- }
-
- public static void validateBlacklistRequest(ResourceBlacklistRequest blacklistRequest)
- throws InvalidResourceBlacklistRequestException {
- if (blacklistRequest != null) {
- List<String> plus = blacklistRequest.getBlacklistAdditions();
- if (plus != null && plus.contains(ResourceRequest.ANY)) {
- throw new InvalidResourceBlacklistRequestException(
- "Cannot add " + ResourceRequest.ANY + " to the blacklist!");
- }
- }
- }
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java Mon Aug 12 21:25:49 2013
@@ -185,12 +185,13 @@ extends org.apache.hadoop.yarn.server.re
* <code>null</code> if it was just a reservation
* @param containerStatus <code>ContainerStatus</code> for the completed
* container
+ * @param childQueue <code>CSQueue</code> to reinsert in childQueues
* @param event event to be sent to the container
*/
public void completedContainer(Resource clusterResource,
FiCaSchedulerApp application, FiCaSchedulerNode node,
RMContainer container, ContainerStatus containerStatus,
- RMContainerEventType event);
+ RMContainerEventType event, CSQueue childQueue);
/**
* Get the number of applications in the queue.
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java Mon Aug 12 21:25:49 2013
@@ -673,7 +673,7 @@ public class CapacityScheduler
SchedulerUtils.createAbnormalContainerStatus(
container.getId(),
SchedulerUtils.UNRESERVED_CONTAINER),
- RMContainerEventType.RELEASED);
+ RMContainerEventType.RELEASED, null);
}
}
@@ -828,7 +828,7 @@ public class CapacityScheduler
// Inform the queue
LeafQueue queue = (LeafQueue)application.getQueue();
queue.completedContainer(clusterResource, application, node,
- rmContainer, containerStatus, event);
+ rmContainer, containerStatus, event, null);
LOG.info("Application " + applicationAttemptId +
" released container " + container.getId() +
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java Mon Aug 12 21:25:49 2013
@@ -171,9 +171,8 @@ public class LeafQueue implements CSQueu
maxApplicationsPerUser =
(int)(maxApplications * (userLimit / 100.0f) * userLimitFactor);
- this.maxAMResourcePerQueuePercent =
- cs.getConfiguration().
- getMaximumApplicationMasterResourcePerQueuePercent(getQueuePath());
+ float maxAMResourcePerQueuePercent = cs.getConfiguration()
+ .getMaximumApplicationMasterResourcePerQueuePercent(getQueuePath());
int maxActiveApplications =
CSQueueUtils.computeMaxActiveApplications(
resourceCalculator,
@@ -202,9 +201,9 @@ public class LeafQueue implements CSQueu
capacity, absoluteCapacity,
maximumCapacity, absoluteMaxCapacity,
userLimit, userLimitFactor,
- maxApplications, maxApplicationsPerUser,
- maxActiveApplications, maxActiveApplicationsPerUser,
- state, acls, cs.getConfiguration().getNodeLocalityDelay());
+ maxApplications, maxAMResourcePerQueuePercent, maxApplicationsPerUser,
+ maxActiveApplications, maxActiveApplicationsPerUser, state, acls, cs
+ .getConfiguration().getNodeLocalityDelay());
if(LOG.isDebugEnabled()) {
LOG.debug("LeafQueue:" + " name=" + queueName
@@ -223,10 +222,10 @@ public class LeafQueue implements CSQueu
float capacity, float absoluteCapacity,
float maximumCapacity, float absoluteMaxCapacity,
int userLimit, float userLimitFactor,
- int maxApplications, int maxApplicationsPerUser,
- int maxActiveApplications, int maxActiveApplicationsPerUser,
- QueueState state, Map<QueueACL, AccessControlList> acls,
- int nodeLocalityDelay)
+ int maxApplications, float maxAMResourcePerQueuePercent,
+ int maxApplicationsPerUser, int maxActiveApplications,
+ int maxActiveApplicationsPerUser, QueueState state,
+ Map<QueueACL, AccessControlList> acls, int nodeLocalityDelay)
{
// Sanity check
CSQueueUtils.checkMaxCapacity(getQueueName(), capacity, maximumCapacity);
@@ -243,6 +242,7 @@ public class LeafQueue implements CSQueu
this.userLimitFactor = userLimitFactor;
this.maxApplications = maxApplications;
+ this.maxAMResourcePerQueuePercent = maxAMResourcePerQueuePercent;
this.maxApplicationsPerUser = maxApplicationsPerUser;
this.maxActiveApplications = maxActiveApplications;
@@ -392,6 +392,14 @@ public class LeafQueue implements CSQueu
public float getMinimumAllocationFactor() {
return minimumAllocationFactor;
}
+
+ /**
+ * Used only by tests.
+ */
+ @Private
+ public float getMaxAMResourcePerQueuePercent() {
+ return maxAMResourcePerQueuePercent;
+ }
public int getMaxApplications() {
return maxApplications;
@@ -604,6 +612,7 @@ public class LeafQueue implements CSQueu
newlyParsedLeafQueue.absoluteMaxCapacity,
newlyParsedLeafQueue.userLimit, newlyParsedLeafQueue.userLimitFactor,
newlyParsedLeafQueue.maxApplications,
+ newlyParsedLeafQueue.maxAMResourcePerQueuePercent,
newlyParsedLeafQueue.getMaxApplicationsPerUser(),
newlyParsedLeafQueue.getMaximumActiveApplications(),
newlyParsedLeafQueue.getMaximumActiveApplicationsPerUser(),
@@ -1407,7 +1416,7 @@ public class LeafQueue implements CSQueu
@Override
public void completedContainer(Resource clusterResource,
FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer,
- ContainerStatus containerStatus, RMContainerEventType event) {
+ ContainerStatus containerStatus, RMContainerEventType event, CSQueue childQueue) {
if (application != null) {
// Careful! Locking order is important!
synchronized (this) {
@@ -1442,7 +1451,7 @@ public class LeafQueue implements CSQueu
" cluster=" + clusterResource);
// Inform the parent queue
getParent().completedContainer(clusterResource, application,
- node, rmContainer, null, event);
+ node, rmContainer, null, event, this);
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java Mon Aug 12 21:25:49 2013
@@ -655,7 +655,7 @@ public class ParentQueue implements CSQu
assignment.getResource(), Resources.none())) {
// Remove and re-insert to sort
iter.remove();
- LOG.info("Re-sorting queues since queue: " + childQueue.getQueuePath() +
+ LOG.info("Re-sorting assigned queue: " + childQueue.getQueuePath() +
" stats: " + childQueue);
childQueues.add(childQueue);
if (LOG.isDebugEnabled()) {
@@ -685,7 +685,8 @@ public class ParentQueue implements CSQu
@Override
public void completedContainer(Resource clusterResource,
FiCaSchedulerApp application, FiCaSchedulerNode node,
- RMContainer rmContainer, ContainerStatus containerStatus, RMContainerEventType event) {
+ RMContainer rmContainer, ContainerStatus containerStatus,
+ RMContainerEventType event, CSQueue completedChildQueue) {
if (application != null) {
// Careful! Locking order is important!
// Book keeping
@@ -701,10 +702,24 @@ public class ParentQueue implements CSQu
" cluster=" + clusterResource);
}
+ // reinsert the updated queue
+ for (Iterator<CSQueue> iter=childQueues.iterator(); iter.hasNext();) {
+ CSQueue csqueue = iter.next();
+ if(csqueue.equals(completedChildQueue))
+ {
+ iter.remove();
+ LOG.info("Re-sorting completed queue: " + csqueue.getQueuePath() +
+ " stats: " + csqueue);
+ childQueues.add(csqueue);
+ break;
+ }
+ }
+
// Inform the parent
if (parent != null) {
+ // complete my parent
parent.completedContainer(clusterResource, application,
- node, rmContainer, null, event);
+ node, rmContainer, null, event, this);
}
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueueMetrics.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueueMetrics.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueueMetrics.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueueMetrics.java Mon Aug 12 21:25:49 2013
@@ -94,7 +94,6 @@ public class FSQueueMetrics extends Queu
metrics = ms.register(
sourceName(queueName).toString(),
"Metrics for queue: " + queueName, metrics);
- metrics.initMetrics();
}
queueMetrics.put(queueName, metrics);
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java Mon Aug 12 21:25:49 2013
@@ -604,6 +604,15 @@ public class FairScheduler implements Re
*/
protected synchronized void addApplication(
ApplicationAttemptId applicationAttemptId, String queueName, String user) {
+ if (queueName == null || queueName.isEmpty()) {
+ String message = "Reject application " + applicationAttemptId +
+ " submitted by user " + user + " with an empty queue name.";
+ LOG.info(message);
+ rmContext.getDispatcher().getEventHandler().handle(
+ new RMAppAttemptRejectedEvent(applicationAttemptId, message));
+ return;
+ }
+
RMApp rmApp = rmContext.getRMApps().get(applicationAttemptId);
FSLeafQueue queue = assignToQueue(rmApp, queueName, user);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java Mon Aug 12 21:25:49 2013
@@ -21,6 +21,8 @@ package org.apache.hadoop.yarn.server.re
import java.io.IOException;
import java.util.Collection;
import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import javax.servlet.http.HttpServletRequest;
@@ -50,6 +52,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
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;
@@ -58,6 +61,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedulerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ClusterInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ClusterMetricsInfo;
+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.NodeInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodesInfo;
@@ -143,6 +147,9 @@ public class RMWebServices {
CapacityScheduler cs = (CapacityScheduler) rs;
CSQueue root = cs.getRootQueue();
sinfo = new CapacitySchedulerInfo(root);
+ } else if (rs instanceof FairScheduler) {
+ FairScheduler fs = (FairScheduler) rs;
+ sinfo = new FairSchedulerInfo(fs);
} else if (rs instanceof FifoScheduler) {
sinfo = new FifoSchedulerInfo(this.rm);
} else {
@@ -231,11 +238,13 @@ public class RMWebServices {
@QueryParam("startedTimeBegin") String startedBegin,
@QueryParam("startedTimeEnd") String startedEnd,
@QueryParam("finishedTimeBegin") String finishBegin,
- @QueryParam("finishedTimeEnd") String finishEnd) {
+ @QueryParam("finishedTimeEnd") String finishEnd,
+ @QueryParam("applicationTypes") Set<String> applicationTypes) {
long num = 0;
boolean checkCount = false;
boolean checkStart = false;
boolean checkEnd = false;
+ boolean checkAppTypes = false;
long countNum = 0;
// set values suitable in case both of begin/end not specified
@@ -291,6 +300,27 @@ public class RMWebServices {
"finishTimeEnd must be greater than finishTimeBegin");
}
+ Set<String> appTypes = new HashSet<String>();
+ if (!applicationTypes.isEmpty()) {
+ for (String applicationType : applicationTypes) {
+ if (applicationType != null && !applicationType.trim().isEmpty()) {
+ if (applicationType.indexOf(",") == -1) {
+ appTypes.add(applicationType.trim());
+ } else {
+ String[] types = applicationType.split(",");
+ for (String type : types) {
+ if (!type.trim().isEmpty()) {
+ appTypes.add(type.trim());
+ }
+ }
+ }
+ }
+ }
+ }
+ if (!appTypes.isEmpty()) {
+ checkAppTypes = true;
+ }
+
final ConcurrentMap<ApplicationId, RMApp> apps = rm.getRMContext()
.getRMApps();
AppsInfo allApps = new AppsInfo();
@@ -332,6 +362,10 @@ public class RMWebServices {
continue;
}
}
+ if (checkAppTypes
+ && !appTypes.contains(rmapp.getApplicationType())) {
+ continue;
+ }
if (checkStart
&& (rmapp.getStartTime() < sBegin || rmapp.getStartTime() > sEnd)) {
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ClusterMetricsInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ClusterMetricsInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ClusterMetricsInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ClusterMetricsInfo.java Mon Aug 12 21:25:49 2013
@@ -77,7 +77,7 @@ public class ClusterMetricsInfo {
this.containersPending = metrics.getPendingContainers();
this.containersReserved = metrics.getReservedContainers();
- this.totalMB = availableMB + reservedMB + allocatedMB;
+ this.totalMB = availableMB + allocatedMB;
this.activeNodes = clusterMetrics.getNumActiveNMs();
this.lostNodes = clusterMetrics.getNumLostNMs();
this.unhealthyNodes = clusterMetrics.getUnhealthyNMs();
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java Mon Aug 12 21:25:49 2013
@@ -18,14 +18,31 @@
package org.apache.hadoop.yarn.server.resourcemanager.webapp.dao;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlTransient;
+import javax.xml.bind.annotation.XmlType;
+
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
-public class FairSchedulerInfo {
+@XmlRootElement(name = "fairScheduler")
+@XmlType(name = "fairScheduler")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class FairSchedulerInfo extends SchedulerInfo {
+ private FairSchedulerQueueInfo rootQueue;
+
+ @XmlTransient
private FairScheduler scheduler;
+ public FairSchedulerInfo() {
+ } // JAXB needs this
+
public FairSchedulerInfo(FairScheduler fs) {
scheduler = fs;
+ rootQueue = new FairSchedulerQueueInfo(scheduler.getQueueManager().
+ getRootQueue(), scheduler);
}
public int getAppFairShare(ApplicationAttemptId appAttemptId) {
@@ -34,7 +51,6 @@ public class FairSchedulerInfo {
}
public FairSchedulerQueueInfo getRootQueueInfo() {
- return new FairSchedulerQueueInfo(scheduler.getQueueManager().
- getRootQueue(), scheduler);
+ return rootQueue;
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java Mon Aug 12 21:25:49 2013
@@ -20,14 +20,23 @@ package org.apache.hadoop.yarn.server.re
import java.util.Collection;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.AppSchedulable;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.FIELD)
public class FairSchedulerLeafQueueInfo extends FairSchedulerQueueInfo {
private int numPendingApps;
private int numActiveApps;
+ public FairSchedulerLeafQueueInfo() {
+ }
+
public FairSchedulerLeafQueueInfo(FSLeafQueue queue, FairScheduler scheduler) {
super(queue, scheduler);
Collection<AppSchedulable> apps = queue.getAppSchedulables();
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerQueueInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerQueueInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerQueueInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerQueueInfo.java Mon Aug 12 21:25:49 2013
@@ -22,6 +22,12 @@ package org.apache.hadoop.yarn.server.re
import java.util.ArrayList;
import java.util.Collection;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlSeeAlso;
+import javax.xml.bind.annotation.XmlTransient;
+
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueue;
@@ -29,39 +35,53 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.QueueManager;
import org.apache.hadoop.yarn.util.resource.Resources;
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.FIELD)
+@XmlSeeAlso({FairSchedulerLeafQueueInfo.class})
public class FairSchedulerQueueInfo {
private int maxApps;
+ @XmlTransient
private float fractionMemUsed;
+ @XmlTransient
private float fractionMemFairShare;
+ @XmlTransient
private float fractionMemMinShare;
+ @XmlTransient
private float fractionMemMaxShare;
- private Resource minResources;
- private Resource maxResources;
- private Resource usedResources;
- private Resource fairResources;
- private Resource clusterResources;
+ private ResourceInfo minResources;
+ private ResourceInfo maxResources;
+ private ResourceInfo usedResources;
+ private ResourceInfo fairResources;
+ private ResourceInfo clusterResources;
private String queueName;
+ private String schedulingPolicy;
+
+ private Collection<FairSchedulerQueueInfo> childQueues;
- private Collection<FairSchedulerQueueInfo> childInfos;
+ public FairSchedulerQueueInfo() {
+ }
public FairSchedulerQueueInfo(FSQueue queue, FairScheduler scheduler) {
QueueManager manager = scheduler.getQueueManager();
queueName = queue.getName();
-
- clusterResources = scheduler.getClusterCapacity();
+ schedulingPolicy = queue.getPolicy().getName();
+
+ clusterResources = new ResourceInfo(scheduler.getClusterCapacity());
- usedResources = queue.getResourceUsage();
+ usedResources = new ResourceInfo(queue.getResourceUsage());
fractionMemUsed = (float)usedResources.getMemory() /
clusterResources.getMemory();
- fairResources = queue.getFairShare();
- minResources = queue.getMinShare();
- maxResources = queue.getMaxShare();
- maxResources = Resources.componentwiseMin(maxResources, clusterResources);
+ fairResources = new ResourceInfo(queue.getFairShare());
+ minResources = new ResourceInfo(queue.getMinShare());
+ maxResources = new ResourceInfo(queue.getMaxShare());
+ maxResources = new ResourceInfo(
+ Resources.componentwiseMin(queue.getMaxShare(),
+ scheduler.getClusterCapacity()));
fractionMemFairShare = (float)fairResources.getMemory() / clusterResources.getMemory();
fractionMemMinShare = (float)minResources.getMemory() / clusterResources.getMemory();
@@ -69,13 +89,13 @@ public class FairSchedulerQueueInfo {
maxApps = manager.getQueueMaxApps(queueName);
- Collection<FSQueue> childQueues = queue.getChildQueues();
- childInfos = new ArrayList<FairSchedulerQueueInfo>();
- for (FSQueue child : childQueues) {
+ Collection<FSQueue> children = queue.getChildQueues();
+ childQueues = new ArrayList<FairSchedulerQueueInfo>();
+ for (FSQueue child : children) {
if (child instanceof FSLeafQueue) {
- childInfos.add(new FairSchedulerLeafQueueInfo((FSLeafQueue)child, scheduler));
+ childQueues.add(new FairSchedulerLeafQueueInfo((FSLeafQueue)child, scheduler));
} else {
- childInfos.add(new FairSchedulerQueueInfo(child, scheduler));
+ childQueues.add(new FairSchedulerQueueInfo(child, scheduler));
}
}
}
@@ -90,15 +110,15 @@ public class FairSchedulerQueueInfo {
/**
* Returns the fair share of this queue in megabytes.
*/
- public Resource getFairShare() {
+ public ResourceInfo getFairShare() {
return fairResources;
}
- public Resource getMinResources() {
+ public ResourceInfo getMinResources() {
return minResources;
}
- public Resource getMaxResources() {
+ public ResourceInfo getMaxResources() {
return maxResources;
}
@@ -110,7 +130,7 @@ public class FairSchedulerQueueInfo {
return queueName;
}
- public Resource getUsedResources() {
+ public ResourceInfo getUsedResources() {
return usedResources;
}
@@ -138,7 +158,14 @@ public class FairSchedulerQueueInfo {
return fractionMemMaxShare;
}
+ /**
+ * Returns the name of the scheduling policy used by this queue.
+ */
+ public String getSchedulingPolicy() {
+ return schedulingPolicy;
+ }
+
public Collection<FairSchedulerQueueInfo> getChildQueues() {
- return childInfos;
+ return childQueues;
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ResourceInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ResourceInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ResourceInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/ResourceInfo.java Mon Aug 12 21:25:49 2013
@@ -45,4 +45,9 @@ public class ResourceInfo {
public int getvCores() {
return vCores;
}
+
+ @Override
+ public String toString() {
+ return "<memory:" + memory + ", vCores:" + vCores + ">";
+ }
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/SchedulerInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/SchedulerInfo.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/SchedulerInfo.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/SchedulerInfo.java Mon Aug 12 21:25:49 2013
@@ -22,7 +22,8 @@ import javax.xml.bind.annotation.XmlRoot
import javax.xml.bind.annotation.XmlSeeAlso;
@XmlRootElement
-@XmlSeeAlso({ CapacitySchedulerInfo.class, FifoSchedulerInfo.class })
+@XmlSeeAlso({ CapacitySchedulerInfo.class, FairSchedulerInfo.class,
+ FifoSchedulerInfo.class })
public class SchedulerInfo {
public SchedulerInfo() {
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java Mon Aug 12 21:25:49 2013
@@ -35,7 +35,8 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.net.NetworkTopology;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -212,9 +213,11 @@ public class Application {
NodeManager nodeManager = task.getNodeManager();
ContainerId containerId = task.getContainerId();
task.stop();
- StopContainerRequest stopRequest = recordFactory.newRecordInstance(StopContainerRequest.class);
- stopRequest.setContainerId(containerId);
- nodeManager.stopContainer(stopRequest);
+ List<ContainerId> containerIds = new ArrayList<ContainerId>();
+ containerIds.add(containerId);
+ StopContainersRequest stopRequest =
+ StopContainersRequest.newInstance(containerIds);
+ nodeManager.stopContainers(stopRequest);
Resources.subtractFrom(used, requestSpec.get(task.getPriority()));
@@ -339,10 +342,15 @@ public class Application {
updateResourceRequests(requests.get(priority), type, task);
// Launch the container
- StartContainerRequest startRequest = recordFactory.newRecordInstance(StartContainerRequest.class);
- startRequest.setContainerLaunchContext(createCLC());
- startRequest.setContainerToken(container.getContainerToken());
- nodeManager.startContainer(startRequest);
+ StartContainerRequest scRequest =
+ StartContainerRequest.newInstance(createCLC(),
+ container.getContainerToken());
+ List<StartContainerRequest> list =
+ new ArrayList<StartContainerRequest>();
+ list.add(scRequest);
+ StartContainersRequest allRequests =
+ StartContainersRequest.newInstance(list);
+ nodeManager.startContainers(allRequests);
break;
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java Mon Aug 12 21:25:49 2013
@@ -18,11 +18,15 @@
package org.apache.hadoop.yarn.server.resourcemanager;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.List;
import junit.framework.Assert;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
@@ -35,6 +39,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
@@ -89,13 +94,28 @@ public class MockAM {
waitForState(RMAppAttemptState.LAUNCHED);
}
responseId = 0;
- RegisterApplicationMasterRequest req =
+ final RegisterApplicationMasterRequest req =
Records.newRecord(RegisterApplicationMasterRequest.class);
- req.setApplicationAttemptId(attemptId);
req.setHost("");
req.setRpcPort(1);
req.setTrackingUrl("");
- return amRMProtocol.registerApplicationMaster(req);
+ UserGroupInformation ugi =
+ UserGroupInformation.createRemoteUser(attemptId.toString());
+ Token<AMRMTokenIdentifier> token =
+ context.getRMApps().get(attemptId.getApplicationId())
+ .getRMAppAttempt(attemptId).getAMRMToken();
+ ugi.addTokenIdentifier(token.decodeIdentifier());
+ try {
+ return ugi
+ .doAs(new PrivilegedExceptionAction<RegisterApplicationMasterResponse>() {
+ @Override
+ public RegisterApplicationMasterResponse run() throws Exception {
+ return amRMProtocol.registerApplicationMaster(req);
+ }
+ });
+ } catch (UndeclaredThrowableException e) {
+ throw (Exception) e.getCause();
+ }
}
public void addRequests(String[] hosts, int memory, int priority,
@@ -110,6 +130,9 @@ public class MockAM {
return response;
}
+ public void addContainerToBeReleased(ContainerId containerId) {
+ releases.add(containerId);
+ }
public AllocateResponse allocate(
String host, int memory, int numContainers,
List<ContainerId> releases) throws Exception {
@@ -153,18 +176,44 @@ public class MockAM {
public AllocateResponse allocate(
List<ResourceRequest> resourceRequest, List<ContainerId> releases)
throws Exception {
- AllocateRequest req = AllocateRequest.newInstance(attemptId,
- ++responseId, 0F, resourceRequest, releases, null);
- return amRMProtocol.allocate(req);
+ final AllocateRequest req =
+ AllocateRequest.newInstance(++responseId, 0F, resourceRequest,
+ releases, null);
+ UserGroupInformation ugi =
+ UserGroupInformation.createRemoteUser(attemptId.toString());
+ Token<AMRMTokenIdentifier> token =
+ context.getRMApps().get(attemptId.getApplicationId())
+ .getRMAppAttempt(attemptId).getAMRMToken();
+ ugi.addTokenIdentifier(token.decodeIdentifier());
+ try {
+ return ugi.doAs(new PrivilegedExceptionAction<AllocateResponse>() {
+ @Override
+ public AllocateResponse run() throws Exception {
+ return amRMProtocol.allocate(req);
+ }
+ });
+ } catch (UndeclaredThrowableException e) {
+ throw (Exception) e.getCause();
+ }
}
public void unregisterAppAttempt() throws Exception {
waitForState(RMAppAttemptState.RUNNING);
- FinishApplicationMasterRequest req = Records.newRecord(FinishApplicationMasterRequest.class);
- req.setAppAttemptId(attemptId);
- req.setDiagnostics("");
- req.setFinalApplicationStatus(FinalApplicationStatus.SUCCEEDED);
- req.setTrackingUrl("");
- amRMProtocol.finishApplicationMaster(req);
+ final FinishApplicationMasterRequest req =
+ FinishApplicationMasterRequest.newInstance(
+ FinalApplicationStatus.SUCCEEDED, "", "");
+ UserGroupInformation ugi =
+ UserGroupInformation.createRemoteUser(attemptId.toString());
+ Token<AMRMTokenIdentifier> token =
+ context.getRMApps().get(attemptId.getApplicationId())
+ .getRMAppAttempt(attemptId).getAMRMToken();
+ ugi.addTokenIdentifier(token.decodeIdentifier());
+ ugi.doAs(new PrivilegedExceptionAction<Object>() {
+ @Override
+ public Object run() throws Exception {
+ amRMProtocol.finishApplicationMaster(req);
+ return null;
+ }
+ });
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java Mon Aug 12 21:25:49 2013
@@ -30,6 +30,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.ContainerStatus;
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.server.api.protocolrecords.NodeHeartbeatRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerRequest;
@@ -45,14 +46,24 @@ public class MockNM {
private int responseId;
private NodeId nodeId;
private final int memory;
- private final int vCores = 1;
+ private final int vCores;
private ResourceTrackerService resourceTracker;
private final int httpPort = 2;
private MasterKey currentContainerTokenMasterKey;
private MasterKey currentNMTokenMasterKey;
public MockNM(String nodeIdStr, int memory, ResourceTrackerService resourceTracker) {
+ // scale vcores based on the requested memory
+ this(nodeIdStr, memory,
+ Math.max(1, (memory * YarnConfiguration.DEFAULT_NM_VCORES) /
+ YarnConfiguration.DEFAULT_NM_PMEM_MB),
+ resourceTracker);
+ }
+
+ public MockNM(String nodeIdStr, int memory, int vcores,
+ ResourceTrackerService resourceTracker) {
this.memory = memory;
+ this.vCores = vcores;
this.resourceTracker = resourceTracker;
String[] splits = nodeIdStr.split(":");
nodeId = BuilderUtils.newNodeId(splits[0], Integer.parseInt(splits[1]));
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRMWithCustomAMLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRMWithCustomAMLauncher.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRMWithCustomAMLauncher.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRMWithCustomAMLauncher.java Mon Aug 12 21:25:49 2013
@@ -18,9 +18,15 @@
package org.apache.hadoop.yarn.server.resourcemanager;
+import java.net.InetSocketAddress;
+
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncher;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.ApplicationMasterLauncher;
@@ -52,6 +58,17 @@ public class MockRMWithCustomAMLauncher
ContainerId containerId) {
return containerManager;
}
+ @Override
+ protected Token<AMRMTokenIdentifier> getAMRMToken() {
+ Token<AMRMTokenIdentifier> amRmToken = super.getAMRMToken();
+ InetSocketAddress serviceAddr =
+ getConfig().getSocketAddr(
+ YarnConfiguration.RM_SCHEDULER_ADDRESS,
+ YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
+ YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
+ SecurityUtil.setTokenService(amRmToken, serviceAddr);
+ return amRmToken;
+ }
};
}
};
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/NodeManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/NodeManager.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/NodeManager.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/NodeManager.java Mon Aug 12 21:25:49 2013
@@ -31,12 +31,13 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersResponse;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -158,61 +159,60 @@ public class NodeManager implements Cont
}
@Override
- synchronized public StartContainerResponse startContainer(
- StartContainerRequest request)
+ synchronized public StartContainersResponse startContainers(
+ StartContainersRequest requests)
throws YarnException {
- Token containerToken = request.getContainerToken();
- ContainerTokenIdentifier tokenId = null;
+ for (StartContainerRequest request : requests.getStartContainerRequests()) {
+ Token containerToken = request.getContainerToken();
+ ContainerTokenIdentifier tokenId = null;
+
+ try {
+ tokenId = BuilderUtils.newContainerTokenIdentifier(containerToken);
+ } catch (IOException e) {
+ throw RPCUtil.getRemoteException(e);
+ }
- try {
- tokenId = BuilderUtils.newContainerTokenIdentifier(containerToken);
- } catch (IOException e) {
- throw RPCUtil.getRemoteException(e);
- }
+ ContainerId containerID = tokenId.getContainerID();
+ ApplicationId applicationId =
+ containerID.getApplicationAttemptId().getApplicationId();
+
+ List<Container> applicationContainers = containers.get(applicationId);
+ if (applicationContainers == null) {
+ applicationContainers = new ArrayList<Container>();
+ containers.put(applicationId, applicationContainers);
+ }
- ContainerId containerID = tokenId.getContainerID();
- ApplicationId applicationId =
- containerID.getApplicationAttemptId().getApplicationId();
-
- List<Container> applicationContainers = containers.get(applicationId);
- if (applicationContainers == null) {
- applicationContainers = new ArrayList<Container>();
- containers.put(applicationId, applicationContainers);
- }
-
- // Sanity check
- for (Container container : applicationContainers) {
- if (container.getId().compareTo(containerID)
- == 0) {
- throw new IllegalStateException(
- "Container " + containerID +
- " already setup on node " + containerManagerAddress);
+ // Sanity check
+ for (Container container : applicationContainers) {
+ if (container.getId().compareTo(containerID) == 0) {
+ throw new IllegalStateException("Container " + containerID
+ + " already setup on node " + containerManagerAddress);
+ }
}
- }
- Container container =
- BuilderUtils.newContainer(containerID,
- this.nodeId, nodeHttpAddress,
- tokenId.getResource(),
- null, null // DKDC - Doesn't matter
+ Container container =
+ BuilderUtils.newContainer(containerID, this.nodeId, nodeHttpAddress,
+ tokenId.getResource(), null, null // DKDC - Doesn't matter
);
- ContainerStatus containerStatus =
- BuilderUtils.newContainerStatus(container.getId(), ContainerState.NEW,
- "", -1000);
- applicationContainers.add(container);
- containerStatusMap.put(container, containerStatus);
- Resources.subtractFrom(available, tokenId.getResource());
- Resources.addTo(used, tokenId.getResource());
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("startContainer:" + " node=" + containerManagerAddress
- + " application=" + applicationId + " container=" + container
- + " available=" + available + " used=" + used);
- }
+ ContainerStatus containerStatus =
+ BuilderUtils.newContainerStatus(container.getId(),
+ ContainerState.NEW, "", -1000);
+ applicationContainers.add(container);
+ containerStatusMap.put(container, containerStatus);
+ Resources.subtractFrom(available, tokenId.getResource());
+ Resources.addTo(used, tokenId.getResource());
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("startContainer:" + " node=" + containerManagerAddress
+ + " application=" + applicationId + " container=" + container
+ + " available=" + available + " used=" + used);
+ }
- StartContainerResponse response = recordFactory.newRecordInstance(StartContainerResponse.class);
+ }
+ StartContainersResponse response =
+ StartContainersResponse.newInstance(null, null, null);
return response;
}
@@ -225,76 +225,79 @@ public class NodeManager implements Cont
}
@Override
- synchronized public StopContainerResponse stopContainer(StopContainerRequest request)
+ synchronized public StopContainersResponse stopContainers(StopContainersRequest request)
throws YarnException {
- ContainerId containerID = request.getContainerId();
- String applicationId = String.valueOf(
- containerID.getApplicationAttemptId().getApplicationId().getId());
-
- // Mark the container as COMPLETE
- List<Container> applicationContainers = containers.get(applicationId);
- for (Container c : applicationContainers) {
- if (c.getId().compareTo(containerID) == 0) {
- ContainerStatus containerStatus = containerStatusMap.get(c);
- containerStatus.setState(ContainerState.COMPLETE);
- containerStatusMap.put(c, containerStatus);
+ for (ContainerId containerID : request.getContainerIds()) {
+ String applicationId =
+ String.valueOf(containerID.getApplicationAttemptId()
+ .getApplicationId().getId());
+
+ // Mark the container as COMPLETE
+ List<Container> applicationContainers = containers.get(applicationId);
+ for (Container c : applicationContainers) {
+ if (c.getId().compareTo(containerID) == 0) {
+ ContainerStatus containerStatus = containerStatusMap.get(c);
+ containerStatus.setState(ContainerState.COMPLETE);
+ containerStatusMap.put(c, containerStatus);
+ }
}
- }
-
- // Send a heartbeat
- try {
- heartbeat();
- } catch (IOException ioe) {
- throw RPCUtil.getRemoteException(ioe);
- }
-
- // Remove container and update status
- int ctr = 0;
- Container container = null;
- for (Iterator<Container> i=applicationContainers.iterator(); i.hasNext();) {
- container = i.next();
- if (container.getId().compareTo(containerID) == 0) {
- i.remove();
- ++ctr;
+
+ // Send a heartbeat
+ try {
+ heartbeat();
+ } catch (IOException ioe) {
+ throw RPCUtil.getRemoteException(ioe);
}
- }
-
- if (ctr != 1) {
- throw new IllegalStateException("Container " + containerID +
- " stopped " + ctr + " times!");
- }
-
- Resources.addTo(available, container.getResource());
- Resources.subtractFrom(used, container.getResource());
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("stopContainer:" + " node=" + containerManagerAddress
- + " application=" + applicationId + " container=" + containerID
- + " available=" + available + " used=" + used);
- }
- StopContainerResponse response = recordFactory.newRecordInstance(StopContainerResponse.class);
- return response;
+ // Remove container and update status
+ int ctr = 0;
+ Container container = null;
+ for (Iterator<Container> i = applicationContainers.iterator(); i
+ .hasNext();) {
+ container = i.next();
+ if (container.getId().compareTo(containerID) == 0) {
+ i.remove();
+ ++ctr;
+ }
+ }
+
+ if (ctr != 1) {
+ throw new IllegalStateException("Container " + containerID
+ + " stopped " + ctr + " times!");
+ }
+
+ Resources.addTo(available, container.getResource());
+ Resources.subtractFrom(used, container.getResource());
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("stopContainer:" + " node=" + containerManagerAddress
+ + " application=" + applicationId + " container=" + containerID
+ + " available=" + available + " used=" + used);
+ }
+ }
+ return StopContainersResponse.newInstance(null,null);
}
@Override
- synchronized public GetContainerStatusResponse getContainerStatus(GetContainerStatusRequest request) throws YarnException {
- ContainerId containerId = request.getContainerId();
- List<Container> appContainers =
- containers.get(
- containerId.getApplicationAttemptId().getApplicationId());
- Container container = null;
- for (Container c : appContainers) {
- if (c.getId().equals(containerId)) {
- container = c;
+ synchronized public GetContainerStatusesResponse getContainerStatuses(
+ GetContainerStatusesRequest request) throws YarnException {
+ List<ContainerStatus> statuses = new ArrayList<ContainerStatus>();
+ for (ContainerId containerId : request.getContainerIds()) {
+ List<Container> appContainers =
+ containers.get(containerId.getApplicationAttemptId()
+ .getApplicationId());
+ Container container = null;
+ for (Container c : appContainers) {
+ if (c.getId().equals(containerId)) {
+ container = c;
+ }
+ }
+ if (container != null
+ && containerStatusMap.get(container).getState() != null) {
+ statuses.add(containerStatusMap.get(container));
}
}
- GetContainerStatusResponse response =
- recordFactory.newRecordInstance(GetContainerStatusResponse.class);
- if (container != null && containerStatusMap.get(container).getState() != null) {
- response.setStatus(containerStatusMap.get(container));
- }
- return response;
+ return GetContainerStatusesResponse.newInstance(statuses, null);
}
public static org.apache.hadoop.yarn.server.api.records.NodeStatus