ResourceTrackerService extends AbstractService implements
    ResourceTracker { @SuppressWarnings("unchecked") @Override public RegisterNodeManagerResponse registerNodeManager( RegisterNodeManagerRequest request) throws YarnException, IOException { NodeId nodeId = request.getNodeId(); String host = nodeId.getHost(); int cmPort = nodeId.getPort(); int httpPort = request.getHttpPort(); Resource capability = request.getResource(); String nodeManagerVersion = request.getNMVersion(); Resource physicalResource = request.getPhysicalResource(); NodeStatus nodeStatus = request.getNodeStatus(); RegisterNodeManagerResponse response = recordFactory .newRecordInstance(RegisterNodeManagerResponse.class); if (!minimumNodeManagerVersion.equals("NONE")) { if (minimumNodeManagerVersion.equals("EqualToRM")) { minimumNodeManagerVersion = YarnVersionInfo.getVersion(); } if ((nodeManagerVersion == null) || (VersionUtil.compareVersions(nodeManagerVersion,minimumNodeManagerVersion)) < 0) { String message = "Disallowed NodeManager Version " + nodeManagerVersion + ", is less than the minimum version " + minimumNodeManagerVersion + " sending SHUTDOWN signal to " + "NodeManager."; LOG.info(message); response.setDiagnosticsMessage(message); response.setNodeAction(NodeAction.SHUTDOWN); return response; } } if (checkIpHostnameInRegistration) { InetSocketAddress nmAddress = NetUtils.createSocketAddrForHost(host, cmPort); InetAddress inetAddress = Server.getRemoteIp(); if (inetAddress != null && nmAddress.isUnresolved()) { final String message = "hostname cannot be resolved (ip=" + inetAddress.getHostAddress() + ", hostname=" + host + ")"; LOG.warn("Unresolved nodemanager registration: " + message); response.setDiagnosticsMessage(message); response.setNodeAction(NodeAction.SHUTDOWN); return response; } } if (!this.nodesListManager.isValidNode(host) && !isNodeInDecommissioning(nodeId)) { String message = "Disallowed NodeManager from " + host + ", Sending SHUTDOWN signal to the NodeManager."; LOG.info(message); response.setDiagnosticsMessage(message); response.setNodeAction(NodeAction.SHUTDOWN); return response; } String nid = nodeId.toString(); Resource dynamicLoadCapability = loadNodeResourceFromDRConfiguration(nid); if (dynamicLoadCapability != null) { LOG.debug("Resource for node: {} is adjusted from: {} to: {} due to" + " settings in dynamic-resources.xml.", nid, capability, dynamicLoadCapability); capability = dynamicLoadCapability; response.setResource(capability); } if (capability.getMemorySize() < minAllocMb || capability.getVirtualCores() < minAllocVcores) { String message = "NodeManager from " + host + " doesn't satisfy minimum allocations, Sending SHUTDOWN" + " signal to the NodeManager. Node capabilities are " + capability + "; minimums are " + minAllocMb + "mb and " + minAllocVcores + " vcores"; LOG.info(message); response.setDiagnosticsMessage(message); response.setNodeAction(NodeAction.SHUTDOWN); return response; } response.setContainerTokenMasterKey(containerTokenSecretManager .getCurrentKey()); response.setNMTokenMasterKey(nmTokenSecretManager .getCurrentKey()); RMNode rmNode = new RMNodeImpl(nodeId, rmContext, host, cmPort, httpPort, resolve(host), capability, nodeManagerVersion, physicalResource); RMNode oldNode = this.rmContext.getRMNodes().putIfAbsent(nodeId, rmNode); if (oldNode == null) { RMNodeStartedEvent startEvent = new RMNodeStartedEvent(nodeId, request.getNMContainerStatuses(), request.getRunningApplications(), nodeStatus); if (request.getLogAggregationReportsForApps() != null && !request.getLogAggregationReportsForApps().isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("Found the number of previous cached log aggregation " + "status from nodemanager:" + nodeId + " is :" + request.getLogAggregationReportsForApps().size()); } startEvent.setLogAggregationReportsForApps(request .getLogAggregationReportsForApps()); } this.rmContext.getDispatcher().getEventHandler().handle( startEvent); } else { LOG.info("Reconnect from the node at: " + host); this.nmLivelinessMonitor.unregister(nodeId); if (CollectionUtils.isEmpty(request.getRunningApplications()) && rmNode.getState() != NodeState.DECOMMISSIONING && rmNode.getHttpPort() != oldNode.getHttpPort()) { switch (rmNode.getState()) { case RUNNING: ClusterMetrics.getMetrics().decrNumActiveNodes(); break; case UNHEALTHY: ClusterMetrics.getMetrics().decrNumUnhealthyNMs(); break; default: LOG.debug("Unexpected Rmnode state"); } this.rmContext.getDispatcher().getEventHandler() .handle(new NodeRemovedSchedulerEvent(rmNode)); this.rmContext.getRMNodes().put(nodeId, rmNode); this.rmContext.getDispatcher().getEventHandler() .handle(new RMNodeStartedEvent(nodeId, null, null, nodeStatus)); } else { oldNode.resetLastNodeHeartBeatResponse(); this.rmContext.getDispatcher().getEventHandler() .handle(new RMNodeReconnectEvent(nodeId, rmNode, request.getRunningApplications(), request.getNMContainerStatuses())); } } this.nmTokenSecretManager.removeNodeKey(nodeId); this.nmLivelinessMonitor.register(nodeId); if (!rmContext.isWorkPreservingRecoveryEnabled()) { if (!request.getNMContainerStatuses().isEmpty()) { LOG.info("received container statuses on node manager register :" + request.getNMContainerStatuses()); for (NMContainerStatus status : request.getNMContainerStatuses()) { handleNMContainerStatus(status, nodeId); } } } Set<String> nodeLabels = NodeLabelsUtils.convertToStringSet( request.getNodeLabels()); if (isDistributedNodeLabelsConf && nodeLabels != null) { try { updateNodeLabelsFromNMReport(nodeLabels, nodeId); response.setAreNodeLabelsAcceptedByRM(true); } catch (IOException ex) { response.setDiagnosticsMessage(ex.getMessage()); response.setAreNodeLabelsAcceptedByRM(false); } } else if (isDelegatedCentralizedNodeLabelsConf) { this.rmContext.getRMDelegatedNodeLabelsUpdater().updateNodeLabels(nodeId); } if (request.getNodeAttributes() != null) { try { updateNodeAttributesIfNecessary(nodeId, request.getNodeAttributes()); response.setAreNodeAttributesAcceptedByRM(true); } catch (IOException ex) { String errorMsg = response.getDiagnosticsMessage() == null ? ex.getMessage() : response.getDiagnosticsMessage() + "\n" + ex.getMessage(); response.setDiagnosticsMessage(errorMsg); response.setAreNodeAttributesAcceptedByRM(false); } } StringBuilder message = new StringBuilder(); message.append("NodeManager from node ").append(host).append("(cmPort: ") .append(cmPort).append(" httpPort: "); message.append(httpPort).append(") ") .append("registered with capability: ").append(capability); message.append(", assigned nodeId ").append(nodeId); if (response.getAreNodeLabelsAcceptedByRM()) { message.append(", node labels { ").append( StringUtils.join(",", nodeLabels) + " } "); } if (response.getAreNodeAttributesAcceptedByRM()) { message.append(", node attributes { ") .append(request.getNodeAttributes() + " } "); } LOG.info(message.toString()); response.setNodeAction(NodeAction.NORMAL); response.setRMIdentifier(ResourceManager.getClusterTimeStamp()); response.setRMVersion(YarnVersionInfo.getVersion()); return response; } 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 public void testNodeRegistrationSuccess() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm = new MockRM(conf); rm.start(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord( RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(YarnVersionInfo.getVersion()); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.NORMAL, response.getNodeAction()); }
@Test public void testNodeRegistrationWithLabels() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.set(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE, YarnConfiguration.DISTRIBUTED_NODELABEL_CONFIGURATION_TYPE); final RMNodeLabelsManager nodeLabelsMgr = new NullRMNodeLabelsManager(); rm = new MockRM(conf) { @Override protected RMNodeLabelsManager createNodeLabelManager() { return nodeLabelsMgr; } }; rm.start(); try { nodeLabelsMgr.addToCluserNodeLabelsWithDefaultExclusivity(toSet("A", "B", "C")); } catch (IOException e) { Assert.fail("Caught Exception while initializing"); e.printStackTrace(); } ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest registerReq = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); registerReq.setResource(capability); registerReq.setNodeId(nodeId); registerReq.setHttpPort(1234); registerReq.setNMVersion(YarnVersionInfo.getVersion()); registerReq.setNodeLabels(toSet(NodeLabel.newInstance("A"))); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(registerReq); Assert.assertEquals("Action should be normal on valid Node Labels", NodeAction.NORMAL, response.getNodeAction()); assertCollectionEquals(nodeLabelsMgr.getNodeLabels().get(nodeId), NodeLabelsUtils.convertToStringSet(registerReq.getNodeLabels())); Assert.assertTrue("Valid Node Labels were not accepted by RM", response.getAreNodeLabelsAcceptedByRM()); rm.stop(); }
@Test public void testNodeRegistrationWithInvalidLabels() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.set(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE, YarnConfiguration.DISTRIBUTED_NODELABEL_CONFIGURATION_TYPE); final RMNodeLabelsManager nodeLabelsMgr = new NullRMNodeLabelsManager(); rm = new MockRM(conf) { @Override protected RMNodeLabelsManager createNodeLabelManager() { return nodeLabelsMgr; } }; rm.start(); try { nodeLabelsMgr.addToCluserNodeLabelsWithDefaultExclusivity(toSet("X", "Y", "Z")); } catch (IOException e) { Assert.fail("Caught Exception while initializing"); e.printStackTrace(); } ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest registerReq = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); registerReq.setResource(capability); registerReq.setNodeId(nodeId); registerReq.setHttpPort(1234); registerReq.setNMVersion(YarnVersionInfo.getVersion()); registerReq.setNodeLabels(toNodeLabelSet("A", "B", "C")); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(registerReq); Assert.assertEquals( "On Invalid Node Labels action is expected to be normal", NodeAction.NORMAL, response.getNodeAction()); Assert.assertNull(nodeLabelsMgr.getNodeLabels().get(nodeId)); Assert.assertNotNull(response.getDiagnosticsMessage()); Assert.assertFalse("Node Labels should not accepted by RM If Invalid", response.getAreNodeLabelsAcceptedByRM()); if (rm != null) { rm.stop(); } }
@Test public void testNodeRegistrationWithInvalidLabelsSyntax() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.set(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE, YarnConfiguration.DISTRIBUTED_NODELABEL_CONFIGURATION_TYPE); final RMNodeLabelsManager nodeLabelsMgr = new NullRMNodeLabelsManager(); rm = new MockRM(conf) { @Override protected RMNodeLabelsManager createNodeLabelManager() { return nodeLabelsMgr; } }; rm.start(); try { nodeLabelsMgr.addToCluserNodeLabelsWithDefaultExclusivity(toSet("X", "Y", "Z")); } catch (IOException e) { Assert.fail("Caught Exception while initializing"); e.printStackTrace(); } ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(YarnVersionInfo.getVersion()); req.setNodeLabels(toNodeLabelSet("#Y")); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals( "On Invalid Node Labels action is expected to be normal", NodeAction.NORMAL, response.getNodeAction()); Assert.assertNull(nodeLabelsMgr.getNodeLabels().get(nodeId)); Assert.assertNotNull(response.getDiagnosticsMessage()); Assert.assertFalse("Node Labels should not accepted by RM If Invalid", response.getAreNodeLabelsAcceptedByRM()); if (rm != null) { rm.stop(); } }
@Test public void testNodeRegistrationWithCentralLabelConfig() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.set(YarnConfiguration.NODELABEL_CONFIGURATION_TYPE, YarnConfiguration.DEFAULT_NODELABEL_CONFIGURATION_TYPE); final RMNodeLabelsManager nodeLabelsMgr = new NullRMNodeLabelsManager(); rm = new MockRM(conf) { @Override protected RMNodeLabelsManager createNodeLabelManager() { return nodeLabelsMgr; } }; rm.start(); try { nodeLabelsMgr.addToCluserNodeLabelsWithDefaultExclusivity(toSet("A", "B", "C")); } catch (IOException e) { Assert.fail("Caught Exception while initializing"); e.printStackTrace(); } ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(YarnVersionInfo.getVersion()); req.setNodeLabels(toNodeLabelSet("A")); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.NORMAL, response.getNodeAction()); Assert.assertNull(nodeLabelsMgr.getNodeLabels().get(nodeId)); Assert .assertFalse( "Node Labels should not accepted by RM If its configured with " + "Central configuration", response.getAreNodeLabelsAcceptedByRM()); if (rm != null) { rm.stop(); } }
@Test public void testNodeRegistrationWithAttributes() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.setClass(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_IMPL_CLASS, FileSystemNodeAttributeStore.class, NodeAttributeStore.class); File tempDir = File.createTempFile("nattr", ".tmp"); tempDir.delete(); tempDir.mkdirs(); tempDir.deleteOnExit(); conf.set(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_ROOT_DIR, tempDir.getAbsolutePath()); rm = new MockRM(conf); rm.start(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest registerReq = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); NodeAttribute nodeAttribute1 = NodeAttribute .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr1", NodeAttributeType.STRING, "V1"); NodeAttribute nodeAttribute2 = NodeAttribute .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr2", NodeAttributeType.STRING, "V2"); registerReq.setResource(capability); registerReq.setNodeId(nodeId); registerReq.setHttpPort(1234); registerReq.setNMVersion(YarnVersionInfo.getVersion()); registerReq.setNodeAttributes(toSet(nodeAttribute1, nodeAttribute2)); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(registerReq); Assert.assertEquals("Action should be normal on valid Node Attributes", NodeAction.NORMAL, response.getNodeAction()); Assert.assertTrue(NodeLabelUtil.isNodeAttributesEquals( rm.getRMContext().getNodeAttributesManager() .getAttributesForNode(nodeId.getHost()).keySet(), registerReq.getNodeAttributes())); Assert.assertTrue("Valid Node Attributes were not accepted by RM", response.getAreNodeAttributesAcceptedByRM()); if (rm != null) { rm.stop(); } }
@Test public void testNodeRegistrationWithInvalidAttributes() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile.getAbsolutePath()); conf.setClass(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_IMPL_CLASS, FileSystemNodeAttributeStore.class, NodeAttributeStore.class); conf.set(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_ROOT_DIR, TEMP_DIR.getAbsolutePath()); rm = new MockRM(conf); rm.start(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord(RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); NodeAttribute validNodeAttribute = NodeAttribute .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr1", NodeAttributeType.STRING, "V1"); NodeAttribute invalidPrefixNodeAttribute = NodeAttribute .newInstance("_P", "Attr1", NodeAttributeType.STRING, "V2"); NodeAttribute invalidNameNodeAttribute = NodeAttribute .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "_N", NodeAttributeType.STRING, "V2"); NodeAttribute invalidValueNodeAttribute = NodeAttribute .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr2", NodeAttributeType.STRING, "..."); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(YarnVersionInfo.getVersion()); req.setNodeAttributes( toSet(validNodeAttribute, invalidPrefixNodeAttribute)); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(0, rm.getRMContext().getNodeAttributesManager() .getAttributesForNode(nodeId.getHost()).size()); assertRegisterResponseForInvalidAttributes(response); Assert.assertTrue(response.getDiagnosticsMessage() .endsWith("attributes in HB must have prefix nm.yarn.io")); req.setNodeAttributes(toSet(validNodeAttribute, invalidNameNodeAttribute)); response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(0, rm.getRMContext().getNodeAttributesManager() .getAttributesForNode(nodeId.getHost()).size()); assertRegisterResponseForInvalidAttributes(response); Assert.assertTrue(response.getDiagnosticsMessage() .startsWith("attribute name should only contains")); req.setNodeAttributes(toSet(validNodeAttribute, invalidValueNodeAttribute)); response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(0, rm.getRMContext().getNodeAttributesManager() .getAttributesForNode(nodeId.getHost()).size()); assertRegisterResponseForInvalidAttributes(response); Assert.assertTrue(response.getDiagnosticsMessage() .startsWith("attribute value should only contains")); if (rm != null) { rm.stop(); } }
@Test public void testNodeRegistrationVersionLessThanRM() throws Exception { writeToHostsFile("host2"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile .getAbsolutePath()); conf.set(YarnConfiguration.RM_NODEMANAGER_MINIMUM_VERSION,"EqualToRM" ); rm = new MockRM(conf); rm.start(); String nmVersion = "1.9.9"; ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord( RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(nmVersion); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.SHUTDOWN,response.getNodeAction()); Assert.assertTrue("Diagnostic message did not contain: 'Disallowed NodeManager " + "Version "+ nmVersion + ", is less than the minimum version'", response.getDiagnosticsMessage().contains("Disallowed NodeManager Version " + nmVersion + ", is less than the minimum version ")); }
@Test public void testNodeRegistrationFailure() throws Exception { writeToHostsFile("host1"); Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH, hostFile .getAbsolutePath()); rm = new MockRM(conf); rm.start(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord( RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); req.setNodeId(nodeId); req.setHttpPort(1234); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.SHUTDOWN,response.getNodeAction()); Assert .assertEquals( "Disallowed NodeManager from host2, Sending SHUTDOWN signal to the NodeManager.", response.getDiagnosticsMessage()); }
@Test public void testNodeRegistrationWithMinimumAllocations() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, "2048"); conf.set(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, "4"); rm = new MockRM(conf); rm.start(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); RegisterNodeManagerRequest req = Records.newRecord( RegisterNodeManagerRequest.class); NodeId nodeId = BuilderUtils.newNodeId("host", 1234); req.setNodeId(nodeId); Resource capability = BuilderUtils.newResource(1024, 1); req.setResource(capability); RegisterNodeManagerResponse response1 = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.SHUTDOWN,response1.getNodeAction()); capability.setMemorySize(2048); capability.setVirtualCores(1); req.setResource(capability); RegisterNodeManagerResponse response2 = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.SHUTDOWN,response2.getNodeAction()); capability.setMemorySize(1024); capability.setVirtualCores(4); req.setResource(capability); RegisterNodeManagerResponse response3 = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.SHUTDOWN,response3.getNodeAction()); capability.setMemorySize(2048); capability.setVirtualCores(4); req.setResource(capability); RegisterNodeManagerResponse response4 = resourceTrackerService.registerNodeManager(req); Assert.assertEquals(NodeAction.NORMAL,response4.getNodeAction()); }
@SuppressWarnings("unchecked") @Test public void testHandleOpportunisticContainerStatus() throws Exception{ final DrainDispatcher dispatcher = new DrainDispatcher(); YarnConfiguration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true); conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, true); rm = new MockRM(conf){ @Override protected Dispatcher createDispatcher() { return dispatcher; } }; rm.start(); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1024, rm) .withUnmanagedAM(true) .build(); RMApp app = MockRMAppSubmitter.submit(rm, data); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); ResourceTrackerService resourceTrackerService = rm.getResourceTrackerService(); SchedulerApplicationAttempt applicationAttempt = null; while (applicationAttempt == null) { applicationAttempt = ((AbstractYarnScheduler)rm.getRMContext().getScheduler()) .getApplicationAttempt(appAttemptId); Thread.sleep(100); } Resource currentConsumption = applicationAttempt.getCurrentConsumption(); Assert.assertEquals(Resource.newInstance(0, 0), currentConsumption); Resource allocResources = applicationAttempt.getQueue().getMetrics().getAllocatedResources(); Assert.assertEquals(Resource.newInstance(0, 0), allocResources); RegisterNodeManagerRequest req = Records.newRecord( RegisterNodeManagerRequest.class); NodeId nodeId = NodeId.newInstance("host2", 1234); Resource capability = BuilderUtils.newResource(1024, 1); NodeStatus mockNodeStatus = createMockNodeStatus(); req.setResource(capability); req.setNodeId(nodeId); req.setHttpPort(1234); req.setNMVersion(YarnVersionInfo.getVersion()); req.setNodeStatus(mockNodeStatus); ContainerId c1 = ContainerId.newContainerId(appAttemptId, 1); ContainerId c2 = ContainerId.newContainerId(appAttemptId, 2); ContainerId c3 = ContainerId.newContainerId(appAttemptId, 3); NMContainerStatus queuedOpp = NMContainerStatus.newInstance(c1, 1, ContainerState.RUNNING, Resource.newInstance(1024, 1), "Dummy Queued OC", ContainerExitStatus.INVALID, Priority.newInstance(5), 1234, "", ExecutionType.OPPORTUNISTIC, -1); NMContainerStatus runningOpp = NMContainerStatus.newInstance(c2, 1, ContainerState.RUNNING, Resource.newInstance(2048, 1), "Dummy Running OC", ContainerExitStatus.INVALID, Priority.newInstance(6), 1234, "", ExecutionType.OPPORTUNISTIC, -1); NMContainerStatus runningGuar = NMContainerStatus.newInstance(c3, 1, ContainerState.RUNNING, Resource.newInstance(2048, 1), "Dummy Running GC", ContainerExitStatus.INVALID, Priority.newInstance(6), 1234, "", ExecutionType.GUARANTEED, -1); req.setContainerStatuses(Arrays.asList(queuedOpp, runningOpp, runningGuar)); RegisterNodeManagerResponse response = resourceTrackerService.registerNodeManager(req); dispatcher.await(); Thread.sleep(2000); dispatcher.await(); Assert.assertEquals(NodeAction.NORMAL, response.getNodeAction()); Collection<RMContainer> liveContainers = applicationAttempt .getLiveContainers(); Assert.assertEquals(3, liveContainers.size()); Iterator<RMContainer> iter = liveContainers.iterator(); while (iter.hasNext()) { RMContainer rc = iter.next(); Assert.assertEquals( rc.getContainerId().equals(c3) ? ExecutionType.GUARANTEED : ExecutionType.OPPORTUNISTIC, rc.getExecutionType()); } currentConsumption = applicationAttempt.getCurrentConsumption(); Assert.assertEquals(Resource.newInstance(2048, 1), currentConsumption); allocResources = applicationAttempt.getQueue().getMetrics().getAllocatedResources(); Assert.assertEquals(Resource.newInstance(2048, 1), allocResources); SchedulerNode schedulerNode = rm.getRMContext().getScheduler().getSchedulerNode(nodeId); Assert.assertNotNull(schedulerNode); Resource nodeResources = schedulerNode.getAllocatedResource(); Assert.assertEquals(Resource.newInstance(2048, 1), nodeResources); }
@Test public void testNMIpHostNameResolution() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS, "localhost:" + ServerSocketUtil.getPort(10000, 10)); conf.setBoolean(YarnConfiguration.RM_NM_REGISTRATION_IP_HOSTNAME_CHECK_KEY, true); MockRM mockRM = new MockRM(conf) { @Override protected ResourceTrackerService createResourceTrackerService() { return new ResourceTrackerService(getRMContext(), nodesListManager, this.nmLivelinessMonitor, rmContext.getContainerTokenSecretManager(), rmContext.getNMTokenSecretManager()) { }; } }; mockRM.start(); ResourceTracker rmTracker = ServerRMProxy.createRMProxy(mockRM.getConfig(), ResourceTracker.class); RegisterNodeManagerResponse response = rmTracker.registerNodeManager( RegisterNodeManagerRequest.newInstance( NodeId.newInstance("host1" + System.currentTimeMillis(), 1234), 1236, Resource.newInstance(10000, 10), "2", new ArrayList<>(), new ArrayList<>())); Assert .assertEquals("Shutdown signal should be received", NodeAction.SHUTDOWN, response.getNodeAction()); Assert.assertTrue("Diagnostic Message", response.getDiagnosticsMessage() .contains("hostname cannot be resolved ")); rmTracker = ServerRMProxy.createRMProxy(mockRM.getConfig(), ResourceTracker.class); response = rmTracker.registerNodeManager(RegisterNodeManagerRequest .newInstance(NodeId.newInstance("localhost", 1234), 1236, Resource.newInstance(10000, 10), "2", new ArrayList<>(), new ArrayList<>())); Assert.assertEquals("Successfull registration", NodeAction.NORMAL, response.getNodeAction()); mockRM.stop(); }