import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testExactMatchesRegex() {
    Offer.Builder o = getOfferWithResources()
            .addAttributes(ATTR_TEXT);
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_TEXT_REGEX))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_SCALAR);
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_SCALAR_REGEX))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_RANGES);
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_RANGES_REGEX))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_SET);
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_SET_REGEX))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Before
public void createMocks() {
  pendingTask = new SingularityPendingTask(new SingularityPendingTaskId("test", "1", 0, 1, PendingType.IMMEDIATE, 0), Optional.<List<String>> absent(),
      Optional.<String> absent(), Optional.<String> absent(), Optional.<Boolean> absent(), Optional.<String> absent());

  final SingularitySlaveAndRackHelper slaveAndRackHelper = mock(SingularitySlaveAndRackHelper.class);
  final ExecutorIdGenerator idGenerator = mock(ExecutorIdGenerator.class);

  when(idGenerator.getNextExecutorId()).then(new CreateFakeId());

  builder = new SingularityMesosTaskBuilder(new ObjectMapper(), slaveAndRackHelper, idGenerator, new SingularityConfiguration());

  taskResources = new Resources(1, 1, 0);
  executorResources = new Resources(0.1, 1, 0);

  offer = Offer.newBuilder()
      .setSlaveId(SlaveID.newBuilder().setValue("1"))
      .setId(OfferID.newBuilder().setValue("1"))
      .setFrameworkId(FrameworkID.newBuilder().setValue("1"))
      .setHostname("test")
      .build();

  when(slaveAndRackHelper.getRackId(offer)).thenReturn(Optional.<String> absent());
  when(slaveAndRackHelper.getMaybeTruncatedHost(offer)).thenReturn("host");
  when(slaveAndRackHelper.getRackIdOrDefault(offer)).thenReturn("DEFAULT");
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
protected SingularityTask prepTask(SingularityRequest request, SingularityDeploy deploy, long launchTime, int instanceNo, boolean separateHosts) {
  SingularityPendingTask pendingTask = buildPendingTask(request, deploy, launchTime, instanceNo);
  SingularityTaskRequest taskRequest = new SingularityTaskRequest(request, deploy, pendingTask);

  Offer offer;
  if (separateHosts) {
    offer = createOffer(125, 1024, String.format("slave%s", instanceNo), String.format("host%s", instanceNo));
  } else {
    offer = createOffer(125, 1024);
  }

  SingularityTaskId taskId = new SingularityTaskId(request.getId(), deploy.getId(), launchTime, instanceNo, offer.getHostname(), "rack1");
  TaskID taskIdProto = TaskID.newBuilder().setValue(taskId.toString()).build();

  TaskInfo taskInfo = TaskInfo.newBuilder()
      .setSlaveId(offer.getSlaveId())
      .setTaskId(taskIdProto)
      .setName("name")
      .build();

  SingularityTask task = new SingularityTask(taskRequest, taskId, offer, taskInfo, Optional.of("rack1"));

  taskManager.savePendingTask(pendingTask);

  return task;
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的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.Offer; //导入依赖的package包/类
@Test
public void testLaunchTransientCustomExecutor() {
    Resource resource = ResourceTestUtils.getUnreservedCpus(1.0);
    Offer offer = OfferTestUtils.getOffer(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,
                    false)));
    Assert.assertEquals(1, recorder.getLaunches().size());
    verify(driver, times(0)).acceptOffers(
            anyCollectionOf(OfferID.class),
            anyCollectionOf(Operation.class),
            anyObject());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testExactMatchesString() {
    Offer.Builder o = getOfferWithResources()
            .addAttributes(ATTR_TEXT);
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_TEXT)))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_SCALAR);
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_SCALAR)))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_RANGES);
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_RANGES)))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());

    o = getOfferWithResources()
            .addAttributes(ATTR_SET);
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_SET)))
            .filter(o.build(), POD_INSTANCE, Collections.emptyList()).isPassing());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testTaskInfoRevocable() {
  expect(tierManager.getTier(TASK_CONFIG)).andReturn(REVOCABLE_TIER);
  taskFactory = new MesosTaskFactoryImpl(config, tierManager, SERVER_INFO);

  Resource revocableCPU = OFFER_THERMOS_EXECUTOR.getResources(0).toBuilder()
      .setRevocable(Resource.RevocableInfo.getDefaultInstance())
      .build();
  Offer withRevocable = OFFER_THERMOS_EXECUTOR.toBuilder()
      .removeResources(0)
      .addResources(0, revocableCPU)
      .build();

  control.replay();

  TaskInfo task = taskFactory.createFrom(TASK, withRevocable);
  checkTaskResources(TASK.getTask(), task);
  assertTrue(task.getResourcesList().stream().anyMatch(Resource::hasRevocable));
  checkDiscoveryInfoUnset(task);
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testAnyMatchesString() {
    Offer o = getOfferWithResources()
            .addAttributes(ATTR_TEXT)
            .addAttributes(ATTR_SCALAR)
            .addAttributes(ATTR_RANGES)
            .addAttributes(ATTR_SET)
            .build();
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_TEXT)))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_SCALAR)))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_RANGES)))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(ExactMatcher.create(AttributeStringUtils.toString(ATTR_SET)))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testExpectedDecommissioningVolume() {
    when(mockStateStore.fetchGoalOverrideStatus(TASK_INFO_2.getName()))
            .thenReturn(DecommissionPlanFactory.DECOMMISSIONING_STATUS);
    ResourceCleaner cleaner = new DefaultResourceCleaner(mockStateStore);
    List<Offer> offers = OfferTestUtils.getOffers(EXPECTED_RESOURCE_2);
    List<OfferRecommendation> recommendations = cleaner.evaluate(offers);

    assertEquals("Got: " + recommendations, 2, recommendations.size());

    OfferRecommendation rec = recommendations.get(0);
    assertEquals(Operation.Type.DESTROY, rec.getOperation().getType());

    rec = recommendations.get(1);
    assertEquals(Operation.Type.UNRESERVE, rec.getOperation().getType());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
public void resourceOffers(SchedulerDriver driver, List<Protos.Offer> offers) {
  synchronized (_offersLock) {
    if (_offers == null) {
      return;
    }
    LOG.debug("resourceOffers: Currently have {} offers buffered {}",
              _offers.size(), (_offers.size() > 0 ? (":" + offerMapToString(_offers)) : ""));

    for (Protos.Offer offer : offers) {
      if (isHostAccepted(offer.getHostname())) {
        LOG.debug("resourceOffers: Recording offer from host: {}, offerId: {}",
                  offer.getHostname(), offer.getId().getValue());
        _offers.put(offer.getId(), offer);
      } else {
        LOG.debug("resourceOffers: Declining offer from host: {}, offerId: {}",
                  offer.getHostname(), offer.getId().getValue());
        driver.declineOffer(offer.getId());
      }
    }
    LOG.debug("resourceOffers: After processing offers, now have {} offers buffered: {}",
              _offers.size(), offerMapToString(_offers));
  }
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的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.Offer; //导入依赖的package包/类
@Override
public Collection<OfferID> resourceOffers(
        final SchedulerDriver driver,
        final List<Offer> offers,
        final Collection<? extends Step> steps) {
    if (driver == null || offers == null || steps == null) {
        logger.error("Unexpected null argument(s) encountered: driver='{}' offers='{}', steps='{}'",
                driver, offers, steps);
        return Collections.emptyList();
    }

    List<OfferID> acceptedOfferIds = new ArrayList<>();
    List<Offer> availableOffers = new ArrayList<>(offers);

    for (Step step : steps) {
        acceptedOfferIds.addAll(resourceOffers(driver, availableOffers, step));
        availableOffers = PlanUtils.filterAcceptedOffers(availableOffers, acceptedOfferIds);
    }

    return acceptedOfferIds;
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
public LaunchOfferRecommendation(
        Offer offer,
        TaskInfo originalTaskInfo,
        Protos.ExecutorInfo executorInfo,
        boolean shouldLaunch,
        boolean useDefaultExecutor) {
    this.offer = offer;
    this.shouldLaunch = shouldLaunch;
    this.useDefaultExecutor = useDefaultExecutor;

    TaskInfo.Builder taskBuilder = originalTaskInfo.toBuilder();
    if (!shouldLaunch) {
        new TaskLabelWriter(taskBuilder).setTransient();
        taskBuilder.getTaskIdBuilder().setValue("");
    }

    taskBuilder.setSlaveId(offer.getSlaveId());

    this.taskInfo = taskBuilder.build();
    this.executorInfo = executorInfo;
    this.operation = getLaunchOperation();
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
/**
 * Implementation of task type avoidance. Considers the presence of tasks in the cluster to
 * determine whether the provided task can be launched against a given offer. This rule requires
 * that the offer be located on an agent which doesn't currently have an instance of the
 * specified task type.
 */
private EvaluationOutcome filterAvoid(
        Offer offer,
        PodInstance podInstance,
        Collection<TaskInfo> tasksToAvoid) {

    for (TaskInfo taskToAvoid : tasksToAvoid) {
        if (PlacementUtils.areEquivalent(taskToAvoid, podInstance)) {
            // This is stale data for the same task that we're currently evaluating for
            // placement. Don't worry about avoiding it. This occurs when we're redeploying
            // a given task with a new configuration (old data not deleted yet).
            continue;
        }
        if (taskToAvoid.getSlaveId().equals(offer.getSlaveId())) {
            // The offer is for an agent which has a task to be avoided. Denied!
            return EvaluationOutcome.fail(
                    this,
                    "Found a task matching avoided type '%s' on this agent.", typeToFind)
                    .build();
        }
    }
    // The offer doesn't match any tasks to avoid. Approved!
    return EvaluationOutcome.pass(this, "No tasks of avoided type '%s' found on this agent.", typeToFind).build();
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void stepWithDifferentNameLaunches() throws Exception {
    final List<Offer> offers = getOffers();
    final Protos.TaskStatus status = TaskTestUtils.generateStatus(taskInfo.getTaskId(), Protos.TaskState.TASK_FAILED);
    final Step step = mock(Step.class);

    launchConstrainer.setCanLaunch(true);
    stateStore.storeTasks(taskInfos);
    stateStore.storeStatus(taskInfo.getName(), status);
    stateStore.storeFrameworkId(TestConstants.FRAMEWORK_ID);
    when(offerAccepter.accept(any(), any())).thenReturn(Arrays.asList(offers.get(0).getId()));
    when(step.getName()).thenReturn("different-name");
    when(mockDeployManager.getCandidates(Collections.emptyList())).thenReturn((Collection) Arrays.asList(step));

    recoveryManager.update(status);
    Collection<Protos.OfferID> acceptedOffers = planScheduler.resourceOffers(
            schedulerDriver,
            getOffers(),
            planCoordinator.getCandidates());

    assertEquals(1, acceptedOffers.size());
    reset(mockDeployManager);
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Override
public EvaluationOutcome filter(Offer offer, PodInstance podInstance, Collection<TaskInfo> tasks) {
    if (isAcceptable(offer, podInstance, tasks)) {
        return EvaluationOutcome.pass(
                this,
                "Fits within limit of %d tasks matching filter '%s' on this agent with attribute: %s",
                max, taskFilter.toString(), attributeMatcher.toString())
                .build();
    } else {
        return EvaluationOutcome.fail(
                this,
                "Reached greater than %d tasks matching filter '%s' on this agent with attribute: %s",
                max,
                taskFilter.toString(),
                attributeMatcher.toString())
                .build();
    }
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Override
public EvaluationOutcome filter(Offer offer, PodInstance podInstance, Collection<TaskInfo> tasks) {
    if (isAcceptable(offer, podInstance, tasks)) {
        return EvaluationOutcome.pass(
                this,
                "Match found for attribute pattern: '%s'", getMatcher().toString())
                .build();
    } else {
        return EvaluationOutcome.fail(
                this,
                "None of %d attributes matched pattern: '%s'",
                offer.getAttributesCount(),
                getMatcher().toString())
                .build();
    }
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Test
public void testAnyMatchesRegex() {
    Offer o = getOfferWithResources()
            .addAttributes(ATTR_TEXT)
            .addAttributes(ATTR_SCALAR)
            .addAttributes(ATTR_RANGES)
            .addAttributes(ATTR_SET)
            .build();
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_TEXT_REGEX))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_SCALAR_REGEX))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_RANGES_REGEX))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
    assertTrue(AttributeRuleFactory.getInstance().require(RegexMatcher.create(ATTR_SET_REGEX))
            .filter(o, POD_INSTANCE, Collections.emptyList()).isPassing());
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Override
public EvaluationOutcome filter(Offer offer, PodInstance podInstance, Collection<TaskInfo> tasks) {
    if (isAcceptable(offer, podInstance, tasks)) {
        return EvaluationOutcome.pass(
                this,
                "Fewer than %d tasks matching filter '%s' are present on this host",
                maxTasksPerHostname, taskFilter.toString())
                .build();
    } else {
        return EvaluationOutcome.fail(
                this,
                "%d tasks matching filter '%s' are already present on this host",
                maxTasksPerHostname, taskFilter.toString())
                .build();
    }
}
 

import org.apache.mesos.Protos.Offer; //导入依赖的package包/类
@Override
public void resourceOffers(SchedulerDriver driver, List<Offer> offers) {
  lastOfferTimestamp = Optional.of(System.currentTimeMillis());

  if (!isRunning()) {
    LOG.info(String.format("Scheduler is in state %s, declining %s offer(s)", state.name(), offers.size()));

    for (Protos.Offer offer : offers) {
      driver.declineOffer(offer.getId());
    }

    return;
  }

  final long start = System.currentTimeMillis();

  lock.lock();

  try {
    scheduler.resourceOffers(driver, offers);
  } catch (Throwable t) {
    handleUncaughtSchedulerException(t);
  } finally {
    lock.unlock();

    LOG.debug("Handled {} resource offers in {}", offers.size(), JavaUtils.duration(start));
  }
}