FenzoCapacityCalculator.java

package com.netflix.fenzo;

import com.netflix.fenzo.plugins.SpreadingFitnessCalculators;
import org.apache.mesos.Protos;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;

public class FenzoCapacityCalculator {

    private final static String ZONE = "Zone";
    private final static String ROLE = "Role";


    private static final Logger logger = LoggerFactory.getLogger(FenzoCapacityCalculator.class);

    public static void main(String[] args) {

        Map<String, Protos.Attribute> zoneACompute = new HashMap<>();
        Map<String, Protos.Attribute> zoneBCompute = new HashMap<>();
        Map<String, Protos.Attribute> zoneCCompute = new HashMap<>();
        Map<String, Protos.Attribute> zoneAEdge = new HashMap<>();
        Map<String, Protos.Attribute> zoneBEdge = new HashMap<>();
        Map<String, Protos.Attribute> zoneCEdge = new HashMap<>();

        Protos.Attribute compute = Protos.Attribute.newBuilder().setName(ROLE)
                .setType(Protos.Value.Type.TEXT)
                .setText(Protos.Value.Text.newBuilder().setValue("compute")).build();

        Protos.Attribute edge = Protos.Attribute.newBuilder().setName(ROLE)
                .setType(Protos.Value.Type.TEXT)
                .setText(Protos.Value.Text.newBuilder().setValue("edge")).build();

        Protos.Attribute zoneA = Protos.Attribute.newBuilder().setName(ZONE)
                .setType(Protos.Value.Type.TEXT)
                .setText(Protos.Value.Text.newBuilder().setValue("zoneA")).build();

        Protos.Attribute zoneB = Protos.Attribute.newBuilder().setName(ZONE)
                .setType(Protos.Value.Type.TEXT)
                .setText(Protos.Value.Text.newBuilder().setValue("zoneB")).build();

        Protos.Attribute zoneC = Protos.Attribute.newBuilder().setName(ZONE)
                .setType(Protos.Value.Type.TEXT)
                .setText(Protos.Value.Text.newBuilder().setValue("zoneC")).build();

        zoneACompute.put(ZONE, zoneA);
        zoneACompute.put(ROLE, compute);

        zoneBCompute.put(ZONE, zoneB);
        zoneBCompute.put(ROLE, compute);

        zoneCCompute.put(ZONE, zoneC);
        zoneCCompute.put(ROLE, compute);


        zoneAEdge.put(ZONE, zoneA);
        zoneAEdge.put(ROLE, edge);

        zoneBEdge.put(ZONE, zoneB);
        zoneBEdge.put(ROLE, edge);

        zoneCEdge.put(ZONE, zoneC);
        zoneCEdge.put(ROLE, edge);


        TaskScheduler taskScheduler = new TaskScheduler.Builder()
//                .withFitnessCalculator(BinPackingFitnessCalculators.cpuMemBinPacker)
                .withFitnessCalculator(SpreadingFitnessCalculators.cpuMemSpreader)
                .withLeaseOfferExpirySecs(10)
                .withLeaseRejectAction(virtualMachineLease -> logger.info("Rejecting lease on " + virtualMachineLease.hostname()))
                .build();


        List<VirtualMachineLease.Range> portRanges = Collections.singletonList(new VirtualMachineLease.Range(1, 10));
        List<VirtualMachineLease> leases = Arrays.asList(
                LeaseProvider.getLeaseOffer("compute-1", 2, 32, 4, 0, portRanges, zoneACompute)
                , LeaseProvider.getLeaseOffer("compute-2", 4, 64, 4,0, portRanges, zoneACompute)
                , LeaseProvider.getLeaseOffer("compute-3", 4, 64, 4, 0, portRanges, zoneACompute)
                , LeaseProvider.getLeaseOffer("compute-4", 2, 32, 4, 0, portRanges, zoneBCompute)
                , LeaseProvider.getLeaseOffer("compute-5", 4, 64,4, 0,  portRanges, zoneBCompute)
                , LeaseProvider.getLeaseOffer("compute-6", 4, 64, 4, 0, portRanges, zoneBCompute)
                , LeaseProvider.getLeaseOffer("compute-7", 2, 32, 4, 0, portRanges, zoneCCompute)
                , LeaseProvider.getLeaseOffer("compute-8", 4, 64, 4, 0, portRanges, zoneCCompute)
                , LeaseProvider.getLeaseOffer("edge-1", 4, 64, 4, 0, portRanges, zoneAEdge)
                , LeaseProvider.getLeaseOffer("edge-2", 4, 64,4, 0,  portRanges, zoneBEdge)
                , LeaseProvider.getLeaseOffer("edge-3", 4, 64, 4, 0, portRanges, zoneCEdge)
        );

        ConstraintEvaluator zoneAConstraint = ConstraintsProvider.getHostAttributeHardConstraint(ZONE, "zoneA");
        ConstraintEvaluator zoneBConstraint = ConstraintsProvider.getHostAttributeHardConstraint(ZONE, "zoneB");
        ConstraintEvaluator zoneCConstraint = ConstraintsProvider.getHostAttributeHardConstraint(ZONE, "zoneC");
        ConstraintEvaluator computeConstraint = ConstraintsProvider.getHostAttributeHardConstraint(ROLE, "compute");
        ConstraintEvaluator edgeConstraint = ConstraintsProvider.getHostAttributeHardConstraint(ROLE, "edge");


        List<ConstraintEvaluator> zoneAComputeConstraints = Arrays.asList(zoneAConstraint, computeConstraint);
        List<ConstraintEvaluator> zoneBComputeConstraints = Arrays.asList(zoneBConstraint, computeConstraint);
        List<ConstraintEvaluator> zoneCComputeConstraints = Arrays.asList(zoneCConstraint, computeConstraint);
        List<ConstraintEvaluator> zoneAEdgeConstraints = Arrays.asList(zoneAConstraint, edgeConstraint);
        List<ConstraintEvaluator> zoneBEdgeConstraints = Arrays.asList(zoneBConstraint, edgeConstraint);
        List<ConstraintEvaluator> zoneCEdgeConstraints = Arrays.asList(zoneCConstraint, edgeConstraint);

        List<TaskRequest> taskRequests = Arrays.asList(
                TaskRequestProvider.getTaskRequest("app1", 1, 1, 1,0, 1, zoneAComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app1", 1, 1, 1,0, 1, zoneBComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app1", 1, 1, 1,0, 1, zoneCComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("foo1", 1, 2, 1,0, 1, zoneAComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("foo1", 1, 2, 1,0, 1, zoneBComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("foo1", 1, 2, 1,0, 1, zoneCComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("bar", 2, 4, 1,0, 1, zoneAComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("bar", 2, 4, 1,0, 1, zoneBComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("bar", 2, 4, 1,0, 1, zoneCComputeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app4-edge1", 2, 4, 1,0, 1, zoneAEdgeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app4-edge2", 2, 4, 1,0, 1, zoneBEdgeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app4-edge2", 2, 4, 1,0, 1, zoneCEdgeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app5-edge1", 1, 2, 1,0, 1, zoneAEdgeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app5-edge2", 1, 2, 1,0, 1, zoneBEdgeConstraints, Collections.emptyList())
                , TaskRequestProvider.getTaskRequest("app6-edge2", 1, 2, 1,0, 1, zoneCEdgeConstraints, Collections.emptyList())
        );


        SchedulingResult schedulingResult = taskScheduler.scheduleOnce(taskRequests, leases);
        schedulingResult.getResultMap().values()
                .stream()
                .map(vmAssignmentResult -> "Hostname: " + vmAssignmentResult.getHostname() + " TotalTasksAssigned: " + vmAssignmentResult.getTasksAssigned().stream().count())
                .forEach(System.out::println);

        for (VMAssignmentResult vmAssignmentResult : schedulingResult.getResultMap().values()) {
            vmAssignmentResult.getTasksAssigned()
                    .stream()
                    .map(taskAssignmentResult -> "Assigned TaskGroup:" + taskAssignmentResult.getRequest().taskGroupName() + " Hostname: " + taskAssignmentResult.getHostname() + " TaskId: " + taskAssignmentResult.getTaskId() + " Score: " + taskAssignmentResult.getFitness())
                    .forEach(System.out::println);
        }
        schedulingResult.getFailures()
                .keySet()
                .stream()
                .map(taskRequest -> "UnAssigned TaskGroup: " + taskRequest.taskGroupName() + " TaskId: " + taskRequest.getId())
                .forEach(System.err::println);

        Map<String, Long> assignmentCountPerHost = schedulingResult.getResultMap().values()
                .stream()
                .collect(Collectors.groupingBy(VMAssignmentResult::getHostname, Collectors.counting()));

        System.out.println("Tasks Per VM: " +  assignmentCountPerHost);
        System.out.println("UnUsed VMs: " + schedulingResult.getIdleVMsCount());
        taskScheduler.shutdown();

    }

}