import backtype.storm.scheduler.Topologies; //导入依赖的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.Topologies; //导入依赖的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.Topologies; //导入依赖的package包/类
@Before
public void initialize() {
  defaultScheduler = new DefaultScheduler();
  Map<String, Object> mesosStormConf = new HashMap<>();
  defaultScheduler.prepare(mesosStormConf);

  rotatingMap = new RotatingMap<>(2);

  topologiesMissingAssignments = new HashSet<>();
  topologiesMissingAssignments.add("test-topology1-65-1442255385");
  topologiesMissingAssignments.add("test-topology1-65-1442255385");

  existingSupervisors = new ArrayList<>();
  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(sampleHost, "test-topology1-65-1442255385"), sampleHost, null, null));
  existingSupervisors.add(new SupervisorDetails(MesosCommon.supervisorId(sampleHost, "test-topology10-65-1442255385"), sampleHost, null, null));

  topologyMap = new HashMap<>();
  topologyMap.put(sampleTopologyId, TestUtils.constructTopologyDetails(sampleTopologyId, 1, 0.1, 100));
  topologies = new Topologies(topologyMap);

  mesosWorkerSlotMap = new HashMap<>();
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的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.Topologies; //导入依赖的package包/类
public void testSynTopologies() throws Exception {
    Topologies topologies = TopologyHelper.getTopologyDetails(conf);
    System.out.println("total topology count: " + topologies.getTopologies().size());
    topologyListener.synTopologies(topologies);
    Utils.sleep(10000);

    String topoId = TopologyHelper.getTopologyId("wc", conf);
    Map<String, Integer> newAssignment = Collections.singletonMap("split", 5);
    byte[] data = Utils.serialize(new HashMap<>(newAssignment));
    if (zk.checkExists().forPath("/resa/" + topoId) == null) {
        zk.create().creatingParentsIfNeeded().forPath("/resa/" + topoId, data);
        System.out.println("Set assignment");
    }
    Utils.sleep(1000);
    topologyListener.synTopologies(topologies);
    System.out.println("Reset assignment");
    zk.setData().forPath("/resa/" + topoId, data);
    Utils.sleep(31000);

    TopologyHelper.killTopology("wc", 0, conf);
    Utils.sleep(10000);
    topologies = TopologyHelper.getTopologyDetails(conf);
    System.out.println("total topology count: " + topologies.getTopologies().size());
    topologyListener.synTopologies(topologies);
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的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.Topologies; //导入依赖的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.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
        Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments) {
  synchronized (_offersLock) {
    return _mesosStormScheduler.allSlotsAvailableForScheduling(
            _offers,
            existingSupervisors,
            topologies,
            topologiesMissingAssignments);
  }
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
		Collection<SupervisorDetails> existingSupervisors,
		Topologies topologies, Set<String> topologiesMissingAssignments) {
	Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
	for (SupervisorDetails detail : existingSupervisors) {
		for (Integer port : detail.getAllPorts())
			result.add(new WorkerSlot(detail.getId(), port));
	}
	return result;
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
		Collection<SupervisorDetails> existingSupervisors,
		Topologies topologies, Set<String> topologiesMissingAssignments) {
	// TODO Auto-generated method stub
	Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
	for (SupervisorDetails detail : existingSupervisors) {
		for (Integer port : detail.getAllPorts())
			result.add(new WorkerSlot(detail.getId(), port));
	}
	return result;
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(Collection<SupervisorDetails> existingSupervisors, Topologies topologies,
        Set<String> topologiesMissingAssignments) {
    // TODO Auto-generated method stub
    Collection<WorkerSlot> result = new HashSet<WorkerSlot>();
    for (SupervisorDetails detail : existingSupervisors) {
        for (Integer port : detail.getAllPorts())
            result.add(new WorkerSlot(detail.getId(), port));
    }
    return result;
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的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.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(Collection<SupervisorDetails> existingSupervisors,
                                                             Topologies topologies,
                                                             Set<String> topologiesMissingAssignments) {
    Collection<WorkerSlot> result = new HashSet<>();
    for (SupervisorDetails detail : existingSupervisors) {
        for (Integer port : detail.getAllPorts())
            result.add(new WorkerSlot(detail.getId(), port));
    }
    return result;
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Test
public void testComputeResourcesForSlot() throws Exception {
  MesosNimbus mesosNimbus = new MesosNimbus();

  mesosNimbus._configUrl = new URI("http://127.0.0.1/");

  OfferID offerId = OfferID.newBuilder().setValue("derp").build();
  RotatingMap<OfferID, Offer> offers = new RotatingMap<>(
      new RotatingMap.ExpiredCallback<OfferID, Offer>() {
        @Override
        public void expire(OfferID key, Offer val) {
        }
      }
  );

  offers.put(
      offerId,
      TestUtils.buildOfferWithPorts("offer1", "host1.west", 2.0, 2048, 1000, 1000)
  );

  HashMap<String, TopologyDetails> topologyMap = new HashMap<>();
  Map conf = new HashMap<>();
  conf.put(MesosCommon.WORKER_CPU_CONF, 1);
  conf.put(MesosCommon.WORKER_MEM_CONF, 1024);
  conf.put(MesosCommon.EXECUTOR_CPU_CONF, 1);
  conf.put(MesosCommon.EXECUTOR_MEM_CONF, 1024);
  conf.put(MesosNimbus.CONF_EXECUTOR_URI, "");
  mesosNimbus._conf = conf;

  topologyMap.put("t1", new TopologyDetails("t1", conf, new StormTopology(), 5));
  HashMap<OfferID, List<LaunchTask>> launchList = new HashMap<>();
  HashMap<OfferID, List<WorkerSlot>> slotList = new HashMap<>();
  slotList.put(offerId, Arrays.asList(new WorkerSlot("", 1000)));
  Topologies topologies = new Topologies(topologyMap);

  mesosNimbus.computeResourcesForSlot(
      offers,
      topologies,
      launchList,
      "t1",
      slotList,
      OfferID.newBuilder().setValue("derp").build()
  );
  assertEquals(1, launchList.size());
  assertEquals(1, launchList.get(offerId).size());

  assertEquals(
      TestUtils.buildScalarResource("cpus", 1.0),
      launchList.get(offerId).get(0).getTask().getResources(0)
  );

  assertEquals(0, offers.get(offerId).getResourcesCount());
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies,
		Map<String, Collection<WorkerSlot>> newSlotsByTopologyId) {

}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies,
		Map<String, Collection<WorkerSlot>> newSlotsByTopologyId) {
	// TODO Auto-generated method stub

}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public void schedule(Topologies topologies, Cluster cluster) {
	evenScheduler.schedule(topologies, cluster);
	assignmentTracker.checkAssignment(topologies, cluster);
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> newSlotsByTopologyId) {
    // TODO Auto-generated method stub

}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public Collection<WorkerSlot> allSlotsAvailableForScheduling(
        Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments) {
    synchronized (OFFERS_LOCK) {
        LOG.info("Currently have " + _offers.size() + " offers buffered");
        if (!topologiesMissingAssignments.isEmpty()) {
            LOG.info("Topologies that need assignments: " + topologiesMissingAssignments.toString());
        } else {
            LOG.info("Declining offers because no topologies need assignments");
            _offers.clear();
        }
    }

    Double cpu = null;
    Double mem = null;
    // TODO: maybe this isn't the best approach. if a topology raises #cpus keeps failing,
    // it will mess up scheduling on this cluster permanently
    for (String id : topologiesMissingAssignments) {
        TopologyDetails details = topologies.getById(id);
        double tcpu = MesosCommon.topologyCpu(_conf, details);
        double tmem = MesosCommon.topologyMem(_conf, details);
        if (cpu == null || tcpu > cpu) {
            cpu = tcpu;
        }
        if (mem == null || tmem > mem) {
            mem = tmem;
        }
    }

    // need access to how many slots are currently used to limit number of slots taken up
    List<WorkerSlot> allSlots = new ArrayList();

    if (cpu != null && mem != null) {
        synchronized (OFFERS_LOCK) {
            for (Offer offer : _offers.values()) {
                allSlots.addAll(toSlots(offer, cpu, mem));
            }
        }
    }

    LOG.info("Number of available slots: " + allSlots.size());
    return allSlots;
}
 

import backtype.storm.scheduler.Topologies; //导入依赖的package包/类
@Override
public void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> newSlotsByTopologyId) {
}