import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
/**
 * Creates the Fenzo optimizer (builder).
 * The builder is an indirection to facilitate unit testing of the Launch Coordinator.
 */
private static TaskSchedulerBuilder createOptimizer() {
	return new TaskSchedulerBuilder() {
		TaskScheduler.Builder builder = new TaskScheduler.Builder();

		@Override
		public TaskSchedulerBuilder withLeaseRejectAction(Action1<VirtualMachineLease> action) {
			builder.withLeaseRejectAction(action);
			return this;
		}

		@Override
		public TaskScheduler build() {
			return builder.build();
		}
	};
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
public MagellanFramework(){
    log.log(Level.INFO, "Printing Environment Variables:");
    log.log(Level.INFO, "PRINCIPAL: " + System.getenv("PRINCIPAL"));
    log.log(Level.INFO, "FRAMEWORK_USER: " + System.getenv("FRAMEWORK_USER"));
    log.log(Level.INFO, "MASTER_ADDRESS: " + System.getenv("MASTER_ADDRESS"));
    log.log(Level.INFO, "LIBPROCESS_IP: " + System.getenv("LIBPROCESS_IP"));
    log.log(Level.INFO, "LIBPROCESS_PORT: " + System.getenv("LIBPROCESS_PORT"));
    log.log(Level.INFO, "ZK_IP: " + System.getenv("ZK_IP"));
    log.log(Level.INFO, "ZK_PORT: " + System.getenv("ZK_PORT"));
    log.log(Level.INFO, "EXECUTOR_PATH: " + System.getenv("EXECUTOR_PATH"));

    fenzoScheduler = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000000)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                public void call(VirtualMachineLease lease) {
                    log.log(Level.INFO, "Declining offer on " + lease.hostname());
                    mesosDriver.get().declineOffer(lease.getOffer().getId());
                }
            })
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
public AssignableVirtualMachine(ConcurrentMap<String, String> vmIdToHostnameMap,
                                ConcurrentMap<String, String> leaseIdToHostnameMap,
                                String hostname, Action1<VirtualMachineLease> leaseRejectAction,
                                long leaseOfferExpirySecs, TaskTracker taskTracker, boolean singleLeaseMode) {
    this.vmIdToHostnameMap = vmIdToHostnameMap;
    this.leaseIdToHostnameMap = leaseIdToHostnameMap;
    this.hostname = hostname;
    this.leaseRejectAction = getWrappedLeaseRejectAction(leaseRejectAction);
    this.leaseOfferExpirySecs = leaseOfferExpirySecs;
    this.taskTracker = taskTracker;
    this.leasesMap = new HashMap<>();
    this.leasesToExpire = new LinkedBlockingQueue<>();
    expireAllLeasesNow = new AtomicBoolean(false);
    this.workersToUnAssign = new LinkedBlockingQueue<>();
    this.previouslyAssignedTasksMap = new HashMap<>();
    this.assignmentResults = new HashMap<>();
    this.singleLeaseMode = singleLeaseMode;
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private static TaskScheduler getTaskScheduler() {
    return new TaskScheduler.Builder()
                .withFitnessGoodEnoughFunction(new Func1<Double, Boolean>() {
                    @Override
                    public Boolean call(Double aDouble) {
                        return aDouble >= GOOD_ENOUGH_FITNESS;
                    }
                })
                .withFitnessCalculator(BinPackingFitnessCalculators.cpuBinPacker)
                .withLeaseOfferExpirySecs(1000000)
                .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                    @Override
                    public void call(VirtualMachineLease lease) {
                        System.err.println("Unexpected to reject lease on " + lease.hostname());
                    }
                })
                .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Test
public void testOfferExpiry() throws Exception {
    final AtomicBoolean leaseRejected = new AtomicBoolean(false);
    final long leaseExpirySecs=1;
    TaskScheduler myTaskScheduler = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(leaseExpirySecs)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease virtualMachineLease) {
                    leaseRejected.set(true);
                }
            })
            .build();
    List<VirtualMachineLease> leases = LeaseProvider.getLeases(1, 4, 100, 1, 10);
    List<TaskRequest> taskRequests = new ArrayList<>();
    taskRequests.add(TaskRequestProvider.getTaskRequest(5, 10, 1));
    Map<String,VMAssignmentResult> resultMap = myTaskScheduler.scheduleOnce(taskRequests, leases).getResultMap();
    Assert.assertEquals(0, resultMap.entrySet().size());
    leases.clear(); // don't pass in the same lease again.
    // wait for lease to expire
    try{Thread.sleep(leaseExpirySecs*1000+200);}catch (InterruptedException ie){}
    taskRequests.clear();
    taskRequests.add(TaskRequestProvider.getTaskRequest(5, 10, 1)); // make sure task doesn't get assigned
    resultMap = myTaskScheduler.scheduleOnce(taskRequests, leases).getResultMap();
    Assert.assertEquals(true, leaseRejected.get());
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    taskScheduler = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease virtualMachineLease) {
                    System.out.println("Rejecting offer on host " + virtualMachineLease.hostname());
                }
            })
            .build();
    taskScheduler.setActiveVmGroupAttributeName(activeVmGrpAttrName);
    taskScheduler.setActiveVmGroups(Arrays.asList(activeVmGrp));
    Protos.Attribute attribute1 = Protos.Attribute.newBuilder().setName(activeVmGrpAttrName)
            .setType(Protos.Value.Type.TEXT)
            .setText(Protos.Value.Text.newBuilder().setValue("test1")).build();
    attributes1.put(activeVmGrpAttrName, attribute1);
    Protos.Attribute attribute2 = Protos.Attribute.newBuilder().setName(activeVmGrpAttrName)
            .setType(Protos.Value.Type.TEXT)
            .setText(Protos.Value.Text.newBuilder().setValue("test2")).build();
    attributes2.put(activeVmGrpAttrName, attribute2);
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getTaskScheduler(boolean useBinPacking) {
    TaskScheduler.Builder builder = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease virtualMachineLease) {
                    System.out.println("Rejecting offer on host " + virtualMachineLease.hostname());
                    Assert.fail();
                }
            })
            .withFitnessGoodEnoughFunction(new Func1<Double, Boolean>() {
                @Override
                public Boolean call(Double aDouble) {
                    return aDouble > 1.0;
                }
            });
    if(useBinPacking)
        builder = builder
                .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker);
    return builder.build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
static TaskScheduler getScheduler(final Action1<VirtualMachineLease> leaseRejectCalback, final Action1<AutoScaleAction> callback,
                                  long delayScaleUpBySecs, long delayScaleDownByDecs,
                                  AutoScaleRule... rules) {
    TaskScheduler.Builder builder = new TaskScheduler.Builder()
            .withAutoScaleByAttributeName(hostAttrName);
    for (AutoScaleRule rule : rules)
        builder.withAutoScaleRule(rule);
    if (callback != null)
        builder.withAutoScalerCallback(callback);
    return builder
            .withDelayAutoscaleDownBySecs(delayScaleDownByDecs)
            .withDelayAutoscaleUpBySecs(delayScaleUpBySecs)
            .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker)
            .withLeaseOfferExpirySecs(3600)
            .withLeaseRejectAction(lease -> {
                if (leaseRejectCalback == null)
                    Assert.fail("Unexpected to reject lease " + lease.hostname());
                else
                    leaseRejectCalback.call(lease);
            })
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getScheduler(
        long offerExpirySecs, final long leaseReOfferDelaySecs,
        int maxOffersToReject,
        final BlockingQueue<VirtualMachineLease> offersQ,
        final Func1<String, VirtualMachineLease> offerGenerator
) {
    return new TaskScheduler.Builder()
            .withLeaseRejectAction(
                    new Action1<VirtualMachineLease>() {
                        @Override
                        public void call(final VirtualMachineLease virtualMachineLease) {
                            executorService.schedule(
                                    new Runnable() {
                                        @Override
                                        public void run() {
                                            offersQ.offer(offerGenerator.call(virtualMachineLease.hostname()));
                                        }
                                    },
                                    leaseReOfferDelaySecs, TimeUnit.SECONDS);
                        }
                    }
            )
            .withLeaseOfferExpirySecs(offerExpirySecs)
            .withMaxOffersToReject(maxOffersToReject)
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Test
public void testExpiryOfLease() throws Exception {
    final AtomicInteger expireCount = new AtomicInteger();
    final TaskScheduler scheduler = new TaskScheduler.Builder()
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease virtualMachineLease) {
                    expireCount.incrementAndGet();
                }
            })
            .withLeaseOfferExpirySecs(1000000)
            .build();
    final VirtualMachineLease lease1 = LeaseProvider.getLeaseOffer("host1", 2, 2000, 1, 10);
    scheduler.scheduleOnce(Collections.<TaskRequest>emptyList(), Collections.singletonList(lease1));
    Thread.sleep(100);
    scheduler.expireLease(lease1.getId());
    List<TaskRequest> tasks = new ArrayList<>();
    tasks.add(TaskRequestProvider.getTaskRequest(2, 1000, 1));
    final SchedulingResult result = scheduler.scheduleOnce(tasks, Collections.<VirtualMachineLease>emptyList());
    Assert.assertEquals(0, result.getResultMap().size());
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskSchedulingService getSchedulingService(TaskQueue queue, TaskScheduler scheduler, long loopMillis,
                                                   long maxDelayMillis, Action1<SchedulingResult> resultCallback) {
    return new TaskSchedulingService.Builder()
            .withTaskQueue(queue)
            .withLoopIntervalMillis(loopMillis)
            .withMaxDelayMillis(maxDelayMillis)
            .withPreSchedulingLoopHook(new Action0() {
                @Override
                public void call() {
                    //System.out.println("Pre-scheduling hook");
                }
            })
            .withSchedulingResultCallback(resultCallback)
            .withTaskScheduler(scheduler)
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Test
public void clonePseudoVMsForGroups() throws Exception {
    AutoScaleRule rule = getAutoScaleRule(attributeVal1, 4);
    Action1<VirtualMachineLease> leaseRejectAction = l -> {};
    final ConcurrentMap<String, String> vmIdTohostNames = new ConcurrentHashMap<>();
    final ConcurrentMap<String, String> leasesToHostnames = new ConcurrentHashMap<>();
    final Map<String, AssignableVirtualMachine> avms = new HashMap<>();
    final TaskTracker taskTracker = new TaskTracker();
    VMCollection vms = createVmCollection(vmIdTohostNames, leasesToHostnames, taskTracker, avms);
    for (int i=0; i < rule.getMaxSize()-1; i++) {
        vms.addLease(LeaseProvider.getLeaseOffer("host"+i, 4, 4000,
                0, 0, ports, attributes1, Collections.singletonMap("GPU", 1.0)));
    }
    for (AssignableVirtualMachine avm: avms.values())
        avm.updateCurrTotalLease();
    final Map<String, List<String>> map = vms.clonePseudoVMsForGroups(
            Collections.singletonMap(rule.getRuleName(), 6), s -> rule,
            lease -> true
    );
    Assert.assertNotNull(map);
    Assert.assertEquals(1, map.size());
    Assert.assertEquals(rule.getRuleName(), map.keySet().iterator().next());
    Assert.assertEquals(1, map.values().iterator().next().size());
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private VMCollection createVmCollection(ConcurrentMap<String, String> vmIdTohostNames,
                                        ConcurrentMap<String, String> leasesToHostnames, TaskTracker taskTracker,
                                        Map<String, AssignableVirtualMachine> avms) {
    Action1<VirtualMachineLease> leaseRejectAction = l -> {};
    return new VMCollection(s -> {
        AssignableVirtualMachine avm = new AssignableVirtualMachine(
                vmIdTohostNames,
                leasesToHostnames,
                s,
                leaseRejectAction,
                2,
                taskTracker,
                false
        );
        avms.put(s, avm);
        return avm;
    }, AutoScalerTest.hostAttrName);
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getTaskSchedulerWithAutoscale(boolean useBinPacking, Action1<AutoScaleAction> callback, AutoScaleRule... rules) {
    final TaskScheduler.Builder builder = new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease virtualMachineLease) {
                    System.out.println("Rejecting offer on host " + virtualMachineLease.hostname());
                }
            })
            .withAutoScaleByAttributeName(hostAttrName)
            .withFitnessGoodEnoughFunction(new Func1<Double, Boolean>() {
                @Override
                public Boolean call(Double aDouble) {
                    return aDouble > 1.0;
                }
            });
    if(callback!=null && rules!=null) {
        builder.withAutoScalerCallback(callback);
        for (AutoScaleRule r : rules)
            builder.withAutoScaleRule(r);
    }
    if(useBinPacking)
        builder
                .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker);
    return builder.build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getScheduler() {
    Map<String, ResAllocs> resAllocs = new HashMap<>();
    resAllocs.put(grp1, ResAllocsProvider.create(grp1, 4, 4000, Double.MAX_VALUE, Double.MAX_VALUE));
    resAllocs.put(grp2, ResAllocsProvider.create(grp2, 8, 8000, Double.MAX_VALUE, Double.MAX_VALUE));
    return new TaskScheduler.Builder()
            .withInitialResAllocs(resAllocs)
            .withFitnessCalculator(BinPackingFitnessCalculators.cpuBinPacker)
            .withLeaseOfferExpirySecs(3600)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease lease) {
                    Assert.fail("Unexpected to reject lease " + lease.hostname());
                }
            })
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getAutoscalingScheduler(AutoScaleRule... rules) {
    TaskScheduler.Builder builder = new TaskScheduler.Builder()
            .withAutoScaleByAttributeName(hostAttrName);
    Map<String, ResAllocs> resAllocs = new HashMap<>();
    resAllocs.put(grp1, ResAllocsProvider.create(grp1, cpus1, memory1, Double.MAX_VALUE, Double.MAX_VALUE));
    resAllocs.put(grp2, ResAllocsProvider.create(grp2, cpus1 * 2, memory1 * 2, Double.MAX_VALUE, Double.MAX_VALUE));
    for(AutoScaleRule rule: rules)
        builder.withAutoScaleRule(rule);
    return builder
            .withInitialResAllocs(resAllocs)
            .withFitnessCalculator(BinPackingFitnessCalculators.cpuBinPacker)
            .withLeaseOfferExpirySecs(3600)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                @Override
                public void call(VirtualMachineLease lease) {
                    Assert.fail("Unexpected to reject lease " + lease.hostname());
                }
            })
            .build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Override
public Action1<Map<String, Map<VMResource, Double[]>>> getPrintAction() {
    return new Action1<Map<String, Map<VMResource, Double[]>>>() {
        @Override
        public void call(Map<String, Map<VMResource, Double[]>> stringMapMap) {
            int same=0;
            int diff=0;
            int unused=0;
            for(String hostname: stringMapMap.keySet()) {
                Set<RandomTaskGenerator.GeneratedTask> generatedTasks = hostToTasksMap.get(hostname);
                if(generatedTasks==null)
                    unused++;
                else if(isSame(generatedTasks))
                    same++;
                else
                    diff++;
            }
            outputStream.printf("%d, %d, %d\n", unused, same, diff);
            System.out.printf("Unused=%d, same=%d, different=%d\n", unused, same, diff);
        }
    };
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
private TaskScheduler getTaskScheduler() {
    return new TaskScheduler.Builder()
            .withLeaseOfferExpirySecs(1000000000L)
            .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                
                @Override
                public void call(final VirtualMachineLease lease) {
                    log.warn("Declining offer on '{}'", lease.hostname());
                    schedulerDriver.declineOffer(lease.getOffer().getId());
                }
            }).build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    taskScheduler = new TaskScheduler.Builder().withLeaseOfferExpirySecs(1000000000L).withLeaseRejectAction(new Action1<VirtualMachineLease>() {
        
        @Override
        public void call(final VirtualMachineLease virtualMachineLease) {
        }
    }).build();
}
 

import com.netflix.fenzo.functions.Action1; //导入依赖的package包/类
/**
 * Constructor for the framework
 * 
 * @param processor
 *            The processor that this framework is responsible for
 * @param configuration
 *            Configuration that came from environment variables
 */
public Framework(Processor processor, Configuration configuration) {
	this.configuration = configuration;
	this.leasesQueue = new LinkedBlockingQueue<>();

	scheduler = new TaskScheduler.Builder().withLeaseOfferExpirySecs(10)
			.withLeaseRejectAction(new Action1<VirtualMachineLease>() {
				@Override
				public void call(VirtualMachineLease lease) {
					LOG.trace("Declining offer on " + lease.hostname());
					ref.get().declineOffer(lease.getOffer().getId());
				}
			}).build();

	// Construct the Framework
	String frameworkName = configuration.frameworkName + "_" + processor.name + "_" + VERSION;
	Protos.FrameworkInfo framework = Protos.FrameworkInfo.newBuilder().setName(frameworkName).setUser("").build();

	// Build the Kafka components
	input = new KafkaInput(this, configuration, processor);
	output = new KafkaOutput(this, configuration, processor);

	// Build the Scheduler
	Scheduler mesosScheduler = new MesosScheduler(scheduler, leasesQueue, input, this);
	mesosSchedulerDriver = new MesosSchedulerDriver(mesosScheduler, framework, configuration.mesosMaster);
	ref.set(mesosSchedulerDriver);
}