diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java index 0e60fd5..bc22073 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java @@ -32,6 +32,7 @@ import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.server.resourcemanager.RMContext; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; import org.apache.hadoop.yarn.util.resource.Resources; public class RMAppAttemptMetrics { @@ -49,6 +50,10 @@ private AtomicLong finishedVcoreSeconds = new AtomicLong(0); private RMContext rmContext; + private int[][] localityStatistics = + new int[NodeType.values().length][NodeType.values().length]; + private volatile int totalAllocatedContainers; + public RMAppAttemptMetrics(ApplicationAttemptId attemptId, RMContext rmContext) { this.attemptId = attemptId; @@ -57,7 +62,7 @@ public RMAppAttemptMetrics(ApplicationAttemptId attemptId, this.writeLock = lock.writeLock(); this.rmContext = rmContext; } - + public void updatePreemptionInfo(Resource resource, RMContainer container) { try { writeLock.lock(); @@ -126,4 +131,18 @@ public void updateAggregateAppResourceUsage(long finishedMemorySeconds, this.finishedMemorySeconds.addAndGet(finishedMemorySeconds); this.finishedVcoreSeconds.addAndGet(finishedVcoreSeconds); } + + public void incNumAllocatedContainers(NodeType containerType, + NodeType requestType) { + localityStatistics[containerType.index][requestType.index]++; + totalAllocatedContainers++; + } + + public int[][] getLocalityStatistics() { + return this.localityStatistics; + } + + public int getTotalAllocatedContainers() { + return this.totalAllocatedContainers; + } } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/NodeType.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/NodeType.java index 821ec24..2b193bb 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/NodeType.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/NodeType.java @@ -22,7 +22,10 @@ * Resource classification. */ public enum NodeType { - NODE_LOCAL, - RACK_LOCAL, - OFF_SWITCH + NODE_LOCAL(0), RACK_LOCAL(1), OFF_SWITCH(2); + public int index; + + private NodeType(int index) { + this.index = index; + } } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java index 532df05..ed78097 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java @@ -46,6 +46,7 @@ import org.apache.hadoop.yarn.api.records.ResourceRequest; import org.apache.hadoop.yarn.server.resourcemanager.RMContext; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AggregateAppResourceUsage; +import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; @@ -78,7 +79,7 @@ private long lastVcoreSeconds = 0; protected final AppSchedulingInfo appSchedulingInfo; - + protected ApplicationAttemptId attemptId; protected Map liveContainers = new HashMap(); protected final Map> reservedContainers = @@ -132,6 +133,7 @@ public SchedulerApplicationAttempt(ApplicationAttemptId applicationAttemptId, activeUsersManager, rmContext.getEpoch()); this.queue = queue; this.pendingRelease = new HashSet(); + this.attemptId = applicationAttemptId; if (rmContext.getRMApps() != null && rmContext.getRMApps() .containsKey(applicationAttemptId.getApplicationId())) { @@ -619,4 +621,15 @@ public synchronized void recoverContainer(RMContainer rmContainer) { // schedulingOpportunities // lastScheduledContainer } + + public void incNumAllocatedContainers(NodeType containerType, + NodeType requestType) { + RMAppAttempt attempt = + rmContext.getRMApps().get(attemptId.getApplicationId()) + .getCurrentAppAttempt(); + if (attempt != null) { + attempt.getRMAppAttemptMetrics().incNumAllocatedContainers(containerType, + requestType); + } + } } diff --git 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 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 index 38d4712..d73ac0f 100644 --- 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 +++ 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 @@ -32,6 +32,7 @@ import java.util.TreeSet; import org.apache.commons.lang.StringUtils; +import org.apache.commons.lang.mutable.MutableObject; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience.Private; @@ -1244,15 +1245,25 @@ private CSAssignment assignContainersOnNode(Resource clusterResource, RMContainer reservedContainer, boolean needToUnreserve) { Resource assigned = Resources.none(); + NodeType requestType = null; + MutableObject allocatedContainer = new MutableObject(); // Data-local ResourceRequest nodeLocalResourceRequest = application.getResourceRequest(priority, node.getNodeName()); if (nodeLocalResourceRequest != null) { + requestType = NodeType.NODE_LOCAL; assigned = assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest, - node, application, priority, reservedContainer, needToUnreserve); + node, application, priority, reservedContainer, needToUnreserve, + allocatedContainer); if (Resources.greaterThan(resourceCalculator, clusterResource, assigned, Resources.none())) { + + //update locality statistics + if (allocatedContainer.getValue() != null) { + application.incNumAllocatedContainers(NodeType.NODE_LOCAL, + requestType); + } return new CSAssignment(assigned, NodeType.NODE_LOCAL); } } @@ -1264,12 +1275,23 @@ private CSAssignment assignContainersOnNode(Resource clusterResource, if (!rackLocalResourceRequest.getRelaxLocality()) { return SKIP_ASSIGNMENT; } - + + if (requestType != NodeType.NODE_LOCAL) { + requestType = NodeType.RACK_LOCAL; + } + assigned = assignRackLocalContainers(clusterResource, rackLocalResourceRequest, - node, application, priority, reservedContainer, needToUnreserve); + node, application, priority, reservedContainer, needToUnreserve, + allocatedContainer); if (Resources.greaterThan(resourceCalculator, clusterResource, assigned, Resources.none())) { + + //update locality statistics + if (allocatedContainer.getValue() != null) { + application.incNumAllocatedContainers(NodeType.RACK_LOCAL, + requestType); + } return new CSAssignment(assigned, NodeType.RACK_LOCAL); } } @@ -1281,11 +1303,21 @@ private CSAssignment assignContainersOnNode(Resource clusterResource, if (!offSwitchResourceRequest.getRelaxLocality()) { return SKIP_ASSIGNMENT; } + if (requestType != NodeType.NODE_LOCAL + && requestType != NodeType.RACK_LOCAL) { + requestType = NodeType.OFF_SWITCH; + } - return new CSAssignment( + assigned = assignOffSwitchContainers(clusterResource, offSwitchResourceRequest, - node, application, priority, reservedContainer, needToUnreserve), - NodeType.OFF_SWITCH); + node, application, priority, reservedContainer, needToUnreserve, + allocatedContainer); + + // update locality statistics + if (allocatedContainer.getValue() != null) { + application.incNumAllocatedContainers(NodeType.OFF_SWITCH, requestType); + } + return new CSAssignment(assigned, NodeType.OFF_SWITCH); } return SKIP_ASSIGNMENT; @@ -1372,12 +1404,13 @@ protected boolean checkLimitsToReserve(Resource clusterResource, private Resource assignNodeLocalContainers(Resource clusterResource, ResourceRequest nodeLocalResourceRequest, FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority, - RMContainer reservedContainer, boolean needToUnreserve) { + RMContainer reservedContainer, boolean needToUnreserve, + MutableObject allocatedContainer) { if (canAssign(application, priority, node, NodeType.NODE_LOCAL, reservedContainer)) { return assignContainer(clusterResource, node, application, priority, nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer, - needToUnreserve); + needToUnreserve, allocatedContainer); } return Resources.none(); @@ -1386,12 +1419,13 @@ private Resource assignNodeLocalContainers(Resource clusterResource, private Resource assignRackLocalContainers( Resource clusterResource, ResourceRequest rackLocalResourceRequest, FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority, - RMContainer reservedContainer, boolean needToUnreserve) { + RMContainer reservedContainer, boolean needToUnreserve, + MutableObject allocatedContainer) { if (canAssign(application, priority, node, NodeType.RACK_LOCAL, reservedContainer)) { return assignContainer(clusterResource, node, application, priority, rackLocalResourceRequest, NodeType.RACK_LOCAL, reservedContainer, - needToUnreserve); + needToUnreserve, allocatedContainer); } return Resources.none(); @@ -1400,12 +1434,13 @@ private Resource assignRackLocalContainers( private Resource assignOffSwitchContainers( Resource clusterResource, ResourceRequest offSwitchResourceRequest, FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority, - RMContainer reservedContainer, boolean needToUnreserve) { + RMContainer reservedContainer, boolean needToUnreserve, + MutableObject allocatedContainer) { if (canAssign(application, priority, node, NodeType.OFF_SWITCH, reservedContainer)) { return assignContainer(clusterResource, node, application, priority, offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer, - needToUnreserve); + needToUnreserve, allocatedContainer); } return Resources.none(); @@ -1489,7 +1524,7 @@ Container createContainer(FiCaSchedulerApp application, FiCaSchedulerNode node, private Resource assignContainer(Resource clusterResource, FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority, ResourceRequest request, NodeType type, RMContainer rmContainer, - boolean needToUnreserve) { + boolean needToUnreserve, MutableObject createdContainer) { if (LOG.isDebugEnabled()) { LOG.debug("assignContainers: node=" + node.getNodeName() + " application=" + application.getApplicationId() @@ -1594,7 +1629,7 @@ private Resource assignContainer(Resource clusterResource, FiCaSchedulerNode nod " container=" + container + " queue=" + this + " clusterResource=" + clusterResource); - + createdContainer.setValue(allocatedContainer); return container.getResource(); } else { // if we are allowed to allocate but this node doesn't have space, reserve it or diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/AppBlock.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/AppBlock.java index 62ad8df..b0ebdd3 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/AppBlock.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/AppBlock.java @@ -204,18 +204,55 @@ protected void render(Block html) { table._(); div._(); + createContainerLocalityTable(html, attemptMetrics); createResourceRequestsTable(html, app); } + private void createContainerLocalityTable(Block html, + RMAppAttemptMetrics attemptMetrics) { + if (attemptMetrics == null) { + return; + } + + DIV div = html.div(_INFO_WRAP); + TABLE> table = + div.h3( + "Total Allocated Containers: " + + attemptMetrics.getTotalAllocatedContainers()).h3("Each table cell" + + " represents the number of (NodeLocal/RackLocal/OffSwitch) containers" + + " satisfied by (NodeLocal/RackLocal/OffSwitch) resource requests.").table( + "#containerLocality"); + table. + tr(). + th(_TH, ""). + th(_TH, "Node Local Request"). + th(_TH, "Rack Local Request"). + th(_TH, "Off Switch Request"). + _(); + + String[] containersType = + { "Num Node Local Containers (satisfied by)", "Num Rack Local Containers (satisfied by)", + "Num Off Switch Containers (satisfied by)" }; + boolean odd = false; + for (int i = 0; i < attemptMetrics.getLocalityStatistics().length; i++) { + table.tr((odd = !odd) ? _ODD : _EVEN).td(containersType[i]) + .td(String.valueOf(attemptMetrics.getLocalityStatistics()[i][0])) + .td(i == 0 ? "" : String.valueOf(attemptMetrics.getLocalityStatistics()[i][1])) + .td(i <= 1 ? "" : String.valueOf(attemptMetrics.getLocalityStatistics()[i][2]))._(); + } + table._(); + div._(); + } + private void createResourceRequestsTable(Block html, AppInfo app) { TBODY> tbody = html.table("#ResourceRequests").thead().tr() .th(".priority", "Priority") - .th(".resourceName", "ResourceName") + .th(".resourceName", "Resource Name") .th(".totalResource", "Capability") - .th(".numContainers", "NumContainers") - .th(".relaxLocality", "RelaxLocality") - .th(".nodeLabelExpression", "NodeLabelExpression")._()._().tbody(); + .th(".numContainers", "Num Containers") + .th(".relaxLocality", "Relax Locality") + .th(".nodeLabelExpression", "Node Label Expression")._()._().tbody(); Resource totalResource = Resource.newInstance(0, 0); if (app.getResourceRequests() != null) { diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java index 985609e..1aff74f 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java @@ -52,6 +52,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMContext; import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter; import org.apache.hadoop.yarn.server.resourcemanager.metrics.SystemMetricsPublisher; +import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType; @@ -217,6 +218,7 @@ public void testReservation() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0); @@ -366,6 +368,7 @@ public void testReservationNoContinueLook() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0); @@ -511,6 +514,7 @@ public void testAssignContainersNeedToUnreserve() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0); @@ -747,6 +751,7 @@ public void testAssignToQueue() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0); @@ -921,7 +926,7 @@ public void testAssignToUser() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); - + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0); final ApplicationAttemptId appAttemptId_1 = TestUtils @@ -1047,6 +1052,7 @@ public void testReservationsNoneAvailable() throws Exception { .getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, a, mock(ActiveUsersManager.class), spyRMContext); + rmContext.getRMApps().put(app_0.getApplicationId(), mock(RMApp.class)); a.submitApplicationAttempt(app_0, user_0);