From 455fcdbca931e7c6cf35ddc645fb01b616c4336d Mon Sep 17 00:00:00 2001 From: Adam Antal Date: Wed, 27 Nov 2019 15:45:51 +0100 Subject: [PATCH] YARN-3890. FairScheduler should show the scheduler health metrics similar to ones added in CapacityScheduler --- .../scheduler/AbstractYarnScheduler.java | 5 + .../scheduler/fair/FSAppAttempt.java | 23 ++- .../scheduler/fair/FSPreemptionThread.java | 6 + .../scheduler/fair/FairScheduler.java | 9 +- .../webapp/FairSchedulerPage.java | 138 +++++++++++++++++- .../scheduler/fair/TestFSAppAttempt.java | 73 ++++++--- 6 files changed, 230 insertions(+), 24 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java index d7426833ac8..c61c34233eb 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java @@ -359,6 +359,11 @@ public SchedulerHealth getSchedulerHealth() { return this.schedulerHealth; } + @VisibleForTesting + public void setSchedulerHealth(SchedulerHealth schedulerHealth) { + this.schedulerHealth = schedulerHealth; + } + protected void setLastNodeUpdateTime(long time) { this.lastNodeUpdateTime = time; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java index 62c1142a5c3..8bcca9fb618 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java @@ -178,6 +178,11 @@ void containerCompleted(RMContainer rmContainer, this.attemptResourceUsage.decUsed(containerResource); getQueue().decUsedResource(containerResource); + scheduler.getSchedulerHealth().updateRelease( + scheduler.getClock().getTime(), rmContainer.getNodeId(), + rmContainer.getContainerId(), rmContainer.getQueueName()); + scheduler.getSchedulerHealth().updateSchedulerReleaseCounts(1); + // Clear resource utilization metrics cache. lastMemoryAggregateAllocationUpdateTime = -1; } finally { @@ -469,6 +474,12 @@ public RMContainer allocate(NodeType type, FSSchedulerNode node, // Update resource requests related to "request" and store in RMContainer ((RMContainerImpl) rmContainer).setContainerRequest(containerRequest); + // Update scheduler health metrics for container allocation + scheduler.getSchedulerHealth().updateAllocation( + scheduler.getClock().getTime(), rmContainer.getNodeId(), + rmContainer.getContainerId(), queue.getQueueName()); + scheduler.getSchedulerHealth().updateSchedulerAllocationCounts(1); + // Inform the container rmContainer.handle( new RMContainerEvent(container.getId(), RMContainerEventType.START)); @@ -697,8 +708,8 @@ public synchronized void recoverContainer(SchedulerNode node, * in {@link FSSchedulerNode}.. * return whether reservation was possible with the current threshold limits */ - private boolean reserve(Resource perAllocationResource, FSSchedulerNode node, - Container reservedContainer, NodeType type, + protected boolean reserve(Resource perAllocationResource, + FSSchedulerNode node, Container reservedContainer, NodeType type, SchedulerRequestKey schedulerKey) { RMContainer nodeReservedContainer = node.getReservedContainer(); @@ -718,11 +729,19 @@ private boolean reserve(Resource perAllocationResource, FSSchedulerNode node, super.reserve(node, schedulerKey, null, reservedContainer); node.reserveResource(this, schedulerKey, rmContainer); setReservation(node); + scheduler.getSchedulerHealth().updateReservation( + scheduler.getClock().getTime(), node.getNodeID(), + reservedContainer.getId(), getQueue().getName()); + scheduler.getSchedulerHealth().updateSchedulerReservationCounts(1); } else { RMContainer rmContainer = node.getReservedContainer(); super.reserve(node, schedulerKey, rmContainer, reservedContainer); node.reserveResource(this, schedulerKey, rmContainer); setReservation(node); + scheduler.getSchedulerHealth().updateReservation( + scheduler.getClock().getTime(), node.getNodeID(), + reservedContainer.getId(), getQueue().getName()); + scheduler.getSchedulerHealth().updateSchedulerReservationCounts(1); } return true; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSPreemptionThread.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSPreemptionThread.java index 221bb17ae5b..c4263a00b2e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSPreemptionThread.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSPreemptionThread.java @@ -284,6 +284,12 @@ public void run() { ContainerStatus status = SchedulerUtils.createPreemptedContainerStatus( container.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER); + // Updates scheduler health metrics for preemption + scheduler.getSchedulerHealth().updatePreemption( + scheduler.getClock().getTime(), container.getNodeId(), + container.getContainerId(), container.getQueueName()); + scheduler.getSchedulerHealth().updateSchedulerPreemptionCounts(1); + LOG.info("Killing container " + container); scheduler.completedContainer( container, status, RMContainerEventType.KILL); diff --git a/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 b/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 index 1097f0ecdc3..da39d74a8e4 100644 --- a/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 +++ b/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 @@ -733,6 +733,8 @@ protected void completedContainerInternal( + " released container " + container.getId() + " on node: " + (node == null ? nodeID : node) + " with event: " + event); } + schedulerHealth.updateRelease(getClock().getTime(), nodeID, + container.getId(), application.getQueue().getName()); } finally { writeLock.unlock(); } @@ -1117,6 +1119,7 @@ void attemptScheduling(FSSchedulerNode node) { boolean validReservation = false; if (reservedAppSchedulable != null) { validReservation = reservedAppSchedulable.assignReservedContainer(node); + schedulerHealth.updateSchedulerFulfilledReservationCounts(1); } if (!validReservation) { // No reservation, schedule at queue which is farthest below fair share @@ -1140,8 +1143,12 @@ void attemptScheduling(FSSchedulerNode node) { break; } } + + // TODO: Reserved resource + schedulerHealth.updateSchedulerRunDetails(getClock().getTime(), + assignedResource, Resources.none()); + updateRootQueueMetrics(); } - updateRootQueueMetrics(); } finally { writeLock.unlock(); } 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/FairSchedulerPage.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/FairSchedulerPage.java index 7f31defa066..46e0d778f13 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/FairSchedulerPage.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/FairSchedulerPage.java @@ -21,14 +21,18 @@ import static org.apache.hadoop.yarn.util.StringHelper.join; import java.util.Collection; +import java.util.HashMap; +import java.util.Map; import org.apache.hadoop.util.StringUtils; import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerHealth; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.FairSchedulerInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.FairSchedulerLeafQueueInfo; import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.FairSchedulerQueueInfo; import org.apache.hadoop.yarn.server.webapp.WebPageUtils; +import org.apache.hadoop.yarn.util.Times; import org.apache.hadoop.yarn.webapp.ResponseInfo; import org.apache.hadoop.yarn.webapp.SubView; import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet; @@ -171,6 +175,117 @@ public void render(Block html) { ul.__(); } } + + /** + * Scheduler health block + */ + public static class HealthBlock extends HtmlBlock { + + private final FairScheduler fs; + + @Inject + HealthBlock(ResourceManager rm) { + fs = (FairScheduler) rm.getResourceScheduler(); + } + + @Override + public void render(HtmlBlock.Block html) { + SchedulerHealth healthInfo = fs.getSchedulerHealth(); + UL>> ul = html. + div("#fs-health-wrapper.ui-widget"). + div(".ui-widget-header.ui-corner-top"). + __("Fair scheduler health metrics ▼").__(). + div("#fs-health.ui-widget-content.ui-corner-bottom"). + ul(); + + DIV div = html.div("#health"); + div.h4("Aggregate scheduler counts"); + Hamlet.TBODY>> tbody = + div.table("#lastrun").thead().$class("ui-widget-header") + .tr().th() + .$class("ui-state-default") + .__("Total Container Allocations" + + "(count)") + .__().th().$class("ui-state-default") + .__("Total Container Releases(count)").__().th() + .$class("ui-state-default") + .__("Total Fulfilled Reservations(count)").__().th() + .$class("ui-state-default") + .__("Total Container Preemptions(count)") + .__().__().__().tbody(); + tbody + .$class("ui-widget-content") + .tr() + .td( + String.valueOf(fs.getRootQueueMetrics() + .getAggregateAllocatedContainers())) + .td( + String.valueOf(fs.getRootQueueMetrics() + .getAggegatedReleasedContainers())) + .td(healthInfo.getAggregateFulFilledReservationsCount().toString()) + .td(healthInfo.getAggregatePreemptionCount().toString()).__().__().__(); + div.h4("Last scheduler run"); + tbody = + div.table("#lastrun").thead().$class("ui-widget-header").tr() + .th() + .$class("ui-state-default").__("Time").__().th() + .$class("ui-state-default").__("Allocations(count - resources)") + .__() + .th().$class("ui-state-default") + .__().th().$class("ui-state-default") + .__("Releases(count - resources)") + .__().__().__().tbody(); + tbody + .$class("ui-widget-content") + .tr() + .td(Times.format(healthInfo.getLastSchedulerRunTime())) + .td( + healthInfo.getAllocationCount().toString() + " - " + + healthInfo.getResourcesAllocated().toString()) + .td( + healthInfo.getReservationCount().toString() + " - " + + healthInfo.getResourcesReserved().toString()) + .td( + healthInfo.getReleaseCount().toString() + " - " + + healthInfo.getResourcesReleased().toString()) + .__().__().__(); + Map info = new HashMap<>(); + info.put("Allocation", healthInfo.getLastAllocationDetails()); + info.put("Reservation", healthInfo.getLastReservationDetails()); + info.put("Release", healthInfo.getLastReleaseDetails()); + info.put("Preemption", healthInfo.getLastPreemptionDetails()); + + for (Map.Entry entry : info + .entrySet()) { + String containerId = "N/A"; + String nodeId = "N/A"; + String queue = "N/A"; + String table = "#" + entry.getKey(); + div.h4("Last " + entry.getKey()); + tbody = + div.table(table).thead().$class("ui-widget-header").tr().th() + .$class("ui-state-default").__("Time").__().th() + .$class("ui-state-default").__("Container Id").__().th() + .$class("ui-state-default").__("Node Id").__().th() + .$class("ui-state-default").__("Queue").__().__().__().tbody(); + SchedulerHealth.DetailedInformation di = entry.getValue(); + if (di.getTimestamp() != 0) { + if (di.getContainerId() != null) { + containerId = di.getContainerId().toString(); + } + if (di.getNodeId() != null) { + nodeId = di.getNodeId().toString(); + } + queue = di.getQueue(); + } + tbody.$class("ui-widget-content").tr() + .td(Times.format(di.getTimestamp())).td(containerId).td(nodeId) + .td(queue).__().__().__(); + } + div.__(); + ul.__().__().__(); + } + } static class QueuesBlock extends HtmlBlock { final FairScheduler fs; @@ -184,6 +299,7 @@ public void render(Block html) { @Override public void render(Block html) { html.__(MetricsOverviewTable.class); + html.__(HealthBlock.class); UL>> ul = html. div("#cs-wrapper.ui-widget"). div(".ui-widget-header.ui-corner-top"). @@ -238,17 +354,25 @@ public void render(Block html) { @Override protected void postHead(Page.HTML<__> html) { html. style().$type("text/css"). - __("#cs { padding: 0.5em 0 1em 0; margin-bottom: 1em; position: relative }", + __( + "#cs { padding: 0.5em 0 1em 0; margin-bottom: 1em; " + + "position: relative }", "#cs ul { list-style: none }", "#cs a { font-weight: normal; margin: 2px; position: relative }", "#cs a span { font-weight: normal; font-size: 80% }", "#cs-wrapper .ui-widget-header { padding: 0.2em 0.5em }", + "#fs-health-wrapper {margin-bottom: 1em }", + "#fs-health {padding: 0.2em }", + "#fs-health-wrapper .ui-widget-header { padding: 0.2em 0.5em; " + + "cursor: pointer; }", ".qstats { font-weight: normal; font-size: 80%; position: absolute }", ".qlegend { font-weight: normal; padding: 0 1em; margin: 1em }", "table.info tr th {width: 50%}").__(). // to center info table script("/static/jt/jquery.jstree.js"). script().$type("text/javascript"). __("$(function() {", + " $('#cs a span').addClass('ui-corner-all').css('position', " + + "'absolute');", " $('#cs a span').addClass('ui-corner-all').css('position', 'absolute');", " $('#cs').bind('loaded.jstree', function (e, data) {", " var callback = { call:reopenQueryNodes }", @@ -268,6 +392,18 @@ public void render(Block html) { " $('#apps').dataTable().fnFilter(q, 4, true);", " });", " $('#cs').show();", + " if('health' != window.location.hash.substring(1)) { " + + "$('#fs-health').hide();}", + " $(\"#fs-health-wrapper > div.ui-widget-header.ui-corner-top\")" + + ".click(", + " function() {", + " $(\"#fs-health\").slideToggle('slow',", + " function() {", + " window.location.hash = ($(\"#fs-health\").is", + "(\":visible\"))?\"health\":\"\";", + " });", + " }", + " ); ", "});").__(). __(SchedulerPageUtil.QueueBlockUtil.class); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java index bda04187ef1..29e61c7e2d3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java @@ -25,15 +25,19 @@ import java.util.concurrent.ConcurrentMap; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.yarn.api.records.NodeId; import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.server.resourcemanager.MockNodes; import org.apache.hadoop.yarn.server.resourcemanager.MockRM; + +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationId; @@ -46,6 +50,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerHealth; import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity @@ -76,7 +81,7 @@ public void testDelayScheduling() { FSLeafQueue queue = Mockito.mock(FSLeafQueue.class); Priority pri = Mockito.mock(Priority.class); SchedulerRequestKey prio = TestUtils.toSchedulerKey(pri); - Mockito.when(pri.getPriority()).thenReturn(1); + when(pri.getPriority()).thenReturn(1); double nodeLocalityThreshold = .5; double rackLocalityThreshold = .6; @@ -136,7 +141,7 @@ public void testDelaySchedulingForContinuousScheduling() FSLeafQueue queue = scheduler.getQueueManager().getLeafQueue("queue", true); Priority pri = Mockito.mock(Priority.class); SchedulerRequestKey prio = TestUtils.toSchedulerKey(pri); - Mockito.when(pri.getPriority()).thenReturn(1); + when(pri.getPriority()).thenReturn(1); ControlledClock clock = new ControlledClock(); scheduler.setClock(clock); @@ -195,7 +200,7 @@ public void testLocalityLevelWithoutDelays() { FSLeafQueue queue = Mockito.mock(FSLeafQueue.class); Priority pri = Mockito.mock(Priority.class); SchedulerRequestKey prio = TestUtils.toSchedulerKey(pri); - Mockito.when(pri.getPriority()).thenReturn(1); + when(pri.getPriority()).thenReturn(1); RMContext rmContext = resourceManager.getRMContext(); ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1); @@ -209,7 +214,7 @@ public void testLocalityLevelWithoutDelays() { @Test public void testHeadroom() { final FairScheduler mockScheduler = Mockito.mock(FairScheduler.class); - Mockito.when(mockScheduler.getClock()).thenReturn(scheduler.getClock()); + when(mockScheduler.getClock()).thenReturn(scheduler.getClock()); final FSLeafQueue mockQueue = Mockito.mock(FSLeafQueue.class); @@ -229,16 +234,16 @@ public void testHeadroom() { final QueueMetrics fakeRootQueueMetrics = Mockito.mock(QueueMetrics.class); - Mockito.when(mockQueue.getMaxShare()).thenReturn(queueMaxResources); - Mockito.when(mockQueue.getFairShare()).thenReturn(queueFairShare); - Mockito.when(mockQueue.getResourceUsage()).thenReturn(queueUsage); - Mockito.when(mockScheduler.getClusterResource()).thenReturn + when(mockQueue.getMaxShare()).thenReturn(queueMaxResources); + when(mockQueue.getFairShare()).thenReturn(queueFairShare); + when(mockQueue.getResourceUsage()).thenReturn(queueUsage); + when(mockScheduler.getClusterResource()).thenReturn (clusterResource); - Mockito.when(fakeRootQueueMetrics.getAllocatedResources()).thenReturn + when(fakeRootQueueMetrics.getAllocatedResources()).thenReturn (clusterUsage); - Mockito.when(mockScheduler.getRootQueueMetrics()).thenReturn + when(mockScheduler.getRootQueueMetrics()).thenReturn (fakeRootQueueMetrics); - Mockito.when(mockScheduler.getConf()).thenReturn + when(mockScheduler.getConf()).thenReturn (Mockito.mock(FairSchedulerConfiguration.class)); ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1); @@ -249,7 +254,7 @@ public void testHeadroom() { // Min of Memory and CPU across cluster and queue is used in // DominantResourceFairnessPolicy - Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy + when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy .getInstance(DominantResourceFairnessPolicy.class)); verifyHeadroom(schedulerApp, min(queueStarvation.getMemorySize(), @@ -261,7 +266,7 @@ public void testHeadroom() { ); // Fair and Fifo ignore CPU of queue, so use cluster available CPU - Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy + when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy .getInstance(FairSharePolicy.class)); verifyHeadroom(schedulerApp, min(queueStarvation.getMemorySize(), @@ -272,7 +277,7 @@ public void testHeadroom() { queueMaxResourcesAvailable.getVirtualCores()) ); - Mockito.when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy + when(mockQueue.getPolicy()).thenReturn(SchedulingPolicy .getInstance(FifoPolicy.class)); verifyHeadroom(schedulerApp, min(queueStarvation.getMemorySize(), @@ -367,15 +372,15 @@ public void testNoNextPendingAsk() { rmApps.put(applicationAttemptId.getApplicationId(), rmApp); ApplicationSubmissionContext appContext = Mockito.mock(ApplicationSubmissionContext.class); - Mockito.when(appContext.getUnmanagedAM()).thenReturn(false); - Mockito.when(appContext.getLogAggregationContext()) + when(appContext.getUnmanagedAM()).thenReturn(false); + when(appContext.getLogAggregationContext()) .thenReturn(Mockito.mock(LogAggregationContext.class)); - Mockito.when(rmApp.getApplicationSchedulingEnvs()) + when(rmApp.getApplicationSchedulingEnvs()) .thenReturn(new HashMap<>()); - Mockito.when(rmApp.getApplicationSubmissionContext()) + when(rmApp.getApplicationSubmissionContext()) .thenReturn(appContext); - Mockito.when(rmContext.getRMApps()).thenReturn(rmApps); - Mockito.when(rmContext.getYarnConfiguration()).thenReturn(conf); + when(rmContext.getRMApps()).thenReturn(rmApps); + when(rmContext.getYarnConfiguration()).thenReturn(conf); FSAppAttempt schedulerApp = new FSAppAttempt(scheduler, applicationAttemptId, "user1", queue, null, rmContext); @@ -387,6 +392,34 @@ public void testNoNextPendingAsk() { assertEquals(Resources.none(), resource); } + @Test + public void testHealthMetric() { + //SETUP + FSLeafQueue queue = Mockito.mock(FSLeafQueue.class); + FSQueueMetrics queueMetrics = Mockito.mock(FSQueueMetrics.class); + when(queue.getMetrics()).thenReturn(queueMetrics); + + ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1); + RMContext rmContext = resourceManager.getRMContext(); + FSAppAttempt schedulerApp = + new FSAppAttempt(scheduler, applicationAttemptId, "user1", queue, + null, rmContext); + scheduler.setSchedulerHealth(new SchedulerHealth()); + Resource perAllocationResource = Resource.newInstance(1024, 1); + + FSSchedulerNode node = Mockito.mock(FSSchedulerNode.class); + RMNode rmNode = Mockito.mock(RMNode.class); + when(rmNode.getNodeID()).thenReturn(NodeId.newInstance("host", + 1234)); + when(node.getRMNode()).thenReturn(rmNode); + SchedulerRequestKey schedulerKey = Mockito.mock(SchedulerRequestKey.class); + + schedulerApp.reserve(perAllocationResource, node, null, + NodeType.NODE_LOCAL, schedulerKey); + + assertThat(scheduler.getSchedulerHealth().getReservationCount()).isOne(); +} + private static long min(long value1, long value2, long value3) { return Math.min(Math.min(value1, value2), value3); } -- 2.21.0