import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
private List<TaskRequest> getTasks() {
    List<TaskRequest> result = new ArrayList<>(20);
    for (int i = 0; i < 20; i++) {
        String jobName;
        String appName;
        if (i % 2 == 0) {
            jobName = String.format("foo-%d", i);
            appName = "foo-app";
        } else {
            jobName = String.format("bar-%d", i);
            appName = "bar-app";
        }
        result.add(getTask(jobName));
        when(facadeService.load(jobName)).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration(jobName, appName)));
        
    }
    when(facadeService.loadAppConfig("foo-app")).thenReturn(Optional.of(CloudAppConfigurationBuilder.createCloudAppConfiguration("foo-app")));
    when(facadeService.loadAppConfig("bar-app")).thenReturn(Optional.of(CloudAppConfigurationBuilder.createCloudAppConfiguration("bar-app")));
    return result;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Test
public void assertRunOneIteration() throws Exception {
    when(facadeService.getEligibleJobContext()).thenReturn(Lists.newArrayList(
            JobContext.from(CloudJobConfigurationBuilder.createCloudJobConfiguration("failover_job", CloudJobExecutionType.DAEMON, 1), ExecutionType.FAILOVER)));
    Map<String, VMAssignmentResult> vmAssignmentResultMap = new HashMap<>();
    vmAssignmentResultMap.put("rs1", new VMAssignmentResult("localhost", Lists.<VirtualMachineLease>newArrayList(new VMLeaseObject(OfferBuilder.createOffer("offer_0"))),
            Sets.newHashSet(mockTaskAssignmentResult("failover_job", ExecutionType.FAILOVER))));
    when(taskScheduler.scheduleOnce(ArgumentMatchers.<TaskRequest>anyList(), ArgumentMatchers.<VirtualMachineLease>anyList())).thenReturn(new SchedulingResult(vmAssignmentResultMap));
    when(facadeService.load("failover_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createCloudJobConfiguration("failover_job")));
    when(facadeService.getFailoverTaskId(any(TaskContext.MetaInfo.class)))
            .thenReturn(Optional.of(String.format("%[email protected]@[email protected]@%[email protected]@[email protected]@0", "failover_job",  ExecutionType.FAILOVER.name())));
    when(taskScheduler.getTaskAssigner()).thenReturn(mock(Action2.class));
    taskLaunchScheduledService.runOneIteration();
    verify(facadeService).removeLaunchTasksFromQueue(ArgumentMatchers.<TaskContext>anyList());
    verify(facadeService).loadAppConfig("test_app");
    verify(jobEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Test
public void assertRunOneIterationWithScriptJob() throws Exception {
    when(facadeService.getEligibleJobContext()).thenReturn(Lists.newArrayList(
            JobContext.from(CloudJobConfigurationBuilder.createScriptCloudJobConfiguration("script_job", 1), ExecutionType.READY)));
    Map<String, VMAssignmentResult> vmAssignmentResultMap = new HashMap<>();
    vmAssignmentResultMap.put("rs1", new VMAssignmentResult("localhost", Lists.<VirtualMachineLease>newArrayList(new VMLeaseObject(OfferBuilder.createOffer("offer_0"))),
            Sets.newHashSet(mockTaskAssignmentResult("script_job", ExecutionType.READY))));
    when(taskScheduler.scheduleOnce(ArgumentMatchers.<TaskRequest>anyList(), ArgumentMatchers.<VirtualMachineLease>anyList())).thenReturn(new SchedulingResult(vmAssignmentResultMap));
    when(facadeService.loadAppConfig("test_app")).thenReturn(Optional.of(CloudAppConfigurationBuilder.createCloudAppConfiguration("test_app")));
    when(facadeService.load("script_job")).thenReturn(Optional.of(CloudJobConfigurationBuilder.createScriptCloudJobConfiguration("script_job", 1)));
    when(taskScheduler.getTaskAssigner()).thenReturn(mock(Action2.class));
    taskLaunchScheduledService.runOneIteration();
    verify(facadeService).removeLaunchTasksFromQueue(ArgumentMatchers.<TaskContext>anyList());
    verify(facadeService).isRunning(TaskContext.from(String.format("%[email protected]@[email protected]@%[email protected]@[email protected]@0", "script_job", ExecutionType.READY)));
    verify(facadeService).loadAppConfig("test_app");
    verify(jobEventBus).post(ArgumentMatchers.<JobStatusTraceEvent>any());
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Determines whether a particular target host is appropriate for a particular task request by rejecting any
 * host that has the same value for the unique constraint attribute as another host that is already assigned
 * a co-task of the specified task request.
 *
 * @param taskRequest      describes the task being considered for assignment to the host
 * @param targetVM         describes the host being considered as a target for the task
 * @param taskTrackerState describes the state of tasks previously assigned or already running throughout
 *                         the system
 * @return a successful Result if the target does not have the same value for its unique constraint
 *         attribute as another host that has already been assigned a co-task of {@code taskRequest}, or an
 *         unsuccessful Result otherwise
 */
@Override
public Result evaluate(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    Set<String> coTasks = coTasksGetter.call(taskRequest.getId());
    String targetHostAttrVal = AttributeUtilities.getAttrValue(targetVM.getCurrAvailableResources(), hostAttributeName);
    if(targetHostAttrVal==null || targetHostAttrVal.isEmpty()) {
        return new Result(false, hostAttributeName + " attribute unavailable on host " + targetVM.getCurrAvailableResources().hostname());
    }
    for(String coTask: coTasks) {
        TaskTracker.ActiveTask activeTask = taskTrackerState.getAllRunningTasks().get(coTask);
        if(activeTask==null)
            activeTask = taskTrackerState.getAllCurrentlyAssignedTasks().get(coTask);
        if(activeTask!=null) {
            String usedAttrVal = AttributeUtilities.getAttrValue(activeTask.getTotalLease(), hostAttributeName);
            if(usedAttrVal==null || usedAttrVal.isEmpty())
                return new Result(false, hostAttributeName+" attribute unavailable on host " + activeTask.getTotalLease().hostname() +
                        " running co-task " + coTask);
            if(usedAttrVal.equals(targetHostAttrVal)) {
                return new Result(false, hostAttributeName+" " + targetHostAttrVal + " already used for another co-task " + coTask);
            }
        }
    }
    return new Result(true, "");
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Override
public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    return calculateResourceFitness(taskRequest, targetVM, taskTrackerState,
            new Func1<TaskRequest, Double>() {
                @Override
                public Double call(TaskRequest request) {
                    return request.getCPUs();
                }
            },
            new Func1<VirtualMachineLease, Double>() {
                @Override
                public Double call(VirtualMachineLease l) {
                    return l.cpuCores();
                }
            });
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
private static double calculateResourceFitness(TaskRequest request, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState,
                                               Func1<TaskRequest, Double> taskResourceGetter,
                                               Func1<VirtualMachineLease, Double> leaseResourceGetter) {
    double totalRes = leaseResourceGetter.call(targetVM.getCurrAvailableResources());
    Iterator<TaskRequest> iterator = targetVM.getRunningTasks().iterator();
    double oldJobsTotal=0.0;
    while(iterator.hasNext())
        oldJobsTotal += taskResourceGetter.call(iterator.next());
    double usedResource = taskResourceGetter.call(request);
    Iterator<TaskAssignmentResult> taskAssignmentResultIterator = targetVM.getTasksCurrentlyAssigned().iterator();
    while(taskAssignmentResultIterator.hasNext())
        usedResource += taskResourceGetter.call(taskAssignmentResultIterator.next().getRequest());
    totalRes += oldJobsTotal;
    usedResource += oldJobsTotal;
    return usedResource / totalRes;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Test
public void testCalculatorWeightedAverage() throws Exception {
    VMTaskFitnessCalculator fitnessCalculator1 = mock(VMTaskFitnessCalculator.class);
    when(fitnessCalculator1.calculateFitness(any(), any(), any())).thenReturn(0.5);

    VMTaskFitnessCalculator fitnessCalculator2 = mock(VMTaskFitnessCalculator.class);
    when(fitnessCalculator2.calculateFitness(any(), any(), any())).thenReturn(1.0);

    List<WeightedFitnessCalculator> weightedFitnessCalculators = Arrays.asList(
            new WeightedFitnessCalculator(fitnessCalculator1, 0.5),
            new WeightedFitnessCalculator(fitnessCalculator2, 0.5)
    );
    WeightedAverageFitnessCalculator calculator = new WeightedAverageFitnessCalculator(weightedFitnessCalculators);
    double fitness = calculator.calculateFitness(mock(TaskRequest.class), mock(VirtualMachineCurrentState.class), mock(TaskTrackerState.class));
    Assert.assertEquals(0.75, fitness, 0.0);
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Test
public void testCalculatorWeightedAverageWithZero() throws Exception {
    VMTaskFitnessCalculator fitnessCalculator1 = mock(VMTaskFitnessCalculator.class);
    when(fitnessCalculator1.calculateFitness(any(), any(), any())).thenReturn(0.0);

    VMTaskFitnessCalculator fitnessCalculator2 = mock(VMTaskFitnessCalculator.class);
    when(fitnessCalculator2.calculateFitness(any(), any(), any())).thenReturn(1.0);

    List<WeightedFitnessCalculator> weightedFitnessCalculators = Arrays.asList(
            new WeightedFitnessCalculator(fitnessCalculator1, 0.3),
            new WeightedFitnessCalculator(fitnessCalculator2, 0.7)
    );
    WeightedAverageFitnessCalculator calculator = new WeightedAverageFitnessCalculator(weightedFitnessCalculators);
    double fitness = calculator.calculateFitness(mock(TaskRequest.class), mock(VirtualMachineCurrentState.class), mock(TaskTrackerState.class));
    Assert.assertEquals(0.0, fitness, 0.0);
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
List<TaskRequest> getPendingTasks() {
    List<TaskRequest> result = new ArrayList<>(eligibleJobContextsMap.size() * 10);
    for (JobContext each : eligibleJobContextsMap.values()) {
        result.addAll(createTaskRequests(each));
    }
    return result;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
private Collection<TaskRequest> createTaskRequests(final JobContext jobContext) {
    Collection<TaskRequest> result = new ArrayList<>(jobContext.getAssignedShardingItems().size());
    for (int each : jobContext.getAssignedShardingItems()) {
        result.add(new JobTaskRequest(new TaskContext(jobContext.getJobConfig().getJobName(), Collections.singletonList(each), jobContext.getType()), jobContext.getJobConfig()));
    }
    return result;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
private TaskRequest getTask(final String jobName) {
    TaskRequest result = mock(TaskRequest.class);
    when(result.getCPUs()).thenReturn(1.0d);
    when(result.getMemory()).thenReturn(128.0d);
    when(result.getHardConstraints()).thenAnswer(new Answer<List<? extends ConstraintEvaluator>>() {
        @Override
        public List<? extends ConstraintEvaluator> answer(final InvocationOnMock invocationOnMock) throws Throwable {
            return ImmutableList.of(AppConstraintEvaluator.getInstance());
        }
    });
    when(result.getId()).thenReturn(new TaskContext(jobName, Collections.singletonList(0), ExecutionType.READY).getId());
    return result;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Recover framework/worker information persisted by a prior incarnation of the RM.
 */
private void recoverWorkers() throws Exception {
	// if this resource manager is recovering from failure,
	// then some worker tasks are most likely still alive and we can re-obtain them
	final List<MesosWorkerStore.Worker> tasksFromPreviousAttempts = workerStore.recoverWorkers();

	assert(workersInNew.isEmpty());
	assert(workersInLaunch.isEmpty());
	assert(workersBeingReturned.isEmpty());

	if (!tasksFromPreviousAttempts.isEmpty()) {
		LOG.info("Retrieved {} TaskManagers from previous attempt", tasksFromPreviousAttempts.size());

		List<Tuple2<TaskRequest, String>> toAssign = new ArrayList<>(tasksFromPreviousAttempts.size());

		for (final MesosWorkerStore.Worker worker : tasksFromPreviousAttempts) {
			LaunchableMesosWorker launchable = createLaunchableMesosWorker(worker.taskID(), worker.profile());

			switch(worker.state()) {
				case New:
					// remove new workers because allocation requests are transient
					workerStore.removeWorker(worker.taskID());
					break;
				case Launched:
					workersInLaunch.put(extractResourceID(worker.taskID()), worker);
					toAssign.add(new Tuple2<>(launchable.taskRequest(), worker.hostname().get()));
					break;
				case Released:
					workersBeingReturned.put(extractResourceID(worker.taskID()), worker);
					break;
			}
			taskMonitor.tell(new TaskMonitor.TaskGoalStateUpdated(extractGoalState(worker)), selfActor);
		}

		// tell the launch coordinator about prior assignments
		if (toAssign.size() >= 1) {
			launchCoordinator.tell(new LaunchCoordinator.Assign(toAssign), selfActor);
		}
	}
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Returns a "soft" constraint, in the form of a {@link VMTaskFitnessCalculator}, based on a specified "hard"
 * constraint, in the form of a {@link ConstraintEvaluator}.
 *
 * @param c the "hard" constraint to convert
 * @return a "soft" constraint version of {@code c}
 */
public static VMTaskFitnessCalculator get(final ConstraintEvaluator c) {
    // This fitness calculator return 0 or 1. This can possibly be improved upon by the ConstraintEvaluator using its
    // own logic.
    return new VMTaskFitnessCalculator() {
        @Override
        public String getName() {
            return c.getName();
        }
        @Override
        public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
            return c.evaluate(taskRequest, targetVM, taskTrackerState).isSuccessful()? 1.0 : 0.0;
        }
    };
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Create a sample mesos framework with the given task queue and mesos master connection string. As would be typical
 * for frameworks that wish to use Fenzo task scheduler, a lease queue is created for mesos scheduler callback to
 * insert offers received from mesos. This sample implementation obtains the tasks to run via a task queue. The
 * {@link SampleFramework#runAll()} method implements the scheduling loop that continuously takes pending tasks from
 * the queue and uses Fenzo's task scheduler to assign resources to them.
 *
 * The task scheduler created in this sample is a rather simple one, with no advanced features.
 *
 * @param taskQueue The task queue.
 * @param mesosMaster Connection string for mesos master.
 * @param onTaskComplete A single argument action trigger to invoke upon task completion, with task ID is the argument.
 * @param taskCmdGetter A single argument function to invoke to get the command line to execute for a given task ID,
 *                      passed as the only argument.
 */
public SampleFramework(BlockingQueue<TaskRequest> taskQueue, String mesosMaster, Action1<String> onTaskComplete,
                       Func1<String, String> taskCmdGetter) {
    this.taskQueue = taskQueue;
    this.leasesQueue = new LinkedBlockingQueue<>();
    this.onTaskComplete = onTaskComplete;
    this.taskCmdGetter = taskCmdGetter;
    launchedTasks = new HashMap<>();
    scheduler = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000000)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease lease) {
                    System.out.println("Declining offer on " + lease.hostname());
                    ref.get().declineOffer(lease.getOffer().getId());
                }
            })
            .build();
    Protos.FrameworkInfo framework = Protos.FrameworkInfo.newBuilder()
            .setName("Sample Fenzo Framework")
            .setUser("")
            .build();
    Scheduler mesosScheduler = new MesosScheduler();
    mesosSchedulerDriver = new MesosSchedulerDriver(mesosScheduler, framework, mesosMaster);
    ref.set(mesosSchedulerDriver);
    new Thread() {
        public void run() {
            mesosSchedulerDriver.run();
        }
    }.start();
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Override
public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    double totalWeightedScores = 0.0;
    double totalWeights = 0.0;
    for (WeightedFitnessCalculator calculator : calculators) {
        double score = calculator.getFitnessCalculator().calculateFitness(taskRequest, targetVM, taskTrackerState);
        // If any of the scores are 0.0 then the final score should be 0.0
        if (score == 0.0) {
            return score;
        }
        totalWeightedScores += (score * calculator.getWeight());
        totalWeights += calculator.getWeight();
    }
    return totalWeightedScores / totalWeights;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Override
public Result evaluate(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    Set<String> coTasks = coTasksGetter.call(taskRequest.getId());
    String targetHostAttrVal = AttributeUtilities.getAttrValue(targetVM.getCurrAvailableResources(), hostAttributeName);
    if(targetHostAttrVal==null || targetHostAttrVal.isEmpty()) {
        return new Result(false, hostAttributeName + " attribute unavailable on host " + targetVM.getCurrAvailableResources().hostname());
    }
    Map<String, Integer> usedAttribsMap = null;
    try {
        usedAttribsMap = getUsedAttributesMap(coTasks, taskTrackerState);
    } catch (Exception e) {
        return new Result(false, e.getMessage());
    }
    final Integer integer = usedAttribsMap.get(targetHostAttrVal);
    if(integer==null || usedAttribsMap.isEmpty())
        return new Result(true, "");
    int min=Integer.MAX_VALUE;
    int max=Integer.MIN_VALUE;
    for(Integer i: usedAttribsMap.values()) {
        min = Math.min(min, i);
        max = Math.max(max, i);
    }
    min = expectedValues>usedAttribsMap.size()? 0 : min;
    if(min == max || integer<max)
        return new Result(true, "");
    return new Result(false, "Would further imbalance by host attribute " + hostAttributeName);
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Converts this constraint into a "soft" constraint. By default, a balanced host attribute constraint is a
 * "hard" constraint, which is to say that Fenzo will guarantee that the constraint is applied and will fail
 * to place a task if the only way it can do so is to violate the constraint. This method returns a
 * {@link VMTaskFitnessCalculator} that represents this constraint as a "soft" constraint that will permit
 * Fenzo to place a task in violation of the constraint if it cannot do so otherwise.
 *
 * @return a task fitness calculator that represents the balanced host attribute constraint as a soft
 *         constraint
 */
public VMTaskFitnessCalculator asSoftConstraint() {
    return new VMTaskFitnessCalculator() {
        @Override
        public String getName() {
            return name;
        }

        @Override
        public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
            String targetHostAttrVal = AttributeUtilities.getAttrValue(targetVM.getCurrAvailableResources(), hostAttributeName);
            if(targetHostAttrVal==null || targetHostAttrVal.isEmpty()) {
                return 0.0;
            }
            Set<String> coTasks = coTasksGetter.call(taskRequest.getId());
            Map<String, Integer> usedAttribsMap = null;
            try {
                usedAttribsMap = getUsedAttributesMap(coTasks, taskTrackerState);
            } catch (Exception e) {
                return 0.0;
            }
            final Integer integer = usedAttribsMap.get(targetHostAttrVal);
            if(integer==null)
                return 1.0;
            if(usedAttribsMap.isEmpty())
                return 1.0;
            double avg=0.0;
            for(Integer i: usedAttribsMap.values())
                avg += i;
            avg = Math.ceil(avg+1 / Math.max(expectedValues, usedAttribsMap.size()));
            if(integer<=avg)
                return (avg-(double)integer) / avg;
            return 0.0;
        }
    };
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Override
public double calculateFitness(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    double cpuFitness = cpuSpreader.calculateFitness(taskRequest, targetVM, taskTrackerState);
    double memFitness = memorySpreader.calculateFitness(taskRequest, targetVM, taskTrackerState);
    double networkFitness = networkSpreader.calculateFitness(taskRequest, targetVM, taskTrackerState);
    return (cpuFitness + memFitness + networkFitness) / 3.0;
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
/**
 * Tests a host to determine whether it has an attribute of the required value for this task request.
 *
 * @param taskRequest      describes the task being evaluated for assignment to the host
 * @param targetVM         describes the host being evaluated as a target for the task
 * @param taskTrackerState describes the state of tasks already assigned or running on hosts throughout the
 *                         system
 * @return a successful Result if the host has an attribute with the required value, or an unsuccessful
 *         Result otherwise
 */
@Override
public Result evaluate(TaskRequest taskRequest, VirtualMachineCurrentState targetVM, TaskTrackerState taskTrackerState) {
    String targetHostAttrVal = getAttrValue(targetVM.getCurrAvailableResources());
    if(targetHostAttrVal==null || targetHostAttrVal.isEmpty()) {
        return new Result(false, hostAttributeName + " attribute unavailable on host " + targetVM.getCurrAvailableResources().hostname());
    }
    String requiredAttrVal = hostAttributeValueGetter.call(taskRequest.getId());
    return targetHostAttrVal.equals(requiredAttrVal)?
            new Result(true, "") :
            new Result(false, "Host attribute " + hostAttributeName + ": required=" + requiredAttrVal + ", got=" + targetHostAttrVal);
}
 

import com.netflix.fenzo.TaskRequest; //导入依赖的package包/类
@Test
public void assertGetCustomNamedResources() {
    assertThat(jobTaskRequest.getCustomNamedResources(), is(Collections.<String, TaskRequest.NamedResourceSetRequest>emptyMap()));
}