import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> slots) {
  if (slots.size() == 0) {
    LOG.debug("assignSlots: no slots passed in, nothing to do");
    return;
  }
  for (Map.Entry<String, Collection<WorkerSlot>> topologyToSlots : slots.entrySet()) {
    String topologyId = topologyToSlots.getKey();
    for (WorkerSlot slot : topologyToSlots.getValue()) {
      TopologyDetails details = topologies.getById(topologyId);
      LOG.debug("assignSlots: topologyId: {} worker being assigned to slot: {} with workerCpu: {} workerMem: {}",
                topologyId, slot, MesosCommon.topologyWorkerCpu(_conf, details), MesosCommon.topologyWorkerMem(_conf, details));
    }
  }
  synchronized (_offersLock) {
    computeLaunchList(topologies, slots);
  }
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
private Cluster getSpyCluster(int numWorkers, int numExecutors) {
  Cluster spyCluster = getSpyCluster();

  List<MesosWorkerSlot> mesosWorkerSlots = this.generateMesosWorkerSlots(numWorkers);
  initializeMesosWorkerSlotMap(mesosWorkerSlots);

  Set<ExecutorDetails> executorsToAssign = this.generateExecutorDetailsSet(numExecutors);
  List<WorkerSlot> workerSlotList = getWorkerSlotFromMesosWorkerSlot(mesosWorkerSlots);
  topologyMap.put(sampleTopologyId, TestUtils.constructTopologyDetails(sampleTopologyId, numWorkers, 0.1, 100));
  this.topologies = new Topologies(topologyMap);

  doReturn(workerSlotList).when(spyCluster).getAvailableSlots();
  doReturn(executorsToAssign).when(spyCluster).getUnassignedExecutors(any(TopologyDetails.class));

  return spyCluster;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Test
public void testScheduleWithOneWorkerSlot() {
  Cluster spyCluster = getSpyCluster();

  List<MesosWorkerSlot> mesosWorkerSlots = this.generateMesosWorkerSlots(1);
  initializeMesosWorkerSlotMap(mesosWorkerSlots);

  Set<ExecutorDetails> executorsToAssign = this.generateExecutorDetailsSet(4);
  List<WorkerSlot> workerSlotList = getWorkerSlotFromMesosWorkerSlot(mesosWorkerSlots);

  doReturn(workerSlotList).when(spyCluster).getAvailableSlots();
  doReturn(executorsToAssign).when(spyCluster).getUnassignedExecutors(any(TopologyDetails.class));

  defaultScheduler.schedule(topologies, spyCluster);

  Set<ExecutorDetails> assignedExecutors = spyCluster.getAssignmentById(sampleTopologyId).getExecutors();
  assertEquals(executorsToAssign, assignedExecutors);
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public void schedule(Topologies topologies, Cluster cluster) {
	
	logger.info("Online Scheduler");
	logger.info("+++++++++++++++++++++++++++");
	if (!topologies.getTopologies().isEmpty()) {
		int rescheduleTimeout = DEFAULT_RESCHEDULE_TIMEOUT;
		for (TopologyDetails topology : topologies.getTopologies()) {
			rescheduleTimeout = Integer.parseInt(topology.getConf().get(Utils.RESCHEDULE_TIMEOUT).toString());
		}
		long now = System.currentTimeMillis();
		long elapsedTime = (now - lastRescheduling) / 1000; // s
		if (lastRescheduling == 0 || elapsedTime >= rescheduleTimeout)
			doSchedule(topologies, cluster);
		else
			logger.info("It's not time to reschedule yet, " + elapsedTime + " seconds have passed, other " + (rescheduleTimeout - elapsedTime) + " seconds have to pass");
	}

	logger.info("---------------------------");
	
	logger.info("Calling EvenScheduler to schedule remaining executors...");
	new EvenScheduler().schedule(topologies, cluster);
	logger.info("Ok, EvenScheduler succesfully called");
	
	assignmentTracker.checkAssignment(topologies, cluster);
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public void addTopology(TopologyDetails td) {
    String topId = td.getId();
    LOG.debug("Adding in Topology {}", topId);
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    Set<Node> assignedNodes = new HashSet<Node>();
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            assignedNodes.add(n);
        }
    }
    _usedNodes += assignedNodes.size();
    _topologyIdToNodes.put(topId, assignedNodes);
    _tds.put(topId, td);
    if (td.getConf().get(Config.TOPOLOGY_ISOLATED_MACHINES) != null) {
        _isolated.add(topId);
    }
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Get all running topology's details.
 *
 * @param conf storm's conf, used to connect nimbus
 * @return null if nimbus is not available, otherwise all TopologyDetails
 */
public static Topologies getTopologyDetails(Map<String, Object> conf) {
    NimbusClient nimbusClient = NimbusClient.getConfiguredClient(conf);
    try {
        Nimbus.Client nimbus = nimbusClient.getClient();
        Map<String, TopologyDetails> topologies = nimbus.getClusterInfo().get_topologies().stream()
                .map(topoSummary -> getTopologyDetails(nimbus, topoSummary))
                .filter(Objects::nonNull)
                .collect(Collectors.toMap(topoDetails -> topoDetails.getId(), topoDetails -> topoDetails));
        return new Topologies(topologies);
    } catch (TException e) {
    } finally {
        nimbusClient.close();
    }
    return null;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Get component tasks from TopologyDetails object
 *
 * @param topoDetails
 * @return
 */
private static Map<String, List<Integer>> componentToTasks(TopologyDetails topoDetails,
                                                           Predicate<Map.Entry<ExecutorDetails, String>> p) {
    Stream<Map.Entry<ExecutorDetails, String>> stream = topoDetails.getExecutorToComponent().entrySet().stream();
    if (p != null) {
        stream = stream.filter(p);
    }
    return stream.collect(Collectors.groupingBy(Map.Entry::getValue,
                    Collectors.mapping(e -> getTaskIds(e.getKey()),
                            Collector.of(ArrayList::new, (all, l) -> all.addAll(l),
                                    (all, l) -> {
                                        all.addAll(l);
                                        return all;
                                    }
                            )
                    )
            )
    );
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public void addTopology(TopologyDetails td) {
    String topId = td.getId();
    LOG.debug("Adding in topology {}", topId);
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    Set<Node> assignedNodes = new HashSet<>();
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            assignedNodes.add(n);
        }
    }
    _usedNodes += assignedNodes.size();
    _topologyIdToNodes.put(topId, assignedNodes);
    _tds.put(topId, td);
    if (td.getConf().get(Config.TOPOLOGY_ISOLATED_MACHINES) != null) {
        _isolated.add(topId);
    }
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Schedule function looks in the "mesosWorkerSlotMap" to determine which topology owns the particular
 * WorkerSlot and assigns the executors accordingly.
 */
@Override
public void schedule(Topologies topologies, Cluster cluster) {
  List<WorkerSlot> workerSlots = cluster.getAvailableSlots();
  Map<String, List<MesosWorkerSlot>> perTopologySlotList = getMesosWorkerSlotPerTopology(workerSlots);

  // So far we know how many MesosSlots each of the topologies have got. Lets assign executors for each of them
  for (String topologyId : perTopologySlotList.keySet()) {
    TopologyDetails topologyDetails = topologies.getById(topologyId);
    List<MesosWorkerSlot> mesosWorkerSlots = perTopologySlotList.get(topologyId);

    int countSlotsRequested = topologyDetails.getNumWorkers();
    int countSlotsAssigned = cluster.getAssignedNumWorkers(topologyDetails);

    if (mesosWorkerSlots.size() == 0) {
      log.warn("No slots found for topology {} while scheduling", topologyId);
      continue;
    }

    int countSlotsAvailable = Math.min(mesosWorkerSlots.size(), (countSlotsRequested - countSlotsAssigned));

    List<List<ExecutorDetails>> executorsPerWorkerList = executorsPerWorkerList(cluster, topologyDetails, countSlotsAvailable);

    for (int i = 0; i < countSlotsAvailable; i++) {
      cluster.assign(mesosWorkerSlots.remove(0), topologyId, executorsPerWorkerList.remove(0));
    }
  }
  mesosWorkerSlotMap.clear();
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
public static boolean isFit(Map mesosStormConf, OfferResources offerResources, TopologyDetails topologyDetails, boolean supervisorExists) {
  double requestedWorkerCpu = MesosCommon.topologyWorkerCpu(mesosStormConf, topologyDetails);
  double requestedWorkerMem = MesosCommon.topologyWorkerMem(mesosStormConf, topologyDetails);

  requestedWorkerCpu += supervisorExists ? 0 : MesosCommon.executorCpu(mesosStormConf);
  requestedWorkerMem += supervisorExists ? 0 : MesosCommon.executorMem(mesosStormConf);

  if (requestedWorkerCpu <= offerResources.getCpu() && requestedWorkerMem <= offerResources.getMem()) {
    return true;
  }
  return false;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
public static MesosWorkerSlot createWorkerSlotFromOfferResources(Map mesosStormConf, OfferResources offerResources,
                                                          TopologyDetails topologyDetails, boolean supervisorExists) {
  double requestedWorkerCpu = MesosCommon.topologyWorkerCpu(mesosStormConf, topologyDetails);
  double requestedWorkerMem = MesosCommon.topologyWorkerMem(mesosStormConf, topologyDetails);

  requestedWorkerCpu += supervisorExists ? 0 : MesosCommon.executorCpu(mesosStormConf);
  requestedWorkerMem += supervisorExists ? 0 : MesosCommon.executorMem(mesosStormConf);

  if (requestedWorkerCpu > offerResources.getCpu()) {
    log.warn("Refusing to create worker slot. requestedWorkerCpu: {} but OfferedCpu: {} at node: {}",
             requestedWorkerCpu, offerResources.getCpu(), offerResources.getHostName());
    return null;
  }

  if (requestedWorkerMem > offerResources.getMem()) {
    log.warn("Refusing to create worker slot. requestedWorkerMem: {} but OfferedMem: {} at node: {}",
             requestedWorkerMem, offerResources.getMem(), offerResources.getHostName());
    return null;
  }

  long port = offerResources.getPort();

  if (port == -1) {
    log.warn("Refusing to create worker slot. There are no ports available with offer {}", offerResources.toString());
    return null;
  }

  offerResources.decCpu(requestedWorkerCpu);
  offerResources.decMem(requestedWorkerMem);

  return new MesosWorkerSlot(offerResources.getHostName(), port, topologyDetails.getId());
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
private TopologyDetails constructTopologyDetails(String topologyName, int numWorkers) {
  Map<String, TopologyDetails> topologyConf1 = new HashMap<>();

  StormTopology stormTopology = new StormTopology();
  TopologyDetails topologyDetails= new TopologyDetails(topologyName, topologyConf1, stormTopology, numWorkers);

  return topologyDetails;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
public static TopologyDetails constructTopologyDetails(String topologyName, int numWorkers, double numCpus, double memSize) {
  Map<String, TopologyDetails> topologyConf = new HashMap<>();

  StormTopology stormTopology = new StormTopology();
  TopologyDetails topologyDetails= new TopologyDetails(topologyName, topologyConf, stormTopology, numWorkers);
  topologyDetails.getConf().put(MesosCommon.WORKER_CPU_CONF, Double.valueOf(numCpus));
  topologyDetails.getConf().put(MesosCommon.WORKER_MEM_CONF, Double.valueOf(memSize));

  return topologyDetails;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public void addTopology(TopologyDetails td) {
    String topId = td.getId();
    LOG.debug("Adding in Topology {}", topId);
    _tds.put(topId, td);
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            _nodes.add(n);
        }
    }
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
@Override
public boolean canAdd(TopologyDetails td) {
    // Only add topologies that are not sharing nodes with other topologies
    String topId = td.getId();
    SchedulerAssignment assignment = _cluster.getAssignmentById(topId);
    if (assignment != null) {
        for (WorkerSlot ws : assignment.getSlots()) {
            Node n = _nodeIdToNode.get(ws.getNodeId());
            if (n.getRunningTopologies().size() > 1) {
                return false;
            }
        }
    }
    return true;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Get the nodes needed to schedule a non-isolated topology.
 * 
 * @param td the topology to be scheduled
 * @param allNodes the nodes already scheduled for this topology. This will be updated to include new nodes if needed.
 * @param lesserPools node pools we can steal nodes from
 * @return the number of additional slots that should be used for scheduling.
 */
private int getNodesForNotIsolatedTop(TopologyDetails td, Set<Node> allNodes, NodePool[] lesserPools) {
    String topId = td.getId();
    LOG.debug("Topology {} is not isolated", topId);
    int totalTasks = td.getExecutors().size();
    int origRequest = td.getNumWorkers();
    int slotsRequested = Math.min(totalTasks, origRequest);
    int slotsUsed = Node.countSlotsUsed(topId, allNodes);
    int slotsFree = Node.countFreeSlotsAlive(allNodes);
    // Check to see if we have enough slots before trying to get them
    int slotsAvailable = 0;
    if (slotsRequested > slotsFree) {
        slotsAvailable = NodePool.slotsAvailable(lesserPools);
    }
    int slotsToUse = Math.min(slotsRequested - slotsUsed, slotsFree + slotsAvailable);
    LOG.debug("Slots... requested {} used {} free {} available {} to be used {}", new Object[] { slotsRequested, slotsUsed, slotsFree, slotsAvailable,
            slotsToUse });
    if (slotsToUse <= 0) {
        _cluster.setStatus(topId, "Not Enough Slots Available to Schedule Topology");
        return 0;
    }
    int slotsNeeded = slotsToUse - slotsFree;
    int numNewNodes = NodePool.getNodeCountIfSlotsWereTaken(slotsNeeded, lesserPools);
    LOG.debug("Nodes... new {} used {} max {}", new Object[] { numNewNodes, _usedNodes, _maxNodes });
    if ((numNewNodes + _usedNodes) > _maxNodes) {
        _cluster.setStatus(topId, "Max Nodes(" + _maxNodes + ") for this user would be exceeded. " + (numNewNodes - (_maxNodes - _usedNodes))
                + " more nodes needed to run topology.");
        return 0;
    }

    Collection<Node> found = NodePool.takeNodesBySlot(slotsNeeded, lesserPools);
    _usedNodes += found.size();
    allNodes.addAll(found);
    return slotsToUse;
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Synchronize the watching topologies with alive topologies in the cluster. Dead topology will
 * be removed from the watching list.
 *
 * @param topologies alive topologies
 */
public void synTopologies(Topologies topologies) {
    Set<String> aliveTopoIds = topologies.getTopologies().stream().map(TopologyDetails::getId)
            .collect(Collectors.toSet());
    // remove topologies that dead
    watchingTopologies.keySet().retainAll(aliveTopoIds);
    topologies.getTopologies().stream().forEach(this::addOrUpdateTopology);
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
/**
 * Add a topology to the watching list or update corresponding TopologyDetails object
 * if this topology is under watching.
 *
 * @param topoDetails
 */
public void addOrUpdateTopology(TopologyDetails topoDetails) {
    String topoId = topoDetails.getId();
    // For a new joined topology, set a new watcher on it and add it to watching list
    // For a watching topology, update its running detail
    watchingTopologies.compute(topoId, (topologyId, context) ->
            (context == null ? watchTopology(topoDetails) : context.setTopologyDetails(topoDetails)));
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
private AllocationContext watchTopology(TopologyDetails topoDetails) {
    // get current assignment and set a watcher on the zk node
    Map<String, Integer> compExecutors = getCompExecutorsAndWatch(topoDetails.getId());
    if (compExecutors == null) {
        return null;
    }
    LOG.info("Begin to watching topology " + topoDetails.getId());
    return new AllocationContext(topoDetails, compExecutors);
}
 

import backtype.storm.scheduler.TopologyDetails; //导入依赖的package包/类
public static double topologyCpu(Map conf, TopologyDetails info) {
    conf = getFullTopologyConfig(conf, info);
    Object cpuObj = conf.get(CPU_CONF);
    if(cpuObj!=null && !(cpuObj instanceof Number)) {
        LOG.warn("Topology has invalid mesos cpu configuration: " + cpuObj + " for topology " + info.getId());
        cpuObj = null;
    }
    if(cpuObj==null) return DEFAULT_CPU;
    else return ((Number)cpuObj).doubleValue();
}