RMServerUtils { public static int getApplicableNodeCountForAM(RMContext rmContext, Configuration conf, List<ResourceRequest> amReqs) { Set<NodeId> nodesForReqs = new HashSet<>(); for (ResourceRequest amReq : amReqs) { if (amReq.getRelaxLocality() && !amReq.getResourceName().equals(ResourceRequest.ANY)) { nodesForReqs.addAll( rmContext.getScheduler().getNodeIds(amReq.getResourceName())); } } if (YarnConfiguration.areNodeLabelsEnabled(conf)) { String amNodeLabelExpression = amReqs.get(0).getNodeLabelExpression(); Set<NodeId> nodesForLabels = getNodeIdsForLabel(rmContext, amNodeLabelExpression); if (nodesForLabels != null && !nodesForLabels.isEmpty()) { if (nodesForReqs.isEmpty()) { for (Iterator<NodeId> it = nodesForLabels.iterator(); it.hasNext();) { if (it.next().getPort() == 0) { it.remove(); } } return nodesForLabels.size(); } else { return Sets.intersection(nodesForReqs, nodesForLabels).size(); } } } if (nodesForReqs.isEmpty()) { return rmContext.getScheduler().getNumClusterNodes(); } return nodesForReqs.size(); }  static List<RMNode> queryRMNodes(RMContext context,
      EnumSet<NodeState> acceptedStates); static ContainerUpdates validateAndSplitUpdateResourceRequests(RMContext rmContext,
      AllocateRequest request, Resource maximumAllocation,
      List<UpdateContainerError> updateErrors); static void normalizeAndValidateRequests(List<ResourceRequest> ask,
      Resource maximumAllocation, String queueName, YarnScheduler scheduler,
      RMContext rmContext, boolean nodeLabelsEnabled); static void checkSchedContainerChangeRequest(
      SchedContainerChangeRequest request, boolean increase); static void validateBlacklistRequest(
      ResourceBlacklistRequest blacklistRequest); static void validateContainerReleaseRequest(List<ContainerId> containerReleaseList,
      ApplicationAttemptId appAttemptId); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, final Logger LOG); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, String module,
      final Logger LOG); static YarnApplicationState createApplicationState(
      RMAppState rmAppState); static YarnApplicationAttemptState createApplicationAttemptState(
      RMAppAttemptState rmAppAttemptState); static void processRMProxyUsersConf(Configuration conf); static void validateApplicationTimeouts(
      Map<ApplicationTimeoutType, Long> timeouts); static Map<ApplicationTimeoutType, Long> validateISO8601AndConvertToLocalTimeEpoch(
      Map<ApplicationTimeoutType, String> timeoutsInISO8601); static int getApplicableNodeCountForAM(RMContext rmContext,
      Configuration conf, List<ResourceRequest> amReqs); static Long getOrDefault(Map<String, Long> map, String key,
      Long defaultValue); static final String UPDATE_OUTSTANDING_ERROR; static final String RESOURCE_OUTSIDE_ALLOWED_RANGE; static final ApplicationResourceUsageReport DUMMY_APPLICATION_RESOURCE_USAGE_REPORT; }
@Test public void testGetApplicableNodeCountForAMLocality() throws Exception { List<NodeId> rack1Nodes = new ArrayList<>(); for (int i = 0; i < 29; i++) { rack1Nodes.add(NodeId.newInstance("host" + i, 1234)); } NodeId node1 = NodeId.newInstance("node1", 1234); NodeId node2 = NodeId.newInstance("node2", 1234); rack1Nodes.add(node2); YarnConfiguration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, false); ResourceScheduler scheduler = Mockito.mock(ResourceScheduler.class); Mockito.when(scheduler.getNumClusterNodes()).thenReturn(100); Mockito.when(scheduler.getNodeIds("/rack1")).thenReturn(rack1Nodes); Mockito.when(scheduler.getNodeIds("node1")) .thenReturn(Collections.singletonList(node1)); Mockito.when(scheduler.getNodeIds("node2")) .thenReturn(Collections.singletonList(node2)); RMContext rmContext = Mockito.mock(RMContext.class); Mockito.when(rmContext.getScheduler()).thenReturn(scheduler); ResourceRequest anyReq = createResourceRequest(ResourceRequest.ANY, true, null); List<ResourceRequest> reqs = new ArrayList<>(); reqs.add(anyReq); Assert.assertEquals(100, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest rackReq = createResourceRequest("/rack1", true, null); reqs.add(rackReq); Assert.assertEquals(30, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); anyReq.setRelaxLocality(false); Assert.assertEquals(30, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(100, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest node1Req = createResourceRequest("node1", false, null); reqs.add(node1Req); Assert.assertEquals(100, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(true); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(true); Assert.assertEquals(31, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest node2Req = createResourceRequest("node2", false, null); reqs.add(node2Req); Assert.assertEquals(31, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(true); Assert.assertEquals(31, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(2, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(false); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(false); Assert.assertEquals(100, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); }
@Test public void testGetApplicableNodeCountForAMLabels() throws Exception { Set<NodeId> noLabelNodes = new HashSet<>(); for (int i = 0; i < 80; i++) { noLabelNodes.add(NodeId.newInstance("host" + i, 1234)); } Set<NodeId> label1Nodes = new HashSet<>(); for (int i = 80; i < 90; i++) { label1Nodes.add(NodeId.newInstance("host" + i, 1234)); } label1Nodes.add(NodeId.newInstance("host101", 0)); label1Nodes.add(NodeId.newInstance("host102", 0)); Map<String, Set<NodeId>> label1NodesMap = new HashMap<>(); label1NodesMap.put("label1", label1Nodes); YarnConfiguration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true); ResourceScheduler scheduler = Mockito.mock(ResourceScheduler.class); Mockito.when(scheduler.getNumClusterNodes()).thenReturn(100); RMContext rmContext = Mockito.mock(RMContext.class); Mockito.when(rmContext.getScheduler()).thenReturn(scheduler); RMNodeLabelsManager labMan = Mockito.mock(RMNodeLabelsManager.class); Mockito.when(labMan.getNodesWithoutALabel()).thenReturn(noLabelNodes); Mockito.when(labMan.getLabelsToNodes(Collections.singleton("label1"))) .thenReturn(label1NodesMap); Mockito.when(rmContext.getNodeLabelManager()).thenReturn(labMan); ResourceRequest anyReq = createResourceRequest(ResourceRequest.ANY, true, null); List<ResourceRequest> reqs = new ArrayList<>(); reqs.add(anyReq); Assert.assertEquals(80, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); anyReq.setNodeLabelExpression("label1"); Assert.assertEquals(10, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); }
@Test public void testGetApplicableNodeCountForAMLocalityAndLabels() throws Exception { List<NodeId> rack1Nodes = new ArrayList<>(); for (int i = 0; i < 29; i++) { rack1Nodes.add(NodeId.newInstance("host" + i, 1234)); } NodeId node1 = NodeId.newInstance("node1", 1234); NodeId node2 = NodeId.newInstance("node2", 1234); rack1Nodes.add(node2); Set<NodeId> noLabelNodes = new HashSet<>(); for (int i = 0; i < 19; i++) { noLabelNodes.add(rack1Nodes.get(i)); } noLabelNodes.add(node2); for (int i = 29; i < 89; i++) { noLabelNodes.add(NodeId.newInstance("host" + i, 1234)); } Set<NodeId> label1Nodes = new HashSet<>(); label1Nodes.add(node1); for (int i = 89; i < 93; i++) { label1Nodes.add(NodeId.newInstance("host" + i, 1234)); } for (int i = 19; i < 29; i++) { label1Nodes.add(rack1Nodes.get(i)); } label1Nodes.add(NodeId.newInstance("host101", 0)); label1Nodes.add(NodeId.newInstance("host102", 0)); Map<String, Set<NodeId>> label1NodesMap = new HashMap<>(); label1NodesMap.put("label1", label1Nodes); YarnConfiguration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true); ResourceScheduler scheduler = Mockito.mock(ResourceScheduler.class); Mockito.when(scheduler.getNumClusterNodes()).thenReturn(100); Mockito.when(scheduler.getNodeIds("/rack1")).thenReturn(rack1Nodes); Mockito.when(scheduler.getNodeIds("node1")) .thenReturn(Collections.singletonList(node1)); Mockito.when(scheduler.getNodeIds("node2")) .thenReturn(Collections.singletonList(node2)); RMContext rmContext = Mockito.mock(RMContext.class); Mockito.when(rmContext.getScheduler()).thenReturn(scheduler); RMNodeLabelsManager labMan = Mockito.mock(RMNodeLabelsManager.class); Mockito.when(labMan.getNodesWithoutALabel()).thenReturn(noLabelNodes); Mockito.when(labMan.getLabelsToNodes(Collections.singleton("label1"))) .thenReturn(label1NodesMap); Mockito.when(rmContext.getNodeLabelManager()).thenReturn(labMan); ResourceRequest anyReq = createResourceRequest(ResourceRequest.ANY, true, null); List<ResourceRequest> reqs = new ArrayList<>(); reqs.add(anyReq); Assert.assertEquals(80, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest rackReq = createResourceRequest("/rack1", true, null); reqs.add(rackReq); Assert.assertEquals(20, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); anyReq.setRelaxLocality(false); Assert.assertEquals(20, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(80, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest node1Req = createResourceRequest("node1", false, null); reqs.add(node1Req); Assert.assertEquals(80, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(true); Assert.assertEquals(0, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(true); Assert.assertEquals(20, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); ResourceRequest node2Req = createResourceRequest("node2", false, null); reqs.add(node2Req); Assert.assertEquals(20, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(true); Assert.assertEquals(20, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(false); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(false); Assert.assertEquals(80, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); anyReq.setNodeLabelExpression("label1"); rackReq.setNodeLabelExpression("label1"); node1Req.setNodeLabelExpression("label1"); node2Req.setNodeLabelExpression("label1"); anyReq.setRelaxLocality(true); reqs = new ArrayList<>(); reqs.add(anyReq); Assert.assertEquals(15, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(true); reqs.add(rackReq); Assert.assertEquals(10, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); anyReq.setRelaxLocality(false); Assert.assertEquals(10, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(15, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(false); reqs.add(node1Req); Assert.assertEquals(15, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(true); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(true); Assert.assertEquals(11, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(false); reqs.add(node2Req); Assert.assertEquals(11, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(true); Assert.assertEquals(11, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); rackReq.setRelaxLocality(false); Assert.assertEquals(1, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node1Req.setRelaxLocality(false); Assert.assertEquals(0, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); node2Req.setRelaxLocality(false); Assert.assertEquals(15, RMServerUtils.getApplicableNodeCountForAM(rmContext, conf, reqs)); }