/**
 * Creates string out an offer in a nice format.
 * @param resource - mesos resource to make into string.
 * @return string representing a resource.
 */
public static String getResourceMessage(Resource resource) {
    Type type = resource.getType();
    String ret = resource.getName() +" "+resource.getRole()+ ": ";
    switch (type) {
        case SCALAR:
            ret += resource.getScalar().getValue();
            break;
        case RANGES:
            for (Range range : resource.getRanges().getRangeList())
                ret += range.getBegin() + " - "+range.getEnd()+",";
            break;
        case TEXT:
            ret += " TEXT type...cannot find.";
            break;
        case SET:
            for (String string : resource.getSet().getItemList())
                ret += string + ",";
            break;
    }
    return ret;
}
 

@SuppressWarnings("deprecation")
private void freeAtomicResource(MesosResource mesosResource) {
  Resource.Builder resBuilder = Resource.newBuilder(mesosResource.getResource());
  resBuilder.clearReservation();
  resBuilder.setRole(Constants.ANY_ROLE);

  if (resBuilder.hasDisk()) {
    Resource.DiskInfo.Builder diskBuilder = Resource.DiskInfo.newBuilder(resBuilder.getDisk());
    diskBuilder.clearPersistence();
    diskBuilder.clearVolume();
    resBuilder.setDisk(diskBuilder.build());
  }

  Resource releasedResource = resBuilder.build();

  List<MesosResource> resList = unreservedAtomicPool.get(mesosResource.getName());
  if (resList == null) {
    resList = new ArrayList<>();
  }

  resList.add(new MesosResource(releasedResource));
  unreservedAtomicPool.put(mesosResource.getName(), resList);
}
 

public static Value getValue(Resource resource) {
  Type type = resource.getType();
  Value.Builder builder = Value.newBuilder();
  builder.setType(type);

  switch (type) {
    case SCALAR:
      return builder.setScalar(resource.getScalar()).build();
    case RANGES:
      return builder.setRanges(resource.getRanges()).build();
    case SET:
      return builder.setSet(resource.getSet()).build();
    default:
      throw new IllegalArgumentException(String.format("Unsupported value type %s in resource %s",
          type,
          TextFormat.shortDebugString(resource)));
  }
}
 

public UnreserveOfferRecommendation(Offer offer, Resource resource) {
  this.offer = offer;
  Resource.Builder resourceBuilder = resource.toBuilder();

  Resource resourceR;
  // 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()) {
    resourceR = resourceBuilder
        .setDisk(Resource.DiskInfo.newBuilder().setSource(resource.getDisk().getSource()))
        .build();
  } else {
    resourceR = resourceBuilder.clearDisk().clearRevocable().build();
  }

  this.operation = Operation.newBuilder()
      .setType(Operation.Type.UNRESERVE)
      .setUnreserve(
          Operation.Unreserve.newBuilder().addAllResources(Collections.singletonList(resourceR))
      )
      .build();
  this.resource = resourceR;
}
 

private Iterator<Long> getPortMappingIterator(TaskInfo taskInfo) {
    List<Resource> list = taskInfo.getResourcesList();
    List<Long> ports = new ArrayList<Long>();
    for (Resource resource : list) {
        String name = resource.getName();
        if ("ports".equals(name)) {
            Ranges ranges = resource.getRanges();
            for (Range range : ranges.getRangeList()) {
                long startPort = range.getBegin();
                long endPort = range.getEnd();
                for (int i = 0; i <= endPort - startPort; i++) {
                    ports.add(startPort + i);
                }
            }
        }
    }
    return ports.iterator();
}
 

@Test
public void testConsumeMultipleMountVolumesFailure() throws Exception {
    Resource offeredResource = ResourceTestUtils.getUnreservedMountVolume(2000, Optional.empty());
    ResourceSet volumeResourceSet = DefaultResourceSet.newBuilder(TestConstants.ROLE, Constants.ANY_ROLE, TestConstants.PRINCIPAL)
            .id(TestConstants.RESOURCE_SET_ID)
            .cpus(1.0)
            .addMountVolume(1000.0, TestConstants.CONTAINER_PATH + "-A", Collections.emptyList())
            .addMountVolume(1000.0, TestConstants.CONTAINER_PATH + "-B", Collections.emptyList())
            .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());
}
 

/**
 * Builds a TaskInfo from the given jobspec
 * @param job - JobSpec to TaskInfo-ify
 * @param offer - offer add extra data (SlaveId)
 * @return TaskInfo fully formed
 */
private TaskInfo getTaskInfo(JobSpec job,Offer offer) {
    TaskID taskId = TaskID.newBuilder().setValue(job.getJob().getId()).build();
    TaskInfo info = TaskInfo.newBuilder().setName("task " + taskId.getValue())
             .setTaskId(taskId)
             .setSlaveId(offer.getSlaveId())
             .addResources(Resource.newBuilder()
                           .setName("cpus")
                           .setType(Value.Type.SCALAR)
                           .setScalar(Value.Scalar.newBuilder().setValue(job.getJob().getLoadValue()*1.0)))
             .addResources(Resource.newBuilder()
                           .setName("mem")
                           .setType(Value.Type.SCALAR)
                           .setScalar(Value.Scalar.newBuilder().setValue(job.getJob().getLoadValue()*1024.0)))
             .setExecutor(ExecutorInfo.newBuilder(executor)).setData(MesosUtilities.jobSpecToByteString(job)).build();
    return info;
}
 

@SuppressWarnings("deprecation") // for Resource.setRole()
private static Protos.Resource.Builder addReservation(
        Protos.Resource.Builder builder, String resourceId) {
    Protos.Resource.ReservationInfo.Builder reservationBuilder;
    if (Capabilities.getInstance().supportsPreReservedResources()) {
        reservationBuilder = builder.addReservationsBuilder()
                .setRole(TestConstants.ROLE)
                .setPrincipal(TestConstants.PRINCIPAL);
    } else {
        builder.setRole(TestConstants.ROLE);
        reservationBuilder = builder.getReservationBuilder()
                .setPrincipal(TestConstants.PRINCIPAL);
    }
    AuxLabelAccess.setResourceId(reservationBuilder, resourceId);
    AuxLabelAccess.setResourceNamespace(reservationBuilder, TestConstants.SERVICE_NAME);
    AuxLabelAccess.setFrameworkId(reservationBuilder, TestConstants.FRAMEWORK_ID.getValue());
    return builder;
}
 

/**
 * Convert offer to hostResources
 */
HostResources parseOffer(Offer offer) {
	HostResources hr = new HostResources();

	hr.setMem(0);
	hr.setCpus(0);

	for (Resource r : offer.getResourcesList()) {
		String resourceName = r.getName();
		int value = (int) r.getScalar().getValue();

		switch (resourceName) {
		case OFFER_MEM:
			hr.setMem(MB * value);
			break;
		case OFFER_CPUS:
			hr.setCpus(value);
			break;
		}
	}

	return hr;
}
 

@Test
public void testConsumeUnreservedAtomicResource() {
    Resource offerResource = ResourceTestUtils.getUnreservedMountVolume(1000, Optional.empty());
    VolumeSpec spec = DefaultVolumeSpec.createMountVolume(
            ValueUtils.getValue(offerResource).getScalar().getValue(),
            TestConstants.CONTAINER_PATH,
            Collections.emptyList(),
            TestConstants.ROLE,
            Constants.ANY_ROLE,
            TestConstants.PRINCIPAL);
    Offer offer = OfferTestUtils.getOffer(offerResource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getUnreservedAtomicPool().size());
    MesosResource resourceToConsume = pool.consumeAtomic(offerResource.getName(), spec).get();
    Assert.assertEquals(offerResource, resourceToConsume.getResource());
    Assert.assertEquals(0, pool.getUnreservedAtomicPool().size());
}
 

@Test
public void testConsumeUnreservedAtomicResourceMatchedProfile() {
    Resource offerResource = ResourceTestUtils.getUnreservedMountVolume(1000, Optional.of("bar"));
    VolumeSpec spec = DefaultVolumeSpec.createMountVolume(
            ValueUtils.getValue(offerResource).getScalar().getValue(),
            TestConstants.CONTAINER_PATH,
            Arrays.asList("foo", "bar"),
            TestConstants.ROLE,
            Constants.ANY_ROLE,
            TestConstants.PRINCIPAL);
    Offer offer = OfferTestUtils.getOffer(offerResource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getUnreservedAtomicPool().size());
    MesosResource resourceToConsume = pool.consumeAtomic(offerResource.getName(), spec).get();
    Assert.assertEquals(offerResource, resourceToConsume.getResource());
    Assert.assertEquals(0, pool.getUnreservedAtomicPool().size());
}
 

@Test
public void testConsumeUnreservedAtomicResourceUnmatchedProfile1() {
    Resource offerResource = ResourceTestUtils.getUnreservedMountVolume(1000, Optional.of("bar"));
    VolumeSpec spec = DefaultVolumeSpec.createMountVolume(
            ValueUtils.getValue(offerResource).getScalar().getValue(),
            TestConstants.CONTAINER_PATH,
            Arrays.asList("foo"),
            TestConstants.ROLE,
            Constants.ANY_ROLE,
            TestConstants.PRINCIPAL);
    Offer offer = OfferTestUtils.getOffer(offerResource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getUnreservedAtomicPool().size());
    Assert.assertFalse(pool.consumeAtomic(offerResource.getName(), spec).isPresent());
}
 

@Test
public void testConsumeUnreservedAtomicResourceUnmatchedProfile3() {
    Resource offerResource = ResourceTestUtils.getUnreservedMountVolume(1000, Optional.of("bar"));
    VolumeSpec spec = DefaultVolumeSpec.createMountVolume(
            ValueUtils.getValue(offerResource).getScalar().getValue(),
            TestConstants.CONTAINER_PATH,
            Collections.emptyList(),
            TestConstants.ROLE,
            Constants.ANY_ROLE,
            TestConstants.PRINCIPAL);
    Offer offer = OfferTestUtils.getOffer(offerResource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getUnreservedAtomicPool().size());
    Assert.assertFalse(pool.consumeAtomic(offerResource.getName(), spec).isPresent());
}
 

private Resource createMesosRangeResource(ResourceType resourceType, RangeResourceEntry rangeResourceEntry) {
  Ranges.Builder rangesBuilder = Ranges.newBuilder();
  Range rangeBuilder = Range.newBuilder()
                            .setBegin(rangeResourceEntry.getBegin())
                            .setEnd(rangeResourceEntry.getEnd()).build();
  rangesBuilder.addRange(rangeBuilder);

  Resource.Builder resourceBuilder = Resource.newBuilder()
                                             .setName(resourceType.toString())
                                             .setType(Protos.Value.Type.RANGES)
                                             .setRanges(rangesBuilder.build());
  if (rangeResourceEntry.getReservationType() != null && rangeResourceEntry.getReservationType().equals(ReservationType.STATIC)) {
    resourceBuilder.setRole(MesosCommon.getRole(mesosStormConf));
  }
  return resourceBuilder.build();
}
 

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;
}
 

private static boolean consumableResource(Optional<String> podRole, Resource resource) {
  if (!podRole.isPresent()
      || !resource.hasAllocationInfo()
      || !resource.getAllocationInfo().hasRole())
  {
    return true;
  }

  String allocationRole = resource.getAllocationInfo().getRole();
  return podRole.get().equals(allocationRole);
}
 

public Optional<MesosResource> consumeReserved(String name, Value value, String resourceId) {
  MesosResource mesosResource = dynamicallyReservedPoolByResourceId.get(resourceId);

  if (mesosResource != null) {
    if (mesosResource.isAtomic()) {
      if (sufficientValue(value, mesosResource.getValue())) {
        dynamicallyReservedPoolByResourceId.remove(resourceId);
      } else {
        LOGGER.warn("Reserved atomic quantity of {} is insufficient: desired {}, reserved {}",
            name,
            TextFormat.shortDebugString(value),
            TextFormat.shortDebugString(mesosResource.getValue()));
        return Optional.empty();
      }
    } else {
      Value availableValue = mesosResource.getValue();
      if (ValueUtils.compare(availableValue, value) > 0) {
        // Update the value in pool with the remaining unclaimed resource amount
        Resource remaining = ResourceBuilder.fromExistingResource(mesosResource.getResource())
            .setValue(ValueUtils.subtract(availableValue, value))
            .build();
        dynamicallyReservedPoolByResourceId.put(resourceId, new MesosResource(remaining));
        // Return only the claimed resource amount from this reservation
      } else {
        dynamicallyReservedPoolByResourceId.remove(resourceId);
      }
    }
  } else {
    LOGGER.warn("Failed to find reserved {} resource with ID: {}. Reserved resource IDs are: {}",
        name,
        resourceId,
        dynamicallyReservedPoolByResourceId.keySet());
  }

  return Optional.ofNullable(mesosResource);
}
 

private List<Offer> generateOffers(List<Resource> resources) {
  FrameworkID fidOne = Protos.FrameworkID.newBuilder().setValue("framework-1").build();
  FrameworkID fidTwo = Protos.FrameworkID.newBuilder().setValue("framework-2").build();
  FrameworkID fidThree = Protos.FrameworkID.newBuilder().setValue("framework-3").build();
  FrameworkID fidFour = Protos.FrameworkID.newBuilder().setValue("framework-4").build();

  OfferID oidOne = Protos.OfferID.newBuilder().setValue("offer-1").build();
  OfferID oidTwo = Protos.OfferID.newBuilder().setValue("offer-2").build(); 
  OfferID oidThree = Protos.OfferID.newBuilder().setValue("offer-3").build(); 
  OfferID oidFour = Protos.OfferID.newBuilder().setValue("offer-4").build(); 

  SlaveID sidOne = Protos.SlaveID.newBuilder().setValue("slave-1").build();
  SlaveID sidTwo = Protos.SlaveID.newBuilder().setValue("slave-2").build();
  SlaveID sidThree = Protos.SlaveID.newBuilder().setValue("slave-3").build();
  SlaveID sidFour = Protos.SlaveID.newBuilder().setValue("slave-4").build();

  Offer offerOne = Protos.Offer.newBuilder().setFrameworkId(fidOne).setHostname("10.0.0.1").setId(oidOne).setSlaveId(sidOne).
      addResources(resources.get(0)).addResources(resources.get(1)).build();
  Offer offerTwo = Protos.Offer.newBuilder().setFrameworkId(fidTwo).setHostname("10.0.0.2").setId(oidTwo).setSlaveId(sidTwo).
      addResources(resources.get(2)).addResources(resources.get(3)).build();
  Offer offerThree = Protos.Offer.newBuilder().setFrameworkId(fidThree).setHostname("10.0.0.3").setId(oidThree).setSlaveId(sidThree).
      addResources(resources.get(0)).addResources(resources.get(3)).build();
  Offer offerFour = Protos.Offer.newBuilder().setFrameworkId(fidFour).setHostname("10.0.0.4").setId(oidFour).setSlaveId(sidFour).
      addResources(resources.get(2)).addResources(resources.get(1)).build();

  return Lists.newArrayList(offerOne, offerTwo, offerThree, offerFour);
}
 

public static Protos.Resource getReservedMountVolume(
        double diskSize,
        Optional<String> profile,
        String resourceId,
        String persistenceId) {
    Protos.Resource.Builder builder = getUnreservedMountVolume(diskSize, profile).toBuilder();
    builder.getDiskBuilder().getPersistenceBuilder()
            .setId(persistenceId)
            .setPrincipal(TestConstants.PRINCIPAL);
    builder.getDiskBuilder().getVolumeBuilder()
            .setContainerPath(TestConstants.CONTAINER_PATH)
            .setMode(Volume.Mode.RW);
    return addReservation(builder, resourceId).build();
}
 

@Override
public int compare(Resource left, Resource right) {
  if (left.getRole().equals(right.getRole())) {
    return 0;
  } else if (left.getRole().equals("*")) {
    return 1;
  }
  return -1;
}
 

@Test
public void testFailToCreateVolumeWithWrongResource() throws Exception {
    Resource wrongOfferedResource = ResourceTestUtils.getUnreservedMountVolume(2000, Optional.empty());
    Resource offeredCpu = ResourceTestUtils.getUnreservedCpus(1.0);

    List<OfferRecommendation> recommendations = evaluator.evaluate(
            PodInstanceRequirementTestUtils.getRootVolumeRequirement(1.0, 1500),
            Arrays.asList(OfferTestUtils.getCompleteOffer(Arrays.asList(offeredCpu, wrongOfferedResource))));
    Assert.assertEquals(0, recommendations.size());
    Assert.assertEquals(0, recommendations.size());
}
 

public static Protos.Resource getUnreservedMountVolume(double diskSize, Optional<String> profile) {
    Protos.Resource.Builder builder = getUnreservedDisk(diskSize).toBuilder();
    builder.getDiskBuilder().setSource(TestConstants.MOUNT_DISK_SOURCE);
    if (profile.isPresent()) {
        builder.getDiskBuilder().getSourceBuilder().setProfile(profile.get());
    }
    return builder.build();
}
 

protected List<Resource> recordLaunchWithOfferedResources(
        PodInstanceRequirement podInstanceRequirement, Resource... offeredResources)
        throws InvalidRequirementException, IOException {
    return recordLaunchWithOfferedResources(
            OfferTestUtils.getOffer(Arrays.asList(offeredResources)),
            podInstanceRequirement,
            offeredResources);
}
 

protected List<Resource> recordLaunchWithCompleteOfferedResources(
        PodInstanceRequirement podInstanceRequirement, String preReservedRole, Resource... offeredResources)
        throws InvalidRequirementException, IOException {
    return recordLaunchWithOfferedResources(
            OfferTestUtils.getCompleteOffer(Arrays.asList(offeredResources), preReservedRole),
            podInstanceRequirement,
            offeredResources);
}
 

public static Protos.Offer getCompleteOffer(Protos.ExecutorID executorId, List<Protos.Resource> resources) {
    Protos.Offer.Builder builder = getEmptyOfferBuilder();
    builder.addAllResources(resources);

    if (executorId != null) {
        builder.addExecutorIds(executorId);
    }

    return builder.build();
}
 

/**
 * Minimum to keep required field errors away.
 */
public static void addResource(Offer.Builder o, String name, String role) {
    Resource.Builder b = o.addResourcesBuilder().setType(Value.Type.RANGES).setName(name);
    if (role != null) {
        b.setRole(role);
    }
}
 

@Test
public void testCreateSingleReservedAtomicPool() {
    Resource resource = ResourceTestUtils.getReservedMountVolume(1000, Optional.empty());
    Offer offer = OfferTestUtils.getOffer(resource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));
    String resourceId = new MesosResource(resource).getResourceId().get();

    Assert.assertEquals(0, pool.getUnreservedAtomicPool().size());
    Assert.assertEquals(1, pool.getDynamicallyReservedPoolByResourceId().size());
    Assert.assertEquals(resource, pool.getDynamicallyReservedPoolByResourceId().get(resourceId).getResource());
}
 

@Test
public void testConsumeUnreservedMergedResource() {
    Resource resource = ResourceTestUtils.getUnreservedCpus(1.0);
    Protos.Value resourceValue = ValueUtils.getValue(resource);
    Offer offer = OfferTestUtils.getOffer(resource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertEquals(1, pool.getUnreservedMergedPool().size());
    Assert.assertEquals(resource.getScalar().getValue(),
            pool.getUnreservedMergedPool().get("cpus").getScalar().getValue(), 0.0);
    MesosResource resourceToConsume = pool.consumeReservableMerged(resource.getName(), resourceValue, Constants.ANY_ROLE).get();
    Assert.assertEquals(resource, resourceToConsume.getResource());
    Assert.assertEquals(ValueUtils.getZero(Protos.Value.Type.SCALAR),
            pool.getUnreservedMergedPool().get("cpus"));
}
 

@Test
public void testConsumeInsufficientUnreservedMergedResource() {
    Resource desiredUnreservedResource = ResourceTestUtils.getUnreservedCpus(2.0);
    Protos.Value resourceValue = ValueUtils.getValue(desiredUnreservedResource);
    Resource offeredUnreservedResource = ResourceTestUtils.getUnreservedCpus(1.0);
    Offer offer = OfferTestUtils.getOffer(offeredUnreservedResource);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Assert.assertFalse(
            pool.consumeReservableMerged(desiredUnreservedResource.getName(), resourceValue, Constants.ANY_ROLE)
                    .isPresent());
}
 

@Test
public void testConsumeUnreservedWhenNoUnreservedResources() {
    Resource reservedCpu = ResourceTestUtils.getReservedCpus(1.0, UUID.randomUUID().toString());
    Offer offer = OfferTestUtils.getOffer(reservedCpu);
    MesosResourcePool pool = new MesosResourcePool(offer, Optional.of(Constants.ANY_ROLE));

    Map<String, Protos.Value> map = pool.getUnreservedMergedPool();
    Assert.assertTrue(map != null);
    Assert.assertTrue(map.isEmpty());
}