MesosNimbus implements INimbus { protected void computeResourcesForSlot(final RotatingMap<OfferID, Offer> offers, Topologies topologies, Map<OfferID, List<LaunchTask>> toLaunch, String topologyId, Map<OfferID, List<WorkerSlot>> slotList, OfferID id) { Offer offer = offers.get(id); List<WorkerSlot> workerSlots = slotList.get(id); boolean usingExistingOffer = false; boolean subtractedExecutorResources = false; for (WorkerSlot slot : workerSlots) { TopologyDetails details = topologies.getById(topologyId); String workerPrefix = ""; if (_conf.containsKey(MesosCommon.WORKER_NAME_PREFIX)) { workerPrefix = MesosCommon.getWorkerPrefix(_conf, details); } TaskID taskId = TaskID.newBuilder() .setValue(MesosCommon.taskId(workerPrefix + slot.getNodeId(), slot.getPort())) .build(); if ((id == null || offer == null) && _usedOffers.containsKey(taskId)) { offer = _usedOffers.get(taskId); if (offer != null) { id = offer.getId(); usingExistingOffer = true; } } if (id != null && offer != null) { if (!toLaunch.containsKey(id)) { toLaunch.put(id, new ArrayList<LaunchTask>()); } double workerCpu = MesosCommon.topologyWorkerCpu(_conf, details); double workerMem = MesosCommon.topologyWorkerMem(_conf, details); double executorCpu = MesosCommon.executorCpu(_conf); double executorMem = MesosCommon.executorMem(_conf); Map executorData = new HashMap(); executorData.put(MesosCommon.SUPERVISOR_ID, MesosCommon.supervisorId(slot.getNodeId(), details.getId())); executorData.put(MesosCommon.ASSIGNMENT_ID, workerPrefix + slot.getNodeId()); Offer.Builder newBuilder = Offer.newBuilder(); newBuilder.mergeFrom(offer); newBuilder.clearResources(); Offer.Builder existingBuilder = Offer.newBuilder(); existingBuilder.mergeFrom(offer); existingBuilder.clearResources(); String extraConfig = ""; List<Resource> offerResources = new ArrayList<>(); offerResources.addAll(offer.getResourcesList()); if (_preferReservedResources) { Collections.sort(offerResources, new ResourceRoleComparator()); } List<Resource> executorCpuResources = getResourcesScalar(offerResources, executorCpu, "cpus"); List<Resource> executorMemResources = getResourcesScalar(offerResources, executorMem, "mem"); List<Resource> executorPortsResources = null; if (!subtractedExecutorResources) { offerResources = subtractResourcesScalar(offerResources, executorCpu, "cpus"); offerResources = subtractResourcesScalar(offerResources, executorMem, "mem"); subtractedExecutorResources = true; } List<Resource> workerCpuResources = getResourcesScalar(offerResources, workerCpu, "cpus"); offerResources = subtractResourcesScalar(offerResources, workerCpu, "cpus"); List<Resource> workerMemResources = getResourcesScalar(offerResources, workerMem, "mem"); offerResources = subtractResourcesScalar(offerResources, workerMem, "mem"); List<Resource> workerPortsResources = getResourcesRange(offerResources, slot.getPort(), "ports"); offerResources = subtractResourcesRange(offerResources, slot.getPort(), "ports"); if (!subtractedExecutorResources && MesosCommon.startLogViewer(_conf)) { List<Integer> portList = new ArrayList<>(); collectPorts(offerResources, portList, 1); int port = Optional.fromNullable((Number) _conf.get(Config.LOGVIEWER_PORT)).or(8000).intValue(); executorPortsResources = getResourcesRange(offerResources, port, "ports"); if (!executorPortsResources.isEmpty()) { extraConfig = " -c " + MesosCommon.AUTO_START_LOGVIEWER_CONF + "=true"; offerResources = subtractResourcesRange(offerResources, port, "ports"); } } Offer remainingOffer = existingBuilder.addAllResources(offerResources).build(); offers.put(id, remainingOffer); String configUri; try { configUri = new URL(_configUrl.toURL(), _configUrl.getPath() + "/storm.yaml").toString(); } catch (MalformedURLException e) { throw new RuntimeException(e); } String delimiter = MesosCommon.getMesosComponentNameDelimiter(_conf, details); String topologyAndNodeId = details.getId() + delimiter + slot.getNodeId(); String executorName = "storm-supervisor" + delimiter + topologyAndNodeId; String taskName = "storm-worker" + delimiter + topologyAndNodeId + ":" + slot.getPort(); String executorDataStr = JSONValue.toJSONString(executorData); ExecutorInfo.Builder executorInfoBuilder = ExecutorInfo.newBuilder(); executorInfoBuilder .setName(executorName) .setExecutorId(ExecutorID.newBuilder().setValue(details.getId())) .setData(ByteString.copyFromUtf8(executorDataStr)) .addAllResources(executorCpuResources) .addAllResources(executorMemResources); if (executorPortsResources != null) { executorInfoBuilder.addAllResources(executorPortsResources); } ICommandLineShim commandLineShim = CommandLineShimFactory.makeCommandLineShim(_container.isPresent(), extraConfig); if (_container.isPresent()) { executorInfoBuilder .setCommand(CommandInfo.newBuilder() .addUris(URI.newBuilder().setValue(configUri)) .setValue(commandLineShim.getCommandLine(details.getId()))) .setContainer( ContainerInfo.newBuilder() .setType(ContainerInfo.Type.DOCKER) .setDocker( ContainerInfo.DockerInfo.newBuilder() .setImage(_container.get()) .setNetwork(ContainerInfo.DockerInfo.Network.HOST) .setForcePullImage(true) .build() ) .build() ); } else { executorInfoBuilder .setCommand(CommandInfo.newBuilder() .addUris(URI.newBuilder().setValue((String) _conf.get(CONF_EXECUTOR_URI))) .addUris(URI.newBuilder().setValue(configUri)) .setValue(commandLineShim.getCommandLine(details.getId()))); } LOG.info("Launching task with Mesos Executor data: < {} >", executorDataStr); TaskInfo task = TaskInfo.newBuilder() .setName(taskName) .setTaskId(taskId) .setSlaveId(offer.getSlaveId()) .setExecutor(executorInfoBuilder.build()) .addAllResources(workerCpuResources) .addAllResources(workerMemResources) .addAllResources(workerPortsResources) .build(); Offer newOffer = offer.toBuilder() .addAllResources(task.getResourcesList()).build(); LOG.debug("Launching task: {}", task.toString()); toLaunch.get(id).add(new LaunchTask(task, newOffer)); } if (usingExistingOffer) { _driver.killTask(taskId); } } } MesosNimbus(); static void main(String[] args); @Override IScheduler getForcedScheduler(); @Override String getHostName(Map<String, SupervisorDetails> map, String nodeId); @Override void prepare(Map conf, String localDir); void doRegistration(final SchedulerDriver driver, Protos.FrameworkID id); void shutdown(); void resourceOffers(SchedulerDriver driver, List<Protos.Offer> offers); void offerRescinded(OfferID id); void taskLost(final TaskID taskId); boolean isHostAccepted(String hostname); @Override Collection<WorkerSlot> allSlotsAvailableForScheduling(
          Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments); @Override void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> slots); static final String CONF_EXECUTOR_URI; static final String CONF_MASTER_URL; static final String CONF_MASTER_FAILOVER_TIMEOUT_SECS; static final String CONF_MESOS_ALLOWED_HOSTS; static final String CONF_MESOS_DISALLOWED_HOSTS; static final String CONF_MESOS_ROLE; static final String CONF_MESOS_PRINCIPAL; static final String CONF_MESOS_SECRET_FILE; static final String CONF_MESOS_CHECKPOINT; static final String CONF_MESOS_OFFER_LRU_CACHE_SIZE; static final String CONF_MESOS_OFFER_FILTER_SECONDS; static final String CONF_MESOS_OFFER_EXPIRY_MULTIPLIER; static final String CONF_MESOS_LOCAL_FILE_SERVER_PORT; static final String CONF_MESOS_FRAMEWORK_NAME; static final String CONF_MESOS_PREFER_RESERVED_RESOURCES; static final String CONF_MESOS_CONTAINER_DOCKER_IMAGE; static final String FRAMEWORK_ID; }
@Test public void testComputeResourcesForSlot() throws Exception { MesosNimbus mesosNimbus = new MesosNimbus(); mesosNimbus._configUrl = new URI("http: 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()); }