PlacementConstraintsUtil { private static boolean canSatisfyConstraints(ApplicationId appId, PlacementConstraint constraint, SchedulerNode node, AllocationTagsManager atm) throws InvalidAllocationTagsQueryException { if (constraint == null) { if(LOG.isDebugEnabled()) { LOG.debug( "Constraint is found empty during constraint validation for app:" + appId); } return true; } SingleConstraintTransformer singleTransformer = new SingleConstraintTransformer(constraint); constraint = singleTransformer.transform(); AbstractConstraint sConstraintExpr = constraint.getConstraintExpr(); if (sConstraintExpr instanceof SingleConstraint) { SingleConstraint single = (SingleConstraint) sConstraintExpr; return canSatisfySingleConstraint(appId, single, node, atm); } else if (sConstraintExpr instanceof And) { And and = (And) sConstraintExpr; return canSatisfyAndConstraint(appId, and, node, atm); } else if (sConstraintExpr instanceof Or) { Or or = (Or) sConstraintExpr; return canSatisfyOrConstraint(appId, or, node, atm); } else { throw new InvalidAllocationTagsQueryException( "Unsupported type of constraint: " + sConstraintExpr.getClass().getSimpleName()); } } private  PlacementConstraintsUtil(); static boolean canSatisfyConstraints(ApplicationId applicationId,
      SchedulingRequest request, SchedulerNode schedulerNode,
      PlacementConstraintManager pcm, AllocationTagsManager atm);  }
@Test public void testNodeAffinityAssignment() throws InvalidAllocationTagsQueryException { PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); AllocationTagsManager tm = new AllocationTagsManager(rmContext); pcm.registerApplication(appId1, constraintMap1); Iterator<RMNode> nodeIterator = rmNodes.iterator(); while (nodeIterator.hasNext()) { RMNode currentNode = nodeIterator.next(); SchedulerNode schedulerNode =newSchedulerNode(currentNode.getHostName(), currentNode.getRackName(), currentNode.getNodeID()); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode, pcm, tm)); } RMNode n0_r1 = rmNodes.get(0); RMNode n1_r1 = rmNodes.get(1); RMNode n2_r2 = rmNodes.get(2); RMNode n3_r2 = rmNodes.get(3); SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(), n0_r1.getRackName(), n0_r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(), n1_r1.getRackName(), n1_r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(), n2_r2.getRackName(), n2_r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(), n3_r2.getRackName(), n3_r2.getNodeID()); ContainerId hbase_m = ContainerId .newContainerId(ApplicationAttemptId.newInstance(appId1, 0), 0); tm.addContainer(n0_r1.getNodeID(), hbase_m, ImmutableSet.of("hbase-m")); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); }
@Test public void testMultiTagsPlacementConstraints() throws InvalidAllocationTagsQueryException { PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); AllocationTagsManager tm = new AllocationTagsManager(rmContext); rmContext.setAllocationTagsManager(tm); rmContext.setPlacementConstraintManager(pcm); HashSet<String> st1 = new HashSet<>(Arrays.asList("X")); HashSet<String> st2 = new HashSet<>(Arrays.asList("Y")); PlacementConstraint pc1 = PlacementConstraints.build( targetNotIn(NODE, allocationTag("A", "B"))); PlacementConstraint pc2 = PlacementConstraints.build( targetIn(NODE, allocationTag("A", "B"))); Map<Set<String>, PlacementConstraint> constraintMap = ImmutableMap.of(st1, pc1, st2, pc2); pcm.registerApplication(appId1, constraintMap); RMNode n0_r1 = rmNodes.get(0); RMNode n1_r1 = rmNodes.get(1); RMNode n2_r2 = rmNodes.get(2); RMNode n3_r2 = rmNodes.get(3); SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(), n0_r1.getRackName(), n0_r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(), n1_r1.getRackName(), n1_r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(), n2_r2.getRackName(), n2_r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(), n3_r2.getRackName(), n3_r2.getNodeID()); ContainerId ca = newContainerId(appId1, 0); tm.addContainer(n0_r1.getNodeID(), ca, ImmutableSet.of("A")); ContainerId cb = newContainerId(appId1, 1); tm.addContainer(n1_r1.getNodeID(), cb, ImmutableSet.of("B")); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st1), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st1), schedulerNode3, pcm, tm)); ContainerId ca1 = newContainerId(appId1, 2); tm.addContainer(n2_r2.getNodeID(), ca1, ImmutableSet.of("A")); ContainerId cb1 = newContainerId(appId1, 3); tm.addContainer(n2_r2.getNodeID(), cb1, ImmutableSet.of("B")); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st2), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st2), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st2), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(st2), schedulerNode3, pcm, tm)); }
@Test public void testRackAffinityAssignment() throws InvalidAllocationTagsQueryException { PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); AllocationTagsManager tm = new AllocationTagsManager(rmContext); pcm.registerApplication(appId1, constraintMap1); RMNode n0_r1 = rmNodes.get(0); RMNode n1_r1 = rmNodes.get(1); RMNode n2_r2 = rmNodes.get(2); RMNode n3_r2 = rmNodes.get(3); ContainerId hbase_m = ContainerId .newContainerId(ApplicationAttemptId.newInstance(appId1, 0), 0); tm.addContainer(n0_r1.getNodeID(), hbase_m, ImmutableSet.of("hbase-rs")); SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(), n0_r1.getRackName(), n0_r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(), n1_r1.getRackName(), n1_r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(), n2_r2.getRackName(), n2_r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(), n3_r2.getRackName(), n3_r2.getNodeID()); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode3, pcm, tm)); }
@Test public void testNodeAntiAffinityAssignment() throws InvalidAllocationTagsQueryException { PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); AllocationTagsManager tm = new AllocationTagsManager(rmContext); pcm.registerApplication(appId1, constraintMap2); RMNode n0_r1 = rmNodes.get(0); RMNode n1_r1 = rmNodes.get(1); RMNode n2_r2 = rmNodes.get(2); RMNode n3_r2 = rmNodes.get(3); SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(), n0_r1.getRackName(), n0_r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(), n1_r1.getRackName(), n1_r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(), n2_r2.getRackName(), n2_r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(), n3_r2.getRackName(), n3_r2.getNodeID()); ContainerId hbase_m = ContainerId .newContainerId(ApplicationAttemptId.newInstance(appId1, 0), 0); tm.addContainer(n0_r1.getNodeID(), hbase_m, ImmutableSet.of("hbase-m")); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); }
@Test public void testRackAntiAffinityAssignment() throws InvalidAllocationTagsQueryException { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); pcm.registerApplication(appId1, constraintMap2); RMNode n0_r1 = rmNodes.get(0); RMNode n1_r1 = rmNodes.get(1); RMNode n2_r2 = rmNodes.get(2); RMNode n3_r2 = rmNodes.get(3); ContainerId hbase_m = ContainerId .newContainerId(ApplicationAttemptId.newInstance(appId1, 0), 0); tm.addContainer(n0_r1.getNodeID(), hbase_m, ImmutableSet.of("hbase-rs")); SchedulerNode schedulerNode0 =newSchedulerNode(n0_r1.getHostName(), n0_r1.getRackName(), n0_r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1_r1.getHostName(), n1_r1.getRackName(), n1_r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2_r2.getHostName(), n2_r2.getRackName(), n2_r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3_r2.getHostName(), n3_r2.getRackName(), n3_r2.getNodeID()); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode3, pcm, tm)); }
@Test public void testORConstraintAssignment() throws InvalidAllocationTagsQueryException { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); pcm.registerApplication(appId1, constraintMap4); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(appId1, 1), ImmutableSet.of("hbase-m")); tm.addContainer(n2r2.getNodeID(), newContainerId(appId1, 2), ImmutableSet.of("hbase-rs")); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n0r1.getNodeID()) .get("hbase-m").longValue()); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n2r2.getNodeID()) .get("hbase-rs").longValue()); SchedulerNode schedulerNode0 =newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); tm.addContainer(n3r2.getNodeID(), newContainerId(appId1, 2), ImmutableSet.of("hbase-rs")); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); tm.addContainer(n2r2.getNodeID(), newContainerId(appId1, 3), ImmutableSet.of("spark")); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag2), schedulerNode3, pcm, tm)); }
@Test public void testANDConstraintAssignment() throws InvalidAllocationTagsQueryException { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); pcm.registerApplication(appId1, constraintMap3); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(appId1, 0), ImmutableSet.of("hbase-m")); tm.addContainer(n2r2.getNodeID(), newContainerId(appId1, 1), ImmutableSet.of("hbase-m")); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n0r1.getNodeID()) .get("hbase-m").longValue()); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n2r2.getNodeID()) .get("hbase-m").longValue()); SchedulerNode schedulerNode0 =newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); for (int i=0; i<4; i++) { tm.addContainer(n1r1.getNodeID(), newContainerId(appId1, i+2), ImmutableSet.of("spark")); } Assert.assertEquals(4L, tm.getAllocationTagsWithCount(n1r1.getNodeID()) .get("spark").longValue()); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints(appId1, createSchedulingRequest(sourceTag1), schedulerNode3, pcm, tm)); }
@Test public void testGlobalAppConstraints() throws InvalidAllocationTagsQueryException { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); rmContext.setAllocationTagsManager(tm); rmContext.setPlacementConstraintManager(pcm); long ts = System.currentTimeMillis(); ApplicationId application1 = BuilderUtils.newApplicationId(ts, 100); ApplicationId application2 = BuilderUtils.newApplicationId(ts, 101); ApplicationId application3 = BuilderUtils.newApplicationId(ts, 102); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(application1, 0), ImmutableSet.of("A")); tm.addContainer(n0r1.getNodeID(), newContainerId(application2, 1), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3, 2), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3, 3), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3, 4), ImmutableSet.of("A")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1, 5), ImmutableSet.of("A")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1, 6), ImmutableSet.of("A")); SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 = newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 = newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); TargetApplicationsNamespace namespaceAll = new TargetApplicationsNamespace.All(); PlacementConstraint constraint1 = PlacementConstraints.targetNotIn( NODE, allocationTagWithNamespace(namespaceAll.toString(), "A")) .build(); Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>(); Set<String> srcTags1 = ImmutableSet.of("A"); constraintMap.put(srcTags1, constraint1); pcm.registerApplication(application1, constraintMap); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application1); PlacementConstraint constraint2 = PlacementConstraints .maxCardinality(NODE, namespaceAll.toString(), 2, "A") .build(); constraintMap.clear(); Set<String> srcTags2 = ImmutableSet.of("foo"); constraintMap.put(srcTags2, constraint2); pcm.registerApplication(application2, constraintMap); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application2); PlacementConstraint constraint3 = PlacementConstraints .minCardinality(NODE, namespaceAll.toString(), 3, "A") .build(); constraintMap.clear(); Set<String> srcTags3 = ImmutableSet.of("foo"); constraintMap.put(srcTags3, constraint3); pcm.registerApplication(application3, constraintMap); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application3, createSchedulingRequest(srcTags3), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application3, createSchedulingRequest(srcTags3), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application3, createSchedulingRequest(srcTags3), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application3, createSchedulingRequest(srcTags3), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application3); }
@Test public void testNotSelfAppConstraints() throws InvalidAllocationTagsQueryException { long ts = System.currentTimeMillis(); ApplicationId application1 = BuilderUtils.newApplicationId(ts, 100); ApplicationId application2 = BuilderUtils.newApplicationId(ts, 101); ApplicationId application3 = BuilderUtils.newApplicationId(ts, 102); ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>(); allApps.put(application1, new MockRMApp(123, 1000, RMAppState.NEW, "userA", ImmutableSet.of(""))); allApps.put(application2, new MockRMApp(124, 1001, RMAppState.NEW, "userA", ImmutableSet.of(""))); allApps.put(application3, new MockRMApp(125, 1002, RMAppState.NEW, "userA", ImmutableSet.of(""))); RMContext mockedContext = Mockito.spy(rmContext); when(mockedContext.getRMApps()).thenReturn(allApps); AllocationTagsManager tm = new AllocationTagsManager(mockedContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); mockedContext.setAllocationTagsManager(tm); mockedContext.setPlacementConstraintManager(pcm); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(application1), ImmutableSet.of("A")); tm.addContainer(n0r1.getNodeID(), newContainerId(application2), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3), ImmutableSet.of("A")); tm.addContainer(n1r1.getNodeID(), newContainerId(application3), ImmutableSet.of("A")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1), ImmutableSet.of("A")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1), ImmutableSet.of("A")); SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 = newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 = newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); TargetApplicationsNamespace notSelf = new TargetApplicationsNamespace.NotSelf(); PlacementConstraint constraint1 = PlacementConstraints.targetNotIn( NODE, allocationTagWithNamespace(notSelf.toString(), "A")) .build(); Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>(); Set<String> srcTags1 = ImmutableSet.of("A"); constraintMap.put(srcTags1, constraint1); pcm.registerApplication(application1, constraintMap); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode0, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags1), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application1); PlacementConstraint constraint2 = PlacementConstraints.targetIn( NODE, allocationTagWithNamespace(notSelf.toString(), "A")) .build(); Map<Set<String>, PlacementConstraint> cm2 = new HashMap<>(); Set<String> srcTags2 = ImmutableSet.of("A"); cm2.put(srcTags2, constraint2); pcm.registerApplication(application1, cm2); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags2), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags2), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags2), schedulerNode2, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application1, createSchedulingRequest(srcTags2), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application1); }
@Test public void testInterAppConstraintsByAppID() throws InvalidAllocationTagsQueryException { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); rmContext.setAllocationTagsManager(tm); rmContext.setPlacementConstraintManager(pcm); long ts = System.currentTimeMillis(); ApplicationId application1 = BuilderUtils.newApplicationId(ts, 123); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(application1, 0), ImmutableSet.of("hbase-m")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1, 1), ImmutableSet.of("hbase-m")); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n0r1.getNodeID()) .get("hbase-m").longValue()); Assert.assertEquals(1L, tm.getAllocationTagsWithCount(n2r2.getNodeID()) .get("hbase-m").longValue()); SchedulerNode schedulerNode0 =newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 =newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 =newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 =newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); TargetApplicationsNamespace namespace = new TargetApplicationsNamespace.AppID(application1); Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>(); PlacementConstraint constraint2 = PlacementConstraints .targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(), "hbase-m")) .build(); Set<String> srcTags2 = new HashSet<>(); srcTags2.add("app2"); constraintMap.put(srcTags2, constraint2); ts = System.currentTimeMillis(); ApplicationId application2 = BuilderUtils.newApplicationId(ts, 124); pcm.registerApplication(application2, constraintMap); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode3, pcm, tm)); TargetApplicationsNamespace self = new TargetApplicationsNamespace.Self(); PlacementConstraint constraint3 = PlacementConstraints .targetNotIn(NODE, allocationTagWithNamespace(self.toString(), "hbase-m")) .build(); Set<String> srcTags3 = new HashSet<>(); srcTags3.add("app3"); constraintMap.put(srcTags3, constraint3); ts = System.currentTimeMillis(); ApplicationId application3 = BuilderUtils.newApplicationId(ts, 124); pcm.registerApplication(application3, constraintMap); tm.addContainer(n0r1.getNodeID(), newContainerId(application3, 0), ImmutableSet.of("hbase-m")); Assert.assertFalse(PlacementConstraintsUtil .canSatisfyConstraints(application3, createSchedulingRequest(srcTags3), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil .canSatisfyConstraints(application3, createSchedulingRequest(srcTags3), schedulerNode1, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil .canSatisfyConstraints(application3, createSchedulingRequest(srcTags3), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil .canSatisfyConstraints(application3, createSchedulingRequest(srcTags3), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application3); }
@Test public void testInterAppConstriantsByAppTag() throws InvalidAllocationTagsQueryException { ApplicationId application1 = BuilderUtils.newApplicationId(1000, 123); ApplicationId application2 = BuilderUtils.newApplicationId(1001, 124); RMContext mockedContext = Mockito.spy(rmContext); ConcurrentMap<ApplicationId, RMApp> allApps = new ConcurrentHashMap<>(); allApps.put(application1, new MockRMApp(123, 1000, RMAppState.NEW, "userA", ImmutableSet.of("test-tag"))); allApps.put(application2, new MockRMApp(124, 1001, RMAppState.NEW, "userA", ImmutableSet.of(""))); when(mockedContext.getRMApps()).thenReturn(allApps); AllocationTagsManager tm = new AllocationTagsManager(mockedContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); mockedContext.setAllocationTagsManager(tm); mockedContext.setPlacementConstraintManager(pcm); RMNode n0r1 = rmNodes.get(0); RMNode n1r1 = rmNodes.get(1); RMNode n2r2 = rmNodes.get(2); RMNode n3r2 = rmNodes.get(3); tm.addContainer(n0r1.getNodeID(), newContainerId(application1), ImmutableSet.of("hbase-m")); tm.addContainer(n2r2.getNodeID(), newContainerId(application1), ImmutableSet.of("hbase-m")); SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); SchedulerNode schedulerNode1 = newSchedulerNode(n1r1.getHostName(), n1r1.getRackName(), n1r1.getNodeID()); SchedulerNode schedulerNode2 = newSchedulerNode(n2r2.getHostName(), n2r2.getRackName(), n2r2.getNodeID()); SchedulerNode schedulerNode3 = newSchedulerNode(n3r2.getHostName(), n3r2.getRackName(), n3r2.getNodeID()); TargetApplicationsNamespace namespace = new TargetApplicationsNamespace.AppTag("test-tag"); Map<Set<String>, PlacementConstraint> constraintMap = new HashMap<>(); PlacementConstraint constraint2 = PlacementConstraints .targetNotIn(NODE, allocationTagWithNamespace(namespace.toString(), "hbase-m")) .build(); Set<String> srcTags2 = ImmutableSet.of("app2"); constraintMap.put(srcTags2, constraint2); pcm.registerApplication(application2, constraintMap); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode0, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode1, pcm, tm)); Assert.assertFalse(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode2, pcm, tm)); Assert.assertTrue(PlacementConstraintsUtil.canSatisfyConstraints( application2, createSchedulingRequest(srcTags2), schedulerNode3, pcm, tm)); pcm.unregisterApplication(application1); pcm.unregisterApplication(application2); }
@Test public void testInvalidAllocationTagNamespace() { AllocationTagsManager tm = new AllocationTagsManager(rmContext); PlacementConstraintManagerService pcm = new MemoryPlacementConstraintManager(); rmContext.setAllocationTagsManager(tm); rmContext.setPlacementConstraintManager(pcm); long ts = System.currentTimeMillis(); ApplicationId application1 = BuilderUtils.newApplicationId(ts, 123); RMNode n0r1 = rmNodes.get(0); SchedulerNode schedulerNode0 = newSchedulerNode(n0r1.getHostName(), n0r1.getRackName(), n0r1.getNodeID()); PlacementConstraint constraint1 = PlacementConstraints .targetNotIn(NODE, allocationTagWithNamespace("unknown_namespace", "hbase-m")) .build(); Set<String> srcTags1 = new HashSet<>(); srcTags1.add("app1"); try { PlacementConstraintsUtil.canSatisfyConstraints(application1, createSchedulingRequest(srcTags1, constraint1), schedulerNode0, pcm, tm); Assert.fail("This should fail because we gave an invalid namespace"); } catch (Exception e) { Assert.assertTrue(e instanceof InvalidAllocationTagsQueryException); Assert.assertTrue(e.getMessage() .contains("Invalid namespace prefix: unknown_namespace")); } }