DistributedOpportunisticContainerAllocator extends OpportunisticContainerAllocator { @Override public List<Container> allocateContainers(ResourceBlacklistRequest blackList, List<ResourceRequest> oppResourceReqs, ApplicationAttemptId applicationAttemptId, OpportunisticContainerContext opportContext, long rmIdentifier, String appSubmitter) throws YarnException { updateBlacklist(blackList, opportContext); opportContext.addToOutstandingReqs(oppResourceReqs); Set<String> nodeBlackList = new HashSet<>(opportContext.getBlacklist()); Set<String> allocatedNodes = new HashSet<>(); List<Container> allocatedContainers = new ArrayList<>(); boolean continueLoop = true; while (continueLoop) { continueLoop = false; List<Map<Resource, List<Allocation>>> allocations = new ArrayList<>(); for (SchedulerRequestKey schedulerKey : opportContext.getOutstandingOpReqs().descendingKeySet()) { int remAllocs = -1; int maxAllocationsPerAMHeartbeat = getMaxAllocationsPerAMHeartbeat(); if (maxAllocationsPerAMHeartbeat > 0) { remAllocs = maxAllocationsPerAMHeartbeat - allocatedContainers.size() - getTotalAllocations(allocations); if (remAllocs <= 0) { LOG.info("Not allocating more containers as we have reached max " + "allocations per AM heartbeat {}", maxAllocationsPerAMHeartbeat); break; } } Map<Resource, List<Allocation>> allocation = allocate( rmIdentifier, opportContext, schedulerKey, applicationAttemptId, appSubmitter, nodeBlackList, allocatedNodes, remAllocs); if (allocation.size() > 0) { allocations.add(allocation); continueLoop = true; } } matchAllocation(allocations, allocatedContainers, opportContext); } return allocatedContainers; } DistributedOpportunisticContainerAllocator(
      BaseContainerTokenSecretManager tokenSecretManager);  DistributedOpportunisticContainerAllocator(
      BaseContainerTokenSecretManager tokenSecretManager,
      int maxAllocationsPerAMHeartbeat); @Override List<Container> allocateContainers(ResourceBlacklistRequest blackList,
      List<ResourceRequest> oppResourceReqs,
      ApplicationAttemptId applicationAttemptId,
      OpportunisticContainerContext opportContext, long rmIdentifier,
      String appSubmitter);  }
@Test public void testSimpleAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList(ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Assert.assertEquals(1, containers.size()); Assert.assertEquals(0, oppCntxt.getOutstandingOpReqs().size()); }
@Test public void testBlacklistRejection() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( Arrays.asList("h1", "h2"), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList(ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Assert.assertEquals(0, containers.size()); Assert.assertEquals(1, oppCntxt.getOutstandingOpReqs().size()); }
@Test public void testRoundRobinSimpleAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newBuilder().allocationRequestId(1) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(3) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); LOG.info("Containers: {}", containers); Set<String> allocatedHosts = new HashSet<>(); for (Container c : containers) { allocatedHosts.add(c.getNodeHttpAddress()); } Assert.assertTrue(allocatedHosts.contains("h1:1234")); Assert.assertTrue(allocatedHosts.contains("h2:1234")); Assert.assertTrue(allocatedHosts.contains("h3:1234")); Assert.assertEquals(3, containers.size()); }
@Test public void testNodeLocalAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newBuilder().allocationRequestId(1) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(3) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(3) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(3) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); LOG.info("Containers: {}", containers); Assert.assertEquals(3, containers.size()); for (Container c : containers) { if (c.getAllocationRequestId() == 2 || c.getAllocationRequestId() == 3) { Assert.assertEquals("h1:1234", c.getNodeHttpAddress()); } } }
@Test public void testNodeLocalAllocationSameSchedKey() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newBuilder().allocationRequestId(2) .numContainers(2) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .numContainers(2) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .numContainers(2) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); LOG.info("Containers: {}", containers); Set<String> allocatedHosts = new HashSet<>(); for (Container c : containers) { allocatedHosts.add(c.getNodeHttpAddress()); } Assert.assertEquals(2, containers.size()); Assert.assertTrue(allocatedHosts.contains("h1:1234")); Assert.assertFalse(allocatedHosts.contains("h2:1234")); Assert.assertFalse(allocatedHosts.contains("h3:1234")); }
@Test public void testSimpleRackLocalAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h1", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r1", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Set<String> allocatedHosts = new HashSet<>(); for (Container c : containers) { allocatedHosts.add(c.getNodeHttpAddress()); } Assert.assertTrue(allocatedHosts.contains("h2:1234")); Assert.assertFalse(allocatedHosts.contains("h3:1234")); Assert.assertFalse(allocatedHosts.contains("h4:1234")); Assert.assertEquals(1, containers.size()); }
@Test public void testRoundRobinRackLocalAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newBuilder().allocationRequestId(1) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(1) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(1) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build(), ResourceRequest.newBuilder().allocationRequestId(2) .priority(PRIORITY_NORMAL) .resourceName(ResourceRequest.ANY) .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Set<String> allocatedHosts = new HashSet<>(); for (Container c : containers) { allocatedHosts.add(c.getNodeHttpAddress()); } LOG.info("Containers: {}", containers); Assert.assertTrue(allocatedHosts.contains("h2:1234")); Assert.assertTrue(allocatedHosts.contains("h5:1234")); Assert.assertFalse(allocatedHosts.contains("h3:1234")); Assert.assertFalse(allocatedHosts.contains("h4:1234")); Assert.assertEquals(2, containers.size()); }
@Test public void testRoundRobinRackLocalAllocationSameSchedKey() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h1", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r1", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Set<String> allocatedHosts = new HashSet<>(); for (Container c : containers) { allocatedHosts.add(c.getNodeHttpAddress()); } LOG.info("Containers: {}", containers); Assert.assertTrue(allocatedHosts.contains("h2:1234")); Assert.assertTrue(allocatedHosts.contains("h5:1234")); Assert.assertFalse(allocatedHosts.contains("h3:1234")); Assert.assertFalse(allocatedHosts.contains("h4:1234")); Assert.assertEquals(2, containers.size()); }
@Test public void testOffSwitchAllocationWhenNoNodeOrRack() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h6", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r3", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); LOG.info("Containers: {}", containers); Assert.assertEquals(2, containers.size()); }
@Test public void testLotsOfContainersRackLocalAllocationSameSchedKey() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 1000, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h1", CAPABILITY_1GB, 1000, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r1", CAPABILITY_1GB, 1000, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = new ArrayList<>(); for (int i = 0; i < 250; i++) { containers.addAll(allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser")); } Assert.assertEquals(1000, containers.size()); }
@Test public void testLotsOfContainersRackLocalAllocation() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = new ArrayList<>(); for (int i = 0; i < 100; i++) { reqs.add(ResourceRequest.newBuilder().allocationRequestId(i + 1) .priority(PRIORITY_NORMAL) .resourceName("*") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); reqs.add(ResourceRequest.newBuilder().allocationRequestId(i + 1) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); reqs.add(ResourceRequest.newBuilder().allocationRequestId(i + 1) .priority(PRIORITY_NORMAL) .resourceName("/r1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); } ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = new ArrayList<>(); for (int i = 0; i < 25; i++) { containers.addAll(allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser")); } Assert.assertEquals(100, containers.size()); }
@Test public void testAllocationWithNodeLabels() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); List<ResourceRequest> reqs = Arrays.asList(ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 1, true, "label", OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Assert.assertEquals(0, containers.size()); Assert.assertEquals(1, oppCntxt.getOutstandingOpReqs().size()); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1", "label"))); containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); Assert.assertEquals(1, containers.size()); Assert.assertEquals(0, oppCntxt.getOutstandingOpReqs().size()); }
@Test public void testMaxAllocationsPerAMHeartbeat() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); allocator.setMaxAllocationsPerAMHeartbeat(2); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 3, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h6", CAPABILITY_1GB, 3, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r3", CAPABILITY_1GB, 3, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "user1"); LOG.info("Containers: {}", containers); Assert.assertEquals(2, containers.size()); containers = allocator.allocateContainers( blacklistRequest, new ArrayList<>(), appAttId, oppCntxt, 1L, "user1"); LOG.info("Containers: {}", containers); Assert.assertEquals(1, containers.size()); }
@Test public void testMaxAllocationsPerAMHeartbeatDifferentSchedKey() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); allocator.setMaxAllocationsPerAMHeartbeat(2); final ExecutionTypeRequest oppRequest = ExecutionTypeRequest.newInstance( ExecutionType.OPPORTUNISTIC, true); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(Priority.newInstance(1), "*", CAPABILITY_1GB, 1, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(Priority.newInstance(2), "h6", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(Priority.newInstance(3), "/r3", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "user1"); LOG.info("Containers: {}", containers); Assert.assertEquals(2, containers.size()); containers = allocator.allocateContainers( blacklistRequest, new ArrayList<>(), appAttId, oppCntxt, 1L, "user1"); LOG.info("Containers: {}", containers); Assert.assertEquals(2, containers.size()); containers = allocator.allocateContainers( blacklistRequest, new ArrayList<>(), appAttId, oppCntxt, 1L, "user1"); LOG.info("Containers: {}", containers); Assert.assertEquals(1, containers.size()); }
@Test public void testMaxAllocationsPerAMHeartbeatWithNoLimit() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); allocator.setMaxAllocationsPerAMHeartbeat(-1); List<ResourceRequest> reqs = new ArrayList<>(); for (int i = 0; i < 20; i++) { reqs.add(ResourceRequest.newBuilder().allocationRequestId(i + 1) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); } ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "user1"); Assert.assertEquals(20, containers.size()); }
@Test public void testMaxAllocationsPerAMHeartbeatWithHighLimit() throws Exception { ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( new ArrayList<>(), new ArrayList<>()); allocator.setMaxAllocationsPerAMHeartbeat(100); List<ResourceRequest> reqs = new ArrayList<>(); for (int i = 0; i < 20; i++) { reqs.add(ResourceRequest.newBuilder().allocationRequestId(i + 1) .priority(PRIORITY_NORMAL) .resourceName("h1") .capability(CAPABILITY_1GB) .relaxLocality(true) .executionType(ExecutionType.OPPORTUNISTIC).build()); } ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h1", 1234), "h1:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "user1"); Assert.assertEquals(20, containers.size()); }
@Test public void testAllocationLatencyMetrics() throws Exception { oppCntxt = spy(oppCntxt); OpportunisticSchedulerMetrics metrics = mock(OpportunisticSchedulerMetrics.class); when(oppCntxt.getOppSchedulerMetrics()).thenReturn(metrics); ResourceBlacklistRequest blacklistRequest = ResourceBlacklistRequest.newInstance( Collections.emptyList(), Collections.emptyList()); List<ResourceRequest> reqs = Arrays.asList( ResourceRequest.newInstance(PRIORITY_NORMAL, "*", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "h6", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ), ResourceRequest.newInstance(PRIORITY_NORMAL, "/r3", CAPABILITY_1GB, 2, true, null, OPPORTUNISTIC_REQ)); ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance( ApplicationId.newInstance(0L, 1), 1); oppCntxt.updateNodeList( Arrays.asList( RemoteNode.newInstance( NodeId.newInstance("h3", 1234), "h3:1234", "/r2"), RemoteNode.newInstance( NodeId.newInstance("h2", 1234), "h2:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h5", 1234), "h5:1234", "/r1"), RemoteNode.newInstance( NodeId.newInstance("h4", 1234), "h4:1234", "/r2"))); List<Container> containers = allocator.allocateContainers( blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser"); LOG.info("Containers: {}", containers); Assert.assertEquals(2, containers.size()); verify(metrics, times(2)).addAllocateOLatencyEntry(anyLong()); }