ResourceTrackerService extends AbstractService implements
    ResourceTracker { @SuppressWarnings("unchecked") @Override public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request) throws YarnException, IOException { NodeStatus remoteNodeStatus = request.getNodeStatus(); NodeId nodeId = remoteNodeStatus.getNodeId(); if (!this.nodesListManager.isValidNode(nodeId.getHost()) && !isNodeInDecommissioning(nodeId)) { String message = "Disallowed NodeManager nodeId: " + nodeId + " hostname: " + nodeId.getHost(); LOG.info(message); return YarnServerBuilderUtils.newNodeHeartbeatResponse( NodeAction.SHUTDOWN, message); } RMNode rmNode = this.rmContext.getRMNodes().get(nodeId); if (rmNode == null) { String message = "Node not found resyncing " + remoteNodeStatus.getNodeId(); LOG.info(message); return YarnServerBuilderUtils.newNodeHeartbeatResponse(NodeAction.RESYNC, message); } this.nmLivelinessMonitor.receivedPing(nodeId); this.decommissioningWatcher.update(rmNode, remoteNodeStatus); NodeHeartbeatResponse lastNodeHeartbeatResponse = rmNode.getLastNodeHeartBeatResponse(); if (getNextResponseId( remoteNodeStatus.getResponseId()) == lastNodeHeartbeatResponse .getResponseId()) { LOG.info("Received duplicate heartbeat from node " + rmNode.getNodeAddress()+ " responseId=" + remoteNodeStatus.getResponseId()); return lastNodeHeartbeatResponse; } else if (remoteNodeStatus.getResponseId() != lastNodeHeartbeatResponse .getResponseId()) { String message = "Too far behind rm response id:" + lastNodeHeartbeatResponse.getResponseId() + " nm response id:" + remoteNodeStatus.getResponseId(); LOG.info(message); this.rmContext.getDispatcher().getEventHandler().handle( new RMNodeEvent(nodeId, RMNodeEventType.REBOOTING)); return YarnServerBuilderUtils.newNodeHeartbeatResponse(NodeAction.RESYNC, message); } if (rmNode.getState() == NodeState.DECOMMISSIONING && decommissioningWatcher.checkReadyToBeDecommissioned( rmNode.getNodeID())) { String message = "DECOMMISSIONING " + nodeId + " is ready to be decommissioned"; LOG.info(message); this.rmContext.getDispatcher().getEventHandler().handle( new RMNodeEvent(nodeId, RMNodeEventType.DECOMMISSION)); this.nmLivelinessMonitor.unregister(nodeId); return YarnServerBuilderUtils.newNodeHeartbeatResponse( NodeAction.SHUTDOWN, message); } if (timelineServiceV2Enabled) { updateAppCollectorsMap(request); } NodeHeartbeatResponse nodeHeartBeatResponse = YarnServerBuilderUtils.newNodeHeartbeatResponse( getNextResponseId(lastNodeHeartbeatResponse.getResponseId()), NodeAction.NORMAL, null, null, null, null, nextHeartBeatInterval); rmNode.setAndUpdateNodeHeartbeatResponse(nodeHeartBeatResponse); populateKeys(request, nodeHeartBeatResponse); populateTokenSequenceNo(request, nodeHeartBeatResponse); if (timelineServiceV2Enabled) { setAppCollectorsMapToResponse(rmNode.getRunningApps(), nodeHeartBeatResponse); } RMNodeStatusEvent nodeStatusEvent = new RMNodeStatusEvent(nodeId, remoteNodeStatus); if (request.getLogAggregationReportsForApps() != null && !request.getLogAggregationReportsForApps().isEmpty()) { nodeStatusEvent.setLogAggregationReportsForApps(request .getLogAggregationReportsForApps()); } this.rmContext.getDispatcher().getEventHandler().handle(nodeStatusEvent); if (isDistributedNodeLabelsConf && request.getNodeLabels() != null) { try { updateNodeLabelsFromNMReport( NodeLabelsUtils.convertToStringSet(request.getNodeLabels()), nodeId); nodeHeartBeatResponse.setAreNodeLabelsAcceptedByRM(true); } catch (IOException ex) { nodeHeartBeatResponse.setDiagnosticsMessage(ex.getMessage()); nodeHeartBeatResponse.setAreNodeLabelsAcceptedByRM(false); } } String nid = nodeId.toString(); Resource capability = loadNodeResourceFromDRConfiguration(nid); if (capability != null) { nodeHeartBeatResponse.setResource(capability); } if (rmNode.isUpdatedCapability()) { nodeHeartBeatResponse.setResource(rmNode.getTotalCapability()); rmNode.resetUpdatedCapability(); } if (this.rmContext.getNodeManagerQueueLimitCalculator() != null) { nodeHeartBeatResponse.setContainerQueuingLimit( this.rmContext.getNodeManagerQueueLimitCalculator() .createContainerQueuingLimit()); } if (request.getNodeAttributes() != null) { try { updateNodeAttributesIfNecessary(nodeId, request.getNodeAttributes()); nodeHeartBeatResponse.setAreNodeAttributesAcceptedByRM(true); } catch (IOException ex) { String errorMsg = nodeHeartBeatResponse.getDiagnosticsMessage() == null ? ex.getMessage() : nodeHeartBeatResponse.getDiagnosticsMessage() + "\n" + ex .getMessage(); nodeHeartBeatResponse.setDiagnosticsMessage(errorMsg); nodeHeartBeatResponse.setAreNodeAttributesAcceptedByRM(false); } } return nodeHeartBeatResponse; } ResourceTrackerService(RMContext rmContext,
      NodesListManager nodesListManager,
      NMLivelinessMonitor nmLivelinessMonitor,
      RMContainerTokenSecretManager containerTokenSecretManager,
      NMTokenSecretManagerInRM nmTokenSecretManager); void loadDynamicResourceConfiguration(Configuration conf); void updateDynamicResourceConfiguration(
      DynamicResourceConfiguration conf); @SuppressWarnings("unchecked") @Override RegisterNodeManagerResponse registerNodeManager(
      RegisterNodeManagerRequest request); @SuppressWarnings("unchecked") @Override NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request); @SuppressWarnings("unchecked") @Override UnRegisterNodeManagerResponse unRegisterNodeManager(
      UnRegisterNodeManagerRequest request); static Node resolve(String hostName); @VisibleForTesting Server getServer();  }
@Test (timeout = 50000) public void testGetNextHeartBeatInterval() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NM_HEARTBEAT_INTERVAL_MS, "4000"); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals(4000, nodeHeartbeat.getNextHeartBeatInterval()); NodeHeartbeatResponse nodeHeartbeat2 = nm2.nodeHeartbeat(true); Assert.assertEquals(4000, nodeHeartbeat2.getNextHeartBeatInterval()); }
@Test public void testDecommissionWithIncludeHosts() throws Exception { writeToHostsFile("localhost", "host1", "host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); MockNM nm3 = rm.registerNode("localhost:4433", 1024); ClusterMetrics metrics = ClusterMetrics.getMetrics(); assert(metrics != null); int metricCount = metrics.getNumDecommisionedNMs(); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm3.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); String ip = NetUtils.normalizeHostName("localhost"); writeToHostsFile("host1", ip); rm.getNodesListManager().refreshNodes(conf); checkShutdownNMCount(rm, ++metricCount); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); Assert .assertEquals(1, ClusterMetrics.getMetrics().getNumShutdownNMs()); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertTrue("Node is not decommisioned.", NodeAction.SHUTDOWN .equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm3.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); Assert.assertEquals(metricCount, ClusterMetrics.getMetrics() .getNumShutdownNMs()); rm.stop(); }
@Test public void testDecommissionWithExcludeHosts() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile .getAbsolutePath()); writeToHostsFile(""); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); MockNM nm3 = rm.registerNode("localhost:4433", 1024); int metricCount = ClusterMetrics.getMetrics().getNumDecommisionedNMs(); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); rm.drainEvents(); String ip = NetUtils.normalizeHostName("localhost"); writeToHostsFile("host2", ip); rm.getNodesListManager().refreshNodes(conf); checkDecommissionedNMCount(rm, metricCount + 2); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertTrue("The decommisioned metrics are not updated", NodeAction.SHUTDOWN.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm3.nodeHeartbeat(true); Assert.assertTrue("The decommisioned metrics are not updated", NodeAction.SHUTDOWN.equals(nodeHeartbeat.getNodeAction())); rm.drainEvents(); writeToHostsFile(""); rm.getNodesListManager().refreshNodes(conf); nm3 = rm.registerNode("localhost:4433", 1024); nodeHeartbeat = nm3.nodeHeartbeat(true); rm.drainEvents(); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); checkDecommissionedNMCount(rm, metricCount + 1); }
@Test public void testGracefulDecommissionNoApp() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile .getAbsolutePath()); writeToHostsFile(""); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); MockNM nm3 = rm.registerNode("host3:4433", 5120); int metricCount = ClusterMetrics.getMetrics().getNumDecommisionedNMs(); NodeHeartbeatResponse nodeHeartbeat1 = nm1.nodeHeartbeat(true); NodeHeartbeatResponse nodeHeartbeat2 = nm2.nodeHeartbeat(true); NodeHeartbeatResponse nodeHeartbeat3 = nm3.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat1.getNodeAction())); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat2.getNodeAction())); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat3.getNodeAction())); rm.waitForState(nm2.getNodeId(), NodeState.RUNNING); rm.waitForState(nm3.getNodeId(), NodeState.RUNNING); writeToHostsFile("host2", "host3"); rm.getNodesListManager().refreshNodes(conf, true); rm.waitForState(nm2.getNodeId(), NodeState.DECOMMISSIONING); rm.waitForState(nm3.getNodeId(), NodeState.DECOMMISSIONING); nodeHeartbeat1 = nm1.nodeHeartbeat(true); nodeHeartbeat2 = nm2.nodeHeartbeat(true); nodeHeartbeat3 = nm3.nodeHeartbeat(true); checkDecommissionedNMCount(rm, metricCount + 2); rm.waitForState(nm2.getNodeId(), NodeState.DECOMMISSIONED); rm.waitForState(nm3.getNodeId(), NodeState.DECOMMISSIONED); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat1.getNodeAction())); Assert.assertEquals(NodeAction.SHUTDOWN, nodeHeartbeat2.getNodeAction()); Assert.assertEquals(NodeAction.SHUTDOWN, nodeHeartbeat3.getNodeAction()); }
@Test public void testGracefulDecommissionDefaultTimeoutResolution() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, excludeHostXmlFile .getAbsolutePath()); writeToHostsXmlFile(excludeHostXmlFile, Pair.of("", null)); rm = new MockRM(conf); rm.start(); int nodeMemory = 1024; MockNM nm1 = rm.registerNode("host1:1234", nodeMemory); MockNM nm2 = rm.registerNode("host2:5678", nodeMemory); MockNM nm3 = rm.registerNode("host3:9101", nodeMemory); NodeHeartbeatResponse nodeHeartbeat1 = nm1.nodeHeartbeat(true); NodeHeartbeatResponse nodeHeartbeat2 = nm2.nodeHeartbeat(true); NodeHeartbeatResponse nodeHeartbeat3 = nm3.nodeHeartbeat(true); Assert.assertTrue( NodeAction.NORMAL.equals(nodeHeartbeat1.getNodeAction())); Assert.assertTrue( NodeAction.NORMAL.equals(nodeHeartbeat2.getNodeAction())); Assert.assertTrue( NodeAction.NORMAL.equals(nodeHeartbeat3.getNodeAction())); rm.waitForState(nm1.getNodeId(), NodeState.RUNNING); rm.waitForState(nm2.getNodeId(), NodeState.RUNNING); rm.waitForState(nm3.getNodeId(), NodeState.RUNNING); final Integer nm1DecommissionTimeout = 20; writeToHostsXmlFile( excludeHostXmlFile, Pair.of(nm1.getNodeId().getHost(), nm1DecommissionTimeout), Pair.of(nm2.getNodeId().getHost(), null)); rm.getNodesListManager().refreshNodes(conf, true); rm.waitForState(nm1.getNodeId(), NodeState.DECOMMISSIONING); rm.waitForState(nm2.getNodeId(), NodeState.DECOMMISSIONING); Assert.assertEquals( nm1DecommissionTimeout, rm.getDecommissioningTimeout(nm1.getNodeId())); Integer defaultDecTimeout = conf.getInt(YarnConfiguration.RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT, YarnConfiguration.DEFAULT_RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT); Assert.assertEquals( defaultDecTimeout, rm.getDecommissioningTimeout(nm2.getNodeId())); final Integer newDefaultDecTimeout = defaultDecTimeout + 10; writeToHostsXmlFile( excludeHostXmlFile, Pair.of(nm3.getNodeId().getHost(), null)); conf.setInt(YarnConfiguration.RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT, newDefaultDecTimeout); rm.getNodesListManager().refreshNodes(conf, true); rm.waitForState(nm3.getNodeId(), NodeState.DECOMMISSIONING); Assert.assertEquals( newDefaultDecTimeout, rm.getDecommissioningTimeout(nm3.getNodeId())); }
@Test public void testGracefulDecommissionWithApp() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile .getAbsolutePath()); writeToHostsFile(""); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 10240); MockNM nm2 = rm.registerNode("host2:5678", 20480); MockNM nm3 = rm.registerNode("host3:4433", 10240); NodeId id1 = nm1.getNodeId(); NodeId id3 = nm3.getNodeId(); rm.waitForState(id1, NodeState.RUNNING); rm.waitForState(id3, NodeState.RUNNING); RMApp app = MockRMAppSubmitter.submitWithMemory(2000, rm); MockAM am = MockRM.launchAndRegisterAM(app, rm, nm1); ApplicationAttemptId aaid = app.getCurrentAppAttempt().getAppAttemptId(); nm1.nodeHeartbeat(aaid, 2, ContainerState.RUNNING); nm3.nodeHeartbeat(true); writeToHostsFile("host1", "host3"); rm.getNodesListManager().refreshNodes(conf, true); rm.waitForState(id1, NodeState.DECOMMISSIONING); rm.waitForState(id3, NodeState.DECOMMISSIONING); nm1.nodeHeartbeat(true); nm3.nodeHeartbeat(true); rm.waitForState(id1, NodeState.DECOMMISSIONING); rm.waitForState(id3, NodeState.DECOMMISSIONED); nm1.nodeHeartbeat(aaid, 2, ContainerState.RUNNING); NodeHeartbeatResponse nodeHeartbeat1 = nm1.nodeHeartbeat(aaid, 2, ContainerState.COMPLETE); Assert.assertEquals(NodeAction.NORMAL, nodeHeartbeat1.getNodeAction()); MockRM.finishAMAndVerifyAppState(app, rm, nm1, am); rm.waitForState(app.getApplicationId(), RMAppState.FINISHED); nodeHeartbeat1 = nm1.nodeHeartbeat(aaid, 2, ContainerState.COMPLETE); Assert.assertEquals(NodeAction.SHUTDOWN, nodeHeartbeat1.getNodeAction()); rm.waitForState(id1, NodeState.DECOMMISSIONED); }
@Test public void testAddNewIncludePathToConfiguration() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); ClusterMetrics metrics = ClusterMetrics.getMetrics(); assert(metrics != null); int initialMetricCount = metrics.getNumShutdownNMs(); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals( NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertEquals( NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); writeToHostsFile("host1"); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm.getNodesListManager().refreshNodes(conf); checkShutdownNMCount(rm, ++initialMetricCount); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals( "Node should not have been shutdown.", NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); NodeState nodeState = rm.getRMContext().getInactiveRMNodes().get(nm2.getNodeId()).getState(); Assert.assertEquals("Node should have been shutdown but is in state" + nodeState, NodeState.SHUTDOWN, nodeState); }
@Test public void testAddNewExcludePathToConfiguration() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); ClusterMetrics metrics = ClusterMetrics.getMetrics(); assert(metrics != null); int initialMetricCount = metrics.getNumDecommisionedNMs(); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals( NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertEquals( NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); writeToHostsFile("host2"); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm.getNodesListManager().refreshNodes(conf); checkDecommissionedNMCount(rm, ++initialMetricCount); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals( "Node should not have been decommissioned.", NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); nodeHeartbeat = nm2.nodeHeartbeat(true); Assert.assertEquals( "Node should have been decommissioned but is in state" + nodeHeartbeat.getNodeAction(), NodeAction.SHUTDOWN, nodeHeartbeat.getNodeAction()); }
@Test public void testReboot() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:1234", 2048); int initialMetricCount = ClusterMetrics.getMetrics().getNumRebootedNMs(); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(nodeHeartbeat.getNodeAction())); nodeHeartbeat = nm2.nodeHeartbeat( new HashMap<ApplicationId, List<ContainerStatus>>(), true, -100); Assert.assertTrue(NodeAction.RESYNC.equals(nodeHeartbeat.getNodeAction())); Assert.assertEquals("Too far behind rm response id:0 nm response id:-100", nodeHeartbeat.getDiagnosticsMessage()); checkRebootedNMCount(rm, ++initialMetricCount); }
@Test public void testNodeHeartbeatForAppCollectorsMap() throws Exception { Configuration conf = new Configuration(); conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true); conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION, 2.0f); conf.set(YarnConfiguration.NM_AUX_SERVICES, "timeline_collector"); conf.set(YarnConfiguration.NM_AUX_SERVICES + "." + "timeline_collector" + ".class", PerNodeTimelineCollectorsAuxService.class.getName()); conf.setClass(YarnConfiguration.TIMELINE_SERVICE_WRITER_CLASS, FileSystemTimelineWriterImpl.class, TimelineWriter.class); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:1234", 2048); NodeHeartbeatResponse nodeHeartbeat1 = nm1.nodeHeartbeat(true); NodeHeartbeatResponse nodeHeartbeat2 = nm2.nodeHeartbeat(true); RMNodeImpl node1 = (RMNodeImpl) rm.getRMContext().getRMNodes().get(nm1.getNodeId()); RMNodeImpl node2 = (RMNodeImpl) rm.getRMContext().getRMNodes().get(nm2.getNodeId()); RMAppImpl app1 = (RMAppImpl) MockRMAppSubmitter.submitWithMemory(1024, rm); String collectorAddr1 = "1.2.3.4:5"; app1.setCollectorData(AppCollectorData.newInstance( app1.getApplicationId(), collectorAddr1)); String collectorAddr2 = "5.4.3.2:1"; RMAppImpl app2 = (RMAppImpl) MockRMAppSubmitter.submitWithMemory(1024, rm); app2.setCollectorData(AppCollectorData.newInstance( app2.getApplicationId(), collectorAddr2)); String collectorAddr3 = "5.4.3.2:2"; app2.setCollectorData(AppCollectorData.newInstance( app2.getApplicationId(), collectorAddr3, 0, 1)); String collectorAddr4 = "5.4.3.2:3"; app2.setCollectorData(AppCollectorData.newInstance( app2.getApplicationId(), collectorAddr4, 1, 0)); ContainerId runningContainerId1 = BuilderUtils.newContainerId( BuilderUtils.newApplicationAttemptId( app1.getApplicationId(), 0), 0); ContainerStatus status1 = ContainerStatus.newInstance(runningContainerId1, ContainerState.RUNNING, "", 0); List<ContainerStatus> statusList = new ArrayList<ContainerStatus>(); statusList.add(status1); NodeHealthStatus nodeHealth = NodeHealthStatus.newInstance(true, "", System.currentTimeMillis()); NodeStatus nodeStatus = NodeStatus.newInstance(nm1.getNodeId(), 0, statusList, null, nodeHealth, null, null, null); node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus)); Assert.assertEquals(1, node1.getRunningApps().size()); Assert.assertEquals(app1.getApplicationId(), node1.getRunningApps().get(0)); ContainerId runningContainerId2 = BuilderUtils.newContainerId( BuilderUtils.newApplicationAttemptId( app2.getApplicationId(), 0), 0); ContainerStatus status2 = ContainerStatus.newInstance(runningContainerId2, ContainerState.RUNNING, "", 0); statusList = new ArrayList<ContainerStatus>(); statusList.add(status2); nodeStatus = NodeStatus.newInstance(nm1.getNodeId(), 0, statusList, null, nodeHealth, null, null, null); node2.handle(new RMNodeStatusEvent(nm2.getNodeId(), nodeStatus)); Assert.assertEquals(1, node2.getRunningApps().size()); Assert.assertEquals(app2.getApplicationId(), node2.getRunningApps().get(0)); nodeHeartbeat1 = nm1.nodeHeartbeat(true); Map<ApplicationId, AppCollectorData> map1 = nodeHeartbeat1.getAppCollectors(); Assert.assertEquals(1, map1.size()); Assert.assertEquals(collectorAddr1, map1.get(app1.getApplicationId()).getCollectorAddr()); nodeHeartbeat2 = nm2.nodeHeartbeat(true); Map<ApplicationId, AppCollectorData> map2 = nodeHeartbeat2.getAppCollectors(); Assert.assertEquals(1, map2.size()); Assert.assertEquals(collectorAddr4, map2.get(app2.getApplicationId()).getCollectorAddr()); }
@Test public void testUnhealthyNodeStatus() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); Assert.assertEquals(0, ClusterMetrics.getMetrics().getUnhealthyNMs()); nm1.nodeHeartbeat(true); nm1.nodeHeartbeat(false); checkUnhealthyNMCount(rm, nm1, true, 1); nm1.nodeHeartbeat(true); checkUnhealthyNMCount(rm, nm1, false, 0); }
@Test public void testReconnectNode() throws Exception { rm = new MockRM() { @Override protected EventHandler<SchedulerEvent> createSchedulerEventDispatcher() { return new EventDispatcher<SchedulerEvent>(this.scheduler, this.scheduler.getClass().getName()) { @Override public void handle(SchedulerEvent event) { scheduler.handle(event); } }; } }; rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 5120); nm1.nodeHeartbeat(true); nm2.nodeHeartbeat(false); rm.drainEvents(); checkUnhealthyNMCount(rm, nm2, true, 1); final int expectedNMs = ClusterMetrics.getMetrics().getNumActiveNMs(); QueueMetrics metrics = rm.getResourceScheduler().getRootQueueMetrics(); Assert.assertEquals(5120, metrics.getAvailableMB()); nm1 = rm.registerNode("host1:1234", 5120); NodeHeartbeatResponse response = nm1.nodeHeartbeat(true); Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction())); rm.drainEvents(); Assert.assertEquals(expectedNMs, ClusterMetrics.getMetrics().getNumActiveNMs()); checkUnhealthyNMCount(rm, nm2, true, 1); nm2 = rm.registerNode("host2:5678", 5120); response = nm2.nodeHeartbeat(false); Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction())); rm.drainEvents(); Assert.assertEquals(expectedNMs, ClusterMetrics.getMetrics().getNumActiveNMs()); checkUnhealthyNMCount(rm, nm2, true, 1); nm2 = rm.registerNode("host2:5678", 5120); response = nm2.nodeHeartbeat(true); response = nm2.nodeHeartbeat(true); rm.drainEvents(); Assert.assertEquals(5120 + 5120, metrics.getAvailableMB()); nm1 = rm.registerNode("host2:5678", 10240); response = nm1.nodeHeartbeat(true); rm.drainEvents(); Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction())); Assert.assertEquals(5120 + 10240, metrics.getAvailableMB()); List<ApplicationId> runningApps = new ArrayList<ApplicationId>(); runningApps.add(ApplicationId.newInstance(1, 0)); nm1 = rm.registerNode("host2:5678", 15360, 2, runningApps); response = nm1.nodeHeartbeat(true); rm.drainEvents(); Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction())); Assert.assertEquals(5120 + 15360, metrics.getAvailableMB()); nm1 = new MockNM("host1:1234", 5120, rm.getResourceTrackerService()); nm1.setHttpPort(3); nm1.registerNode(); response = nm1.nodeHeartbeat(true); response = nm1.nodeHeartbeat(true); rm.drainEvents(); RMNode rmNode = rm.getRMContext().getRMNodes().get(nm1.getNodeId()); Assert.assertEquals(3, rmNode.getHttpPort()); Assert.assertEquals(5120, rmNode.getTotalCapability().getMemorySize()); Assert.assertEquals(5120 + 15360, metrics.getAvailableMB()); }
@Test(timeout = 30000) public void testInitDecommMetricNoRegistration() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); nm1.nodeHeartbeat(true); nm2.nodeHeartbeat(true); writeToHostsFile(excludeHostFile, "host3", "host2"); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, excludeHostFile.getAbsolutePath()); writeToHostsFile(hostFile, "host1", "host2"); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); rm.getNodesListManager().refreshNodes(conf); rm.drainEvents(); Assert.assertEquals("The decommissioned nodes metric should be 1 ", 1, ClusterMetrics.getMetrics().getNumDecommisionedNMs()); rm.stop(); MockRM rm1 = new MockRM(conf); rm1.start(); rm1.getNodesListManager().refreshNodes(conf); rm1.drainEvents(); Assert.assertEquals("The decommissioned nodes metric should be 2 ", 2, ClusterMetrics.getMetrics().getNumDecommisionedNMs()); rm1.stop(); }
@Test public void testIncorrectRecommission() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); MockNM nm2 = rm.registerNode("host2:5678", 10240); nm1.nodeHeartbeat(true); nm2.nodeHeartbeat(true); writeToHostsFile(excludeHostFile, "host3", "host2"); conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, excludeHostFile.getAbsolutePath()); writeToHostsFile(hostFile, "host1", "host2"); writeToHostsFile(excludeHostFile, "host1"); rm.getNodesListManager().refreshNodesGracefully(conf, null); rm.drainEvents(); nm1.nodeHeartbeat(true); rm.drainEvents(); Assert.assertTrue("Node " + nm1.getNodeId().getHost() + " should be Decommissioned", rm.getRMContext() .getInactiveRMNodes().get(nm1.getNodeId()).getState() == NodeState .DECOMMISSIONED); writeToHostsFile(excludeHostFile, ""); rm.getNodesListManager().refreshNodesGracefully(conf, null); rm.drainEvents(); Assert.assertTrue("Node " + nm1.getNodeId().getHost() + " should be Decommissioned", rm.getRMContext() .getInactiveRMNodes().get(nm1.getNodeId()).getState() == NodeState .DECOMMISSIONED); rm.stop(); }
@Test(timeout = 60000) public void testNodeHeartBeatResponseForUnknownContainerCleanUp() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.init(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); rm.drainEvents(); nm1.nodeHeartbeat(true); ApplicationId applicationId = BuilderUtils.newApplicationId(1, 1); ApplicationAttemptId applicationAttemptId = BuilderUtils .newApplicationAttemptId(applicationId, 1); ContainerId cid1 = BuilderUtils.newContainerId(applicationAttemptId, 2); ContainerId cid2 = BuilderUtils.newContainerId(applicationAttemptId, 3); ArrayList<ContainerStatus> containerStats = new ArrayList<ContainerStatus>(); containerStats.add( ContainerStatus.newInstance(cid1, ContainerState.COMPLETE, "", -1)); containerStats.add( ContainerStatus.newInstance(cid2, ContainerState.COMPLETE, "", -1)); Map<ApplicationId, List<ContainerStatus>> conts = new HashMap<ApplicationId, List<ContainerStatus>>(); conts.put(applicationAttemptId.getApplicationId(), containerStats); RMApp app1 = mock(RMApp.class); when(app1.getApplicationId()).thenReturn(applicationId); rm.getRMContext().getRMApps().put(applicationId, app1); nm1.nodeHeartbeat(conts, true); rm.drainEvents(); int containersToBeRemovedFromNM = 0; while (true) { NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); rm.drainEvents(); containersToBeRemovedFromNM += nodeHeartbeat.getContainersToBeRemovedFromNM().size(); if (containersToBeRemovedFromNM == 2) { break; } } }
@Test public void testResponseIdOverflow() throws Exception { Configuration conf = new Configuration(); rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("host1:1234", 5120); NodeHeartbeatResponse nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals(NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); RMNode node = rm.getRMContext().getRMNodes().get(nm1.getNodeId()); node.getLastNodeHeartBeatResponse().setResponseId(Integer.MAX_VALUE); nm1.setResponseId(Integer.MAX_VALUE); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals(NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); Assert.assertEquals(0, nodeHeartbeat.getResponseId()); nodeHeartbeat = nm1.nodeHeartbeat(true); Assert.assertEquals(NodeAction.NORMAL, nodeHeartbeat.getNodeAction()); Assert.assertEquals(1, nodeHeartbeat.getResponseId()); }