fenzo-core/src/main/java/com/netflix/fenzo/samples/SampleQbasedScheduling.java
public static void main(String[] args) throws Exception {
        if(args.length!=1) {
            System.err.println("Must provide one argument - Mesos master location string");
            System.exit(1);
        }
        int numTasks=10;
        final AtomicInteger numTasksCompleted = new AtomicInteger();

        // create Fenzo TaskScheduler object.
        final TaskScheduler taskScheduler = new TaskScheduler.Builder()
                .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker)
                .withLeaseOfferExpirySecs(1000000)
                .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                    @Override
                    public void call(VirtualMachineLease v) {
                        System.out.println("Unexpected to reject lease on " + v.hostname());
                    }
                })
                .build();

        // Create a queue from Fenzo provided queue implementations.
        final TaskQueue queue = TaskQueues.createTieredQueue(2);

        // Create our Mesos scheduler callback implementation
        AtomicReference<TaskSchedulingService> schedSvcGetter = new AtomicReference<>();
        final MesosScheduler mesosSchedulerCallback = new MesosScheduler(numTasksCompleted, schedSvcGetter);

        // create Mesos driver
        Protos.FrameworkInfo framework = Protos.FrameworkInfo.newBuilder()
                .setName("Sample Fenzo Framework")
                .setUser("")
                .build();
        final MesosSchedulerDriver driver = new MesosSchedulerDriver(mesosSchedulerCallback, framework, args[0]);

        // Build Fenzo task scheduling service using the TaskScheduler and TaskQueue objects created above
        final AtomicInteger schedCounter = new AtomicInteger();
        final TaskSchedulingService schedulingService = new TaskSchedulingService.Builder()
                .withLoopIntervalMillis(1000)
                .withMaxDelayMillis(1500)
                .withPreSchedulingLoopHook(new Action0() {
                    @Override
                    public void call() {
                        System.out.println("Starting scheduling iteration " + schedCounter.incrementAndGet());
                    }
                })
                .withTaskQueue(queue)
                .withTaskScheduler(taskScheduler)
                // TaskSchedulingService will call us back when there are task assignments. Handle them by launching
                // tasks using MesosDriver
                .withSchedulingResultCallback(new Action1<SchedulingResult>() {
                    @Override
                    public void call(SchedulingResult schedulingResult) {
                        final List<Exception> exceptions = schedulingResult.getExceptions();
                        if (exceptions != null && !exceptions.isEmpty()) {
                            System.out.println("Exceptions from scheduling iteration:");
                            for (Exception e: exceptions)
                                e.printStackTrace();
                        }
                        else {
                            for (Map.Entry<String, VMAssignmentResult> e: schedulingResult.getResultMap().entrySet()) {
                                List<Protos.OfferID> offers = new ArrayList<Protos.OfferID>();
                                for (VirtualMachineLease l: e.getValue().getLeasesUsed())
                                    offers.add(l.getOffer().getId());
                                List<Protos.TaskInfo> taskInfos = new ArrayList<Protos.TaskInfo>();
                                for (TaskAssignmentResult r: e.getValue().getTasksAssigned()) {
                                    taskInfos.add(SampleFramework.getTaskInfo(
                                            e.getValue().getLeasesUsed().iterator().next().getOffer().getSlaveId(),
                                            r.getTaskId(),
                                            "sleep 2"
                                    ));
                                    tasksToHostnameMap.put(r.getTaskId(), r.getHostname());
                                }
                                driver.launchTasks(
                                        offers,
                                        taskInfos
                                );
                            }
                        }
                    }
                })
                .build();
        schedSvcGetter.set(schedulingService);

        // set up action in our scheduler callback to send resource offers into our scheduling service
        mesosSchedulerCallback.leaseAction = new Action1<List<Protos.Offer>>() {
            @Override
            public void call(List<Protos.Offer> offers) {
                List<VirtualMachineLease> leases = new ArrayList<>();
                for (Protos.Offer o: offers)
                    leases.add(new VMLeaseObject(o)); // Fenzo lease object adapter for Mesos Offer object
                schedulingService.addLeases(leases);
            }
        };
        schedulingService.start();

        // start Mesos driver
        new Thread() {
            @Override
            public void run() {
                driver.run();
            }
        }.start();

        // submit some tasks
        for (int i=0; i<numTasks; i++) {
            final QueuableTask task = getTask(i);
            allTasks.put(task.getId(), task);
            queue.queueTask(task);
        }

        // wait for tasks to complete
        while (numTasksCompleted.get() < numTasks) {
            Thread.sleep(1000);
            System.out.println("        #tasks completed: " + numTasksCompleted.get() + " of " + numTasks);
        }

        // verify that Fenzo has no tasks in its queue
        final CountDownLatch latch = new CountDownLatch(1);
        schedulingService.requestAllTasks(new Action1<Map<TaskQueue.TaskState, Collection<QueuableTask>>>() {
            @Override
            public void call(Map<TaskQueue.TaskState, Collection<QueuableTask>> taskStateCollectionMap) {
                System.out.println("Fenzo queue has " + taskStateCollectionMap.size() + " items");
                latch.countDown();
            }
        });
        if (!latch.await(5, TimeUnit.SECONDS))
            System.err.println("Timeout waiting for listing all tasks in Fenzo queues");

        System.out.println("ALL DONE");
        System.exit(0);
    }

    private static QueuableTask getTask(final int i) {
        return new QueuableTask() {
            @Override
            public QAttributes getQAttributes() {
                return qAttribs;
            }

            @Override
            public String getId() {
                return "Task-" + i;
            }

            @Override
            public String taskGroupName() {
                return "groupA";
            }

            @Override
            public double getCPUs() {
                return 1;
            }

            @Override
            public double getMemory() {
                return 100;
            }

            @Override
            public double getNetworkMbps() {
                return 0;
            }

            @Override
            public double getDisk() {
                return 0;
            }

            @Override
            public int getPorts() {
                return 0;
            }

            @Override
            public Map<String, Double> getScalarRequests() {
                return null;
            }

            @Override
            public Map<String, NamedResourceSetRequest> getCustomNamedResources() {
                return null;
            }

            @Override
            public List<? extends ConstraintEvaluator> getHardConstraints() {
                return null;
            }

            @Override
            public List<? extends VMTaskFitnessCalculator> getSoftConstraints() {
                return null;
            }

            @Override
            public void setAssignedResources(AssignedResources assignedResources) {
                // no-op, unexpected for this sample
            }

            @Override
            public AssignedResources getAssignedResources() {
                return null;
            }
        };
    }
}
public static void main(String[] args) throws Exception {
        if(args.length!=1) {
            System.err.println("Must provide one argument - Mesos master location string");
            System.exit(1);
        }
        int numTasks=10;
        final AtomicInteger numTasksCompleted = new AtomicInteger();

        // create Fenzo TaskScheduler object.
        final TaskScheduler taskScheduler = new TaskScheduler.Builder()
                .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker)
                .withLeaseOfferExpirySecs(1000000)
                .withLeaseRejectAction(new Action1<VirtualMachineLease>() {
                    @Override
                    public void call(VirtualMachineLease v) {
                        System.out.println("Unexpected to reject lease on " + v.hostname());
                    }
                })
                .build();

        // Create a queue from Fenzo provided queue implementations.
        final TaskQueue queue = TaskQueues.createTieredQueue(2);

        // Create our Mesos scheduler callback implementation
        AtomicReference<TaskSchedulingService> schedSvcGetter = new AtomicReference<>();
        final MesosScheduler mesosSchedulerCallback = new MesosScheduler(numTasksCompleted, schedSvcGetter);

        // create Mesos driver
        Protos.FrameworkInfo framework = Protos.FrameworkInfo.newBuilder()
                .setName("Sample Fenzo Framework")
                .setUser("")
                .build();
        final MesosSchedulerDriver driver = new MesosSchedulerDriver(mesosSchedulerCallback, framework, args[0]);

        // Build Fenzo task scheduling service using the TaskScheduler and TaskQueue objects created above
        final AtomicInteger schedCounter = new AtomicInteger();
        final TaskSchedulingService schedulingService = new TaskSchedulingService.Builder()
                .withLoopIntervalMillis(1000)
                .withMaxDelayMillis(1500)
                .withPreSchedulingLoopHook(new Action0() {
                    @Override
                    public void call() {
                        System.out.println("Starting scheduling iteration " + schedCounter.incrementAndGet());
                    }
                })
                .withTaskQueue(queue)
                .withTaskScheduler(taskScheduler)
                // TaskSchedulingService will call us back when there are task assignments. Handle them by launching
                // tasks using MesosDriver
                .withSchedulingResultCallback(new Action1<SchedulingResult>() {
                    @Override
                    public void call(SchedulingResult schedulingResult) {
                        final List<Exception> exceptions = schedulingResult.getExceptions();
                        if (exceptions != null && !exceptions.isEmpty()) {
                            System.out.println("Exceptions from scheduling iteration:");
                            for (Exception e: exceptions)
                                e.printStackTrace();
                        }
                        else {
                            for (Map.Entry<String, VMAssignmentResult> e: schedulingResult.getResultMap().entrySet()) {
                                List<Protos.OfferID> offers = new ArrayList<Protos.OfferID>();
                                for (VirtualMachineLease l: e.getValue().getLeasesUsed())
                                    offers.add(l.getOffer().getId());
                                List<Protos.TaskInfo> taskInfos = new ArrayList<Protos.TaskInfo>();
                                for (TaskAssignmentResult r: e.getValue().getTasksAssigned()) {
                                    taskInfos.add(SampleFramework.getTaskInfo(
                                            e.getValue().getLeasesUsed().iterator().next().getOffer().getSlaveId(),
                                            r.getTaskId(),
                                            "sleep 2"
                                    ));
                                    tasksToHostnameMap.put(r.getTaskId(), r.getHostname());
                                }
                                driver.launchTasks(
                                        offers,
                                        taskInfos
                                );
                            }
                        }
                    }
                })
                .build();
        schedSvcGetter.set(schedulingService);

        // set up action in our scheduler callback to send resource offers into our scheduling service
        mesosSchedulerCallback.leaseAction = new Action1<List<Protos.Offer>>() {
            @Override
            public void call(List<Protos.Offer> offers) {
                List<VirtualMachineLease> leases = new ArrayList<>();
                for (Protos.Offer o: offers)
                    leases.add(new VMLeaseObject(o)); // Fenzo lease object adapter for Mesos Offer object
                schedulingService.addLeases(leases);
            }
        };
        schedulingService.start();

        // start Mesos driver
        new Thread() {
            @Override
            public void run() {
                driver.run();
            }
        }.start();

        // submit some tasks
        for (int i=0; i<numTasks; i++) {
            final QueuableTask task = getTask(i);
            allTasks.put(task.getId(), task);
            queue.queueTask(task);
        }

        // wait for tasks to complete
        while (numTasksCompleted.get() < numTasks) {
            Thread.sleep(1000);
            System.out.println("        #tasks completed: " + numTasksCompleted.get() + " of " + numTasks);
        }

        // verify that Fenzo has no tasks in its queue
        final CountDownLatch latch = new CountDownLatch(1);
        schedulingService.requestAllTasks(new Action1<Map<TaskQueue.TaskState, Collection<QueuableTask>>>() {
            @Override
            public void call(Map<TaskQueue.TaskState, Collection<QueuableTask>> taskStateCollectionMap) {
                System.out.println("Fenzo queue has " + taskStateCollectionMap.size() + " items");
                latch.countDown();
            }
        });
        if (!latch.await(5, TimeUnit.SECONDS))
            System.err.println("Timeout waiting for listing all tasks in Fenzo queues");

        System.out.println("ALL DONE");
        System.exit(0);
    }

    private static QueuableTask getTask(final int i) {
        return new QueuableTask() {
            @Override
            public QAttributes getQAttributes() {
                return qAttribs;
            }

            @Override
            public String getId() {
                return "Task-" + i;
            }

            @Override
            public String taskGroupName() {
                return "groupA";
            }

            @Override
            public double getCPUs() {
                return 1;
            }

            @Override
            public double getMemory() {
                return 100;
            }

            @Override
            public double getNetworkMbps() {
                return 0;
            }

            @Override
            public double getDisk() {
                return 0;
            }

            @Override
            public int getPorts() {
                return 0;
            }

            @Override
            public Map<String, Double> getScalarRequests() {
                return null;
            }

            @Override
            public Map<String, NamedResourceSetRequest> getCustomNamedResources() {
                return null;
            }

            @Override
            public List<? extends ConstraintEvaluator> getHardConstraints() {
                return null;
            }

            @Override
            public List<? extends VMTaskFitnessCalculator> getSoftConstraints() {
                return null;
            }

            @Override
            public void setAssignedResources(AssignedResources assignedResources) {
                // no-op, unexpected for this sample
            }

            @Override
            public AssignedResources getAssignedResources() {
                return null;
            }
        };
    }
}
public static void main ( String [ ] args ) throws Exception { if ( args . length != 1 ) { System . err . println ( "Must provide one argument - Mesos master location string" ) ; System . exit ( 1 ) ; } int numTasks = 10 ; final AtomicInteger numTasksCompleted = new AtomicInteger ( ) ; // create Fenzo TaskScheduler object. final TaskScheduler taskScheduler = new TaskScheduler . Builder ( ) . withFitnessCalculator ( BinPackingFitnessCalculators . cpuMemBinPacker ) . withLeaseOfferExpirySecs ( 1000000 ) . withLeaseRejectAction ( new Action1 < VirtualMachineLease > ( ) { @ Override public void call ( VirtualMachineLease v ) { System . out . println ( "Unexpected to reject lease on " + v . hostname ( ) ) ; } } ) . build ( ) ; // Create a queue from Fenzo provided queue implementations. final TaskQueue queue = TaskQueues . createTieredQueue ( 2 ) ; // Create our Mesos scheduler callback implementation AtomicReference < TaskSchedulingService > schedSvcGetter = new AtomicReference <> ( ) ; final MesosScheduler mesosSchedulerCallback = new MesosScheduler ( numTasksCompleted , schedSvcGetter ) ; // create Mesos driver Protos . FrameworkInfo framework = Protos . FrameworkInfo . newBuilder ( ) . setName ( "Sample Fenzo Framework" ) . setUser ( "" ) . build ( ) ; final MesosSchedulerDriver driver = new MesosSchedulerDriver ( mesosSchedulerCallback , framework , args [ 0 ] ) ; // Build Fenzo task scheduling service using the TaskScheduler and TaskQueue objects created above final AtomicInteger schedCounter = new AtomicInteger ( ) ; final TaskSchedulingService schedulingService = new TaskSchedulingService . Builder ( ) . withLoopIntervalMillis ( 1000 ) . withMaxDelayMillis ( 1500 ) . withPreSchedulingLoopHook ( new Action0 ( ) { @ Override public void call ( ) { System . out . println ( "Starting scheduling iteration " + schedCounter . incrementAndGet ( ) ) ; } } ) . withTaskQueue ( queue ) . withTaskScheduler ( taskScheduler ) // TaskSchedulingService will call us back when there are task assignments. Handle them by launching // tasks using MesosDriver . withSchedulingResultCallback ( new Action1 < SchedulingResult > ( ) { @ Override public void call ( SchedulingResult schedulingResult ) { final List < Exception > exceptions = schedulingResult . getExceptions ( ) ; if ( exceptions != null && ! exceptions . isEmpty ( ) ) { System . out . println ( "Exceptions from scheduling iteration:" ) ; for ( Exception e : exceptions ) e . printStackTrace ( ) ; } else { for ( Map . Entry < String , VMAssignmentResult > e : schedulingResult . getResultMap ( ) . entrySet ( ) ) { List < Protos . OfferID > offers = new ArrayList < Protos . OfferID > ( ) ; for ( VirtualMachineLease l : e . getValue ( ) . getLeasesUsed ( ) ) offers . ( l . getOffer ( ) . getId ( ) ) ; List < Protos . TaskInfo > taskInfos = new ArrayList < Protos . TaskInfo > ( ) ; for ( TaskAssignmentResult r : e . getValue ( ) . getTasksAssigned ( ) ) { taskInfos . add ( SampleFramework . getTaskInfo ( e . getValue ( ) . getLeasesUsed ( ) . iterator ( ) . next ( ) . getOffer ( ) . getSlaveId ( ) , r . getTaskId ( ) , "sleep 2" ) ) ; tasksToHostnameMap . put ( r . getTaskId ( ) , r . getHostname ( ) ) ; } driver . launchTasks ( offers , taskInfos ) ; } } } } ) . build ( ) ; schedSvcGetter . set ( schedulingService ) ; // set up action in our scheduler callback to send resource offers into our scheduling service mesosSchedulerCallback . leaseAction = new Action1 < List < Protos . Offer > > ( ) { @ Override public void call ( List < Protos . Offer > offers ) { List < VirtualMachineLease > leases = new ArrayList <> ( ) ; for ( Protos . Offer o : offers ) leases . ( new VMLeaseObject ( o ) ) ; // Fenzo lease object adapter for Mesos Offer object schedulingService . addLeases ( leases ) ; } } ; schedulingService . start ( ) ; // start Mesos driver new Thread ( ) { @ Override public void run ( ) { driver . run ( ) ; } } . start ( ) ; // submit some tasks for ( int i = 0 ; i < numTasks ; i ++ ) { final QueuableTask task = getTask ( i ) ; allTasks . put ( task . getId ( ) , task ) ; queue . queueTask ( task ) ; } // wait for tasks to complete while ( numTasksCompleted . get ( ) < numTasks ) { Thread . sleep ( 1000 ) ; System . out . println ( "        #tasks completed: " + numTasksCompleted . get ( ) + " of " + numTasks ) ; } // verify that Fenzo has no tasks in its queue final CountDownLatch latch = new CountDownLatch ( 1 ) ; schedulingService . requestAllTasks ( new Action1 < Map < TaskQueue . TaskState , Collection < QueuableTask > > > ( ) { @ Override public void call ( Map < TaskQueue . TaskState , Collection < QueuableTask > > taskStateCollectionMap ) { System . out . println ( "Fenzo queue has " + taskStateCollectionMap . size ( ) + " items" ) ; latch . countDown ( ) ; } } ) ; if ( ! latch . await ( 5 , TimeUnit . SECONDS ) ) System . err . println ( "Timeout waiting for listing all tasks in Fenzo queues" ) ; System . out . println ( "ALL DONE" ) ; System . exit ( 0 ) ; } private static QueuableTask getTask ( final int i ) { return new QueuableTask ( ) { @ Override public QAttributes getQAttributes ( ) { return qAttribs ; } @ Override public String getId ( ) { return "Task-" + i ; } @ Override public String taskGroupName ( ) { return "groupA" ; } @ Override public double getCPUs ( ) { return 1 ; } @ Override public double getMemory ( ) { return 100 ; } @ Override public double getNetworkMbps ( ) { return 0 ; } @ Override public double getDisk ( ) { return 0 ; } @ Override public int getPorts ( ) { return 0 ; } @ Override public Map < String , Double > getScalarRequests ( ) { return null ; } @ Override public Map < String , NamedResourceSetRequest > getCustomNamedResources ( ) { return null ; } @ Override public List < ? extends ConstraintEvaluator > getHardConstraints ( ) { return null ; } @ Override public List < ? extends VMTaskFitnessCalculator > getSoftConstraints ( ) { return null ; } @ Override public void setAssignedResources ( AssignedResources assignedResources ) { // no-op, unexpected for this sample } @ Override public AssignedResources getAssignedResources ( ) { return null ; } } ; } }
This is the main method of this sample framework. It showcases how to use Fenzo queues for scheduling. It creates
some number of tasks and launches them into Mesos using the Mesos built-in command executor. The tasks launched
are simple sleep commands that sleep for 3 seconds each.
@param args Requires one argument for the location of Mesos master to connect to.
@throws Exception Upon catching any exceptions within the program.
This is the main method of this sample framework . It showcases how to use Fenzo queues for scheduling . It creates some number of tasks and launches them into Mesos using the Mesos built - in command executor . The tasks launched are simple sleep commands that sleep for 3 seconds each .