import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
private List<Resource> createResources(int cpus, int memory, String... ranges) {
  List<Resource> resources = Lists.newArrayList();

  if (cpus > 0) {
    resources.add(Resource.newBuilder().setType(Type.SCALAR).setName(MesosUtils.CPUS).setScalar(Scalar.newBuilder().setValue(cpus).build()).build());
  }

  if (memory > 0) {
    resources.add(Resource.newBuilder().setType(Type.SCALAR).setName(MesosUtils.MEMORY).setScalar(Scalar.newBuilder().setValue(memory).build()).build());
  }

  if (ranges.length > 0) {
    resources.add(buildPortRanges(ranges));
  }

  return resources;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testLiteralHostPortSelection() {
  String[] rangesNotOverlappingRequestedPorts = {"23:24", "25:25", "31:32", "50:51"};
  int numPorts = 1;
  List<Long> requestedPorts = Arrays.asList(50L, 51L);
  Resource resource = MesosUtils.getPortsResource(numPorts, buildOffer(rangesNotOverlappingRequestedPorts).getResourcesList(), requestedPorts);
  Assert.assertTrue(MesosUtils.getAllPorts(Collections.singletonList(resource)).containsAll(requestedPorts));
  Assert.assertEquals(numPorts + requestedPorts.size(), MesosUtils.getNumPorts(Collections.singletonList(resource)));

  String[] rangesOverlappingRequestPorts = {"23:28"};
  numPorts = 4;
  requestedPorts = Arrays.asList(25L, 27L);
  resource = MesosUtils.getPortsResource(numPorts, buildOffer(rangesOverlappingRequestPorts).getResourcesList(), requestedPorts);
  Assert.assertTrue(MesosUtils.getAllPorts(Collections.singletonList(resource)).containsAll(requestedPorts));
  Assert.assertEquals(numPorts + requestedPorts.size(), MesosUtils.getNumPorts(Collections.singletonList(resource)));
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
public static Resource buildPortRanges(String... ranges) {
  Resource.Builder resources = Resource.newBuilder()
      .setType(Type.RANGES)
      .setName(MesosUtils.PORTS);

  Ranges.Builder rangesBuilder = Ranges.newBuilder();

  for (String range : ranges) {
    String[] split = range.split("\\:");

    rangesBuilder.addRange(
        Range.newBuilder()
        .setBegin(Long.parseLong(split[0]))
        .setEnd(Long.parseLong(split[1])));
  }

  resources.setRanges(rangesBuilder);

  return resources.build();
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
public static boolean doesOfferMatchResources(Resources resources, List<Resource> offerResources, List<Long> otherRequestedPorts) {
  double numCpus = getNumCpus(offerResources);

  if (numCpus < resources.getCpus()) {
    return false;
  }

  double memory = getMemory(offerResources);

  if (memory < resources.getMemoryMb()) {
    return false;
  }

  int numPorts = getNumPorts(offerResources);

  if (numPorts < resources.getNumPorts()) {
    return false;
  }

  if (!getAllPorts(offerResources).containsAll(otherRequestedPorts)) {
    return false;
  }

  return true;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
/**
 * The resource passed in is the fully completed Resource which will be launched.  This may include volume/disk
 * information which is not appropriate for the RESERVE operation.  It is filtered out here.
 */
private static Resource getReservedResource(Resource resource) {
    // The resource passed in is the fully completed Resource which will be launched.  This may include volume/disk
    // information which is not appropriate for the RESERVE operation.  It is filtered out here.
    Resource.Builder resBuilder = Resource.newBuilder(resource);
    if (resBuilder.hasDisk() && resBuilder.getDisk().hasSource()) {
        // Mount volume: Copy disk, but without 'persistence' nor 'volume' fields
        resBuilder.setDisk(DiskInfo.newBuilder(resBuilder.getDisk())
                .clearPersistence()
                .clearVolume());
    } else {
        // Root volume: Clear the disk.
        resBuilder.clearDisk();
    }
    resBuilder.clearRevocable();
    return resBuilder.build();
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
public UnreserveOfferRecommendation(Offer offer, Resource resource) {
    this.offer = offer;
    Resource.Builder resourceBuilder = resource.toBuilder();

    // If non-root disk resource, we want to clear ALL fields except for the field indicating the disk source.
    if (resource.hasDisk() && resource.getDisk().hasSource()) {
        resource = resourceBuilder.setDisk(
                Resource.DiskInfo.newBuilder()
                        .setSource(resource.getDisk().getSource()))
                .build();
    } else {
        resource = resourceBuilder.clearDisk().clearRevocable().build();
    }

    this.operation = Operation.newBuilder()
            .setType(Operation.Type.UNRESERVE)
            .setUnreserve(Operation.Unreserve.newBuilder()
                    .addAllResources(Arrays.asList(resource)))
            .build();
    this.resource = resource;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@SuppressWarnings("deprecation") // for Resource.setRole()
private static Protos.Resource getUnreservedResource(String name, Protos.Value value, String role) {
    Protos.Resource.Builder resBuilder = Protos.Resource.newBuilder()
            .setName(name)
            .setType(value.getType())
            .setRole(role);
    if (!role.equals(Constants.ANY_ROLE)) {
        // Fill in the prereserved role info:
        resBuilder.addReservationsBuilder()
                .setRole(role)
                .setPrincipal(TestConstants.PRINCIPAL);
    }
    switch (value.getType()) {
        case SCALAR:
            return resBuilder.setScalar(value.getScalar()).build();
        case RANGES:
            return resBuilder.setRanges(value.getRanges()).build();
        case SET:
            return resBuilder.setSet(value.getSet()).build();
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testConsumeMultipleMountVolumesFailure() throws Exception {
    Resource offeredResource = ResourceTestUtils.getUnreservedMountVolume(2000);
    ResourceSet volumeResourceSet = DefaultResourceSet.newBuilder(TestConstants.ROLE, Constants.ANY_ROLE, TestConstants.PRINCIPAL)
            .id(TestConstants.RESOURCE_SET_ID)
            .cpus(1.0)
            .addVolume(VolumeSpec.Type.MOUNT.name(), 1000.0, TestConstants.CONTAINER_PATH + "-A")
            .addVolume(VolumeSpec.Type.MOUNT.name(), 1000.0, TestConstants.CONTAINER_PATH + "-B")
            .build();
    PodInstanceRequirement podInstanceRequirement =
            PodInstanceRequirementTestUtils.getRequirement(volumeResourceSet, 0);
    Protos.Offer offer = OfferTestUtils.getCompleteOffer(Arrays.asList(offeredResource));

    List<OfferRecommendation> recommendations = evaluator.evaluate(podInstanceRequirement, Arrays.asList(offer));
    Assert.assertEquals(0, recommendations.size());
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testAllocateRange() {
  List<Resource> resources = ImmutableList.<Resource>builder()
      .add(makePortResource(Optional.absent(), 80, 81, 90, 91, 92, 93))
      .add(makePortResource(TEST_ROLE, 100, 101))
      .build();
  AcceptedOffer acceptedOffer = AcceptedOffer.create(
      fakeOffer(resources),
      ResourceSlot.NONE,
      ResourceSlot.NONE,
      ImmutableSet.of(80, 90, 100),
      TaskTestUtil.DEV_TIER);

  List<Resource> expected = ImmutableList.<Resource>builder()
      // Because we prefer reserved resources and handle them before non-reserved resources,
      // result should have ports for the reserved resources first.
      .add(makePortResource(TEST_ROLE, 100))
      .add(makePortResource(Optional.absent(), 80, 90))
      .build();
  assertEquals(expected, acceptedOffer.getTaskResources());
  assertEquals(Collections.emptyList(), acceptedOffer.getExecutorResources());
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test(expected = Resources.InsufficientResourcesException.class)
public void testAllocateSingleRoleInsufficient() {
  List<Resource> resources = ImmutableList.<Resource>builder()
      // EXECUTOR_SLOT's CPU is not included here.
      .add(makeScalar(CPUS.getMesosName(), TEST_ROLE, false, TASK_SLOT.getNumCpus()))
      .add(makeScalar(
          RAM_MB.getMesosName(), TEST_ROLE, false, TOTAL_SLOT.getRam().as(Data.MB)))
      .add(makeScalar(
          DISK_MB.getMesosName(), TEST_ROLE, false, TOTAL_SLOT.getDisk().as(Data.MB)))
      .add(makePortResource(TEST_ROLE, TASK_PORTS))
      .build();
  Protos.Offer offer = fakeOffer(resources);

  AcceptedOffer.create(
      offer, TASK_SLOT, EXECUTOR_SLOT, TASK_PORTS_SET, new TierInfo(false, false));
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
private List<Resource> recordLaunchWithOfferedResources(
        Protos.Offer offer, PodInstanceRequirement podInstanceRequirement, Resource... offeredResources)
        throws InvalidRequirementException, IOException {
    List<OfferRecommendation> recommendations = evaluator.evaluate(
            podInstanceRequirement, Arrays.asList(offer));

    List<Resource> reservedResources = new ArrayList<>();
    for (OfferRecommendation recommendation : recommendations) {
        if (recommendation instanceof ReserveOfferRecommendation) {
            reservedResources.addAll(recommendation.getOperation().getReserve().getResourcesList());
        } else if (recommendation instanceof LaunchOfferRecommendation) {
            // DO NOT extract the TaskInfo from the Launch Operation. That version has a packed CommandInfo.
            stateStore.storeTasks(Arrays.asList(
                    ((LaunchOfferRecommendation) recommendation).getStoreableTaskInfo()));
        }
    }

    return reservedResources;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testReserveTaskDynamicPort() throws Exception {
    Protos.Resource offeredPorts = ResourceTestUtils.getUnreservedPorts(10000, 10000);
    PodInstanceRequirement podInstanceRequirement = PodInstanceRequirementTestUtils.getPortRequirement(0);

    List<OfferRecommendation> recommendations =
            evaluator.evaluate(podInstanceRequirement, OfferTestUtils.getCompleteOffers(offeredPorts));

    Assert.assertEquals(5, recommendations.size());

    Protos.Offer.Operation launchOperation = recommendations.get(4).getOperation();
    Protos.TaskInfo taskInfo = launchOperation.getLaunchGroup().getTaskGroup().getTasks(0);
    Protos.Resource fulfilledPortResource = taskInfo.getResources(0);
    Assert.assertFalse(getResourceId(fulfilledPortResource).isEmpty());

    Map<String, String> envvars = EnvUtils.toMap(
            TaskPackingUtils.unpack(taskInfo).getCommand().getEnvironment());
    Assert.assertEquals(envvars.toString(),
            String.valueOf(10000), envvars.get(TestConstants.PORT_ENV_NAME + "_0"));
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testLaunchTransient() {
    Resource resource = ResourceTestUtils.getUnreservedCpus(1.0);
    Offer offer = OfferTestUtils.getCompleteOffer(resource);
    TaskInfo.Builder taskInfoBuilder = TaskTestUtils.getTaskInfo(resource).toBuilder();
    taskInfoBuilder.setLabels(new TaskLabelWriter(taskInfoBuilder).setTransient().toProto());

    TestOperationRecorder recorder = new TestOperationRecorder();
    OfferAccepter accepter = new OfferAccepter(Arrays.asList(recorder));
    accepter.accept(
            driver,
            Arrays.asList(new LaunchOfferRecommendation(
                    offer,
                    taskInfoBuilder.build(),
                    Protos.ExecutorInfo.newBuilder().setExecutorId(TestConstants.EXECUTOR_ID).build(),
                    false,
                    true)));
    Assert.assertEquals(1, recorder.getLaunches().size());
    verify(driver, times(0)).acceptOffers(
            anyCollectionOf(OfferID.class),
            anyCollectionOf(Operation.class),
            anyObject());
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
private static Resource.Builder setValue(Resource.Builder builder, Value value) {
    builder.setType(value.getType());
    switch (value.getType()) {
    case SCALAR:
        builder.setScalar(value.getScalar());
        return builder;
    case RANGES:
        builder.setRanges(value.getRanges());
        return builder;
    case SET:
        builder.setSet(value.getSet());
        return builder;
    default:
        throw new IllegalArgumentException(String.format("Unsupported spec value type: %s", value.getType()));
    }
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
public Optional<MesosResource> consumeReservableMerged(String name, Value desiredValue, String preReservedRole) {
    Map<String, Value> pool = reservableMergedPoolByRole.get(preReservedRole);
    if (pool == null) {
        logger.info("No unreserved resources available for role '{}'. Reservable roles are: {}",
                preReservedRole, reservableMergedPoolByRole.keySet());
        return Optional.empty();
    }

    Value availableValue = pool.get(name);

    if (sufficientValue(desiredValue, availableValue)) {
        pool.put(name, ValueUtils.subtract(availableValue, desiredValue));
        reservableMergedPoolByRole.put(preReservedRole, pool);

        Resource.Builder builder = ResourceBuilder.fromUnreservedValue(name, desiredValue).build().toBuilder();
        if (Capabilities.getInstance().supportsPreReservedResources() &&
                !preReservedRole.equals(Constants.ANY_ROLE)) {
            builder.addReservations(
                    Resource.ReservationInfo.newBuilder()
                            .setRole(preReservedRole)
                            .setType(Resource.ReservationInfo.Type.STATIC));
        }

        return Optional.of(new MesosResource(builder.build()));
    } else {
        if (availableValue == null) {
            logger.info("Offer lacks any unreserved {} resources for role {}", name, preReservedRole);
        } else {
            logger.info("Offered quantity of {} for role {} is insufficient: desired {}, offered {}",
                    name,
                    preReservedRole,
                    TextFormat.shortDebugString(desiredValue),
                    TextFormat.shortDebugString(availableValue));
        }
        return Optional.empty();
    }
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testConsumeReservedMergedResource() {
    Resource resource = ResourceTestUtils.getReservedCpus(1.0, TestConstants.RESOURCE_ID);
    Protos.Value resourceValue = ValueUtils.getValue(resource);
    String resourceId = ResourceTestUtils.getResourceId(resource);
    Offer offer = OfferTestUtils.getOffer(resource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getDynamicallyReservedPoolByResourceId().size());
    MesosResource resourceToConsume = pool.consumeReserved(resource.getName(), resourceValue, resourceId).get();
    Assert.assertEquals(resource, resourceToConsume.getResource());
    Assert.assertEquals(0, pool.getDynamicallyReservedPoolByResourceId().size());
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
private static Collection<MesosResource> getMesosResources(Offer offer, Optional<String> role) {
    Collection<MesosResource> mesosResources = new ArrayList<MesosResource>();
    for (Resource resource : offer.getResourcesList()) {
        if (consumableResource(role, resource)) {
            mesosResources.add(new MesosResource(resource));
        }
    }

    return mesosResources;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@SuppressWarnings("PMD.AvoidUsingHardCodedIP")
@Test
public void testReserveTaskNamedVIPPort() throws Exception {
    List<OfferRecommendation> recommendations = evaluator.evaluate(
            PodInstanceRequirementTestUtils.getVIPRequirement(80, 10000),
            OfferTestUtils.getCompleteOffers(ResourceTestUtils.getUnreservedPorts(10000, 10000)));

    Assert.assertEquals(5, recommendations.size());

    Operation launchOperation = recommendations.get(4).getOperation();
    TaskInfo taskInfo = launchOperation.getLaunchGroup().getTaskGroup().getTasks(0);
    Resource fulfilledPortResource = taskInfo.getResources(0);
    Assert.assertEquals(10000, fulfilledPortResource.getRanges().getRange(0).getBegin());
    Assert.assertFalse(getResourceId(fulfilledPortResource).isEmpty());

    DiscoveryInfo discoveryInfo = taskInfo.getDiscovery();
    Assert.assertEquals(discoveryInfo.getName(), taskInfo.getName());
    Assert.assertEquals(discoveryInfo.getVisibility(), DiscoveryInfo.Visibility.CLUSTER);

    Port discoveryPort = discoveryInfo.getPorts().getPorts(0);
    Assert.assertEquals(discoveryPort.getProtocol(), "tcp");
    Assert.assertEquals(discoveryPort.getVisibility(), DiscoveryInfo.Visibility.EXTERNAL);
    Assert.assertEquals(discoveryPort.getNumber(), 10000);
    Label vipLabel = discoveryPort.getLabels().getLabels(0);
    Assert.assertTrue(vipLabel.getKey().startsWith("VIP_"));
    Assert.assertEquals(vipLabel.getValue(), TestConstants.VIP_NAME + "-10000:80");

    Map<String, String> envvars = EnvUtils.toMap(
            TaskPackingUtils.unpack(taskInfo).getCommand().getEnvironment());
    Assert.assertEquals(String.valueOf(10000), envvars.get(TestConstants.PORT_ENV_NAME + "_VIP_10000"));
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
public static List<Long> getAllPorts(List<Resource> resources) {
  Ranges ranges = getRanges(resources, PORTS);

  final List<Long> ports = Lists.newArrayList();
  if (ranges != null) {
    for (Range range : ranges.getRangeList()) {
      for (long port = range.getBegin(); port <= range.getEnd(); port++) {
        ports.add(port);
      }
    }
  }

  return ports;
}
 

import org.apache.mesos.Protos.Resource; //导入依赖的package包/类
@Test
public void testUpdateDynamicToStaticPort() throws Exception {
    // Launch for the first time: get port 555 from dynamic port
    Resource reserveResource = recordLaunchWithCompleteOfferedResources(
            PodInstanceRequirementTestUtils.getPortRequirement(0),
            ResourceTestUtils.getUnreservedPorts(555, 555)).get(0);
    String resourceId = getResourceId(reserveResource);
    Collection<Resource> expectedResources = getExpectedExecutorResources(
            stateStore.fetchTasks().iterator().next().getExecutor());
    expectedResources.addAll(Arrays.asList(
            ResourceTestUtils.getReservedPorts(555, 555, resourceId),
            ResourceTestUtils.getUnreservedPorts(666, 666)));

    // Now lets move to port 666:
    List<OfferRecommendation> recommendations = evaluator.evaluate(
            PodInstanceRequirementTestUtils.getPortRequirement(666),
            Arrays.asList(OfferTestUtils.getOffer(expectedResources)));

    // RESERVE, UNRESERVE, LAUNCH
    Assert.assertEquals(recommendations.toString(), 3, recommendations.size());
    Assert.assertEquals(Operation.Type.UNRESERVE, recommendations.get(0).getOperation().getType());
    Assert.assertEquals(Operation.Type.RESERVE, recommendations.get(1).getOperation().getType());
    Assert.assertEquals(Operation.Type.LAUNCH_GROUP, recommendations.get(2).getOperation().getType());

    Operation launchOperation = recommendations.get(2).getOperation();
    TaskInfo taskInfo = launchOperation.getLaunchGroup().getTaskGroup().getTasks(0);

    Map<String, String> envvars = EnvUtils.toMap(
            TaskPackingUtils.unpack(taskInfo).getCommand().getEnvironment());
    Assert.assertEquals(String.valueOf(666), envvars.get(TestConstants.PORT_ENV_NAME + "_666"));
}