You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/08/22 17:21:27 UTC
[1/2] storm git commit: STORM-2673: add support for prioritizing
nodes in scheduling
Repository: storm
Updated Branches:
refs/heads/master b87dcc0f7 -> ee109142d
STORM-2673: add support for prioritizing nodes in scheduling
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9b4c30da
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9b4c30da
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9b4c30da
Branch: refs/heads/master
Commit: 9b4c30dab0bc2828c4675d94a1e834a092ae8635
Parents: 6ae02e4
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Wed Aug 2 14:21:11 2017 -0500
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Wed Aug 2 14:21:11 2017 -0500
----------------------------------------------------------------------
.../src/jvm/org/apache/storm/Config.java | 14 ++
.../org/apache/storm/scheduler/Cluster.java | 12 ++
.../DefaultResourceAwareStrategy.java | 97 ++++++------
.../TestDefaultResourceAwareStrategy.java | 149 +++++++++++++++++++
4 files changed, 226 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9b4c30da/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index bc13754..f9493d5 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -262,6 +262,20 @@ public class Config extends HashMap<String, Object> {
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
/**
+ * A list of host names that this topology would prefer to be scheduled on (no guarantee is given though).
+ * This is intended for debugging only.
+ */
+ @isStringList
+ public static final String TOPOLOGY_SCHEDULER_FAVORED_NODES = "topology.scheduler.favored.nodes";
+
+ /**
+ * A list of host names that this topology would prefer to NOT be scheduled on (no guarantee is given though).
+ * This is intended for debugging only.
+ */
+ @isStringList
+ public static final String TOPOLOGY_SCHEDULER_UNFAVORED_NODES = "topology.scheduler.unfavored.nodes";
+
+ /**
* How many executors to spawn for ackers.
*
* <p>By not setting this variable or setting it as null, Storm will set the number of acker executors
http://git-wip-us.apache.org/repos/asf/storm/blob/9b4c30da/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
index a6ae4cc..b5f105a 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
@@ -514,6 +514,18 @@ public class Cluster implements ISchedulingState {
return false;
}
+ //Lets see if we can make the Memory one fast too, at least in the failure case.
+ //The totalMemReq is not really that accurate because it does not include shared memory, but if it does not fit we know
+ // Even with shared it will not work
+ double minMemNeeded = td.getTotalMemReqTask(exec);
+ if (minMemNeeded > memoryAvailable) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Could not schedule {}:{} on {} not enough Mem {} > {}", td.getName(), exec, ws, minMemNeeded, memoryAvailable);
+ }
+ //Not enough minimum MEM no need to try any more
+ return false;
+ }
+
double currentTotal = 0.0;
double afterTotal = 0.0;
double afterOnHeap = 0.0;
http://git-wip-us.apache.org/repos/asf/storm/blob/9b4c30da/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
index 650f3df..9f644c9 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
@@ -23,7 +23,6 @@ import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
@@ -33,6 +32,7 @@ import java.util.Queue;
import java.util.Set;
import java.util.TreeSet;
+import org.apache.storm.Config;
import org.apache.storm.generated.ComponentType;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.Component;
@@ -53,9 +53,6 @@ public class DefaultResourceAwareStrategy implements IStrategy {
private Map<String, List<String>> networkTopography;
private RAS_Nodes nodes;
- private TreeSet<ObjectResources> sortedRacks = null;
- private Map<String, TreeSet<ObjectResources>> rackIdToSortedNodes = new HashMap<>();
-
@VisibleForTesting
void prepare(Cluster cluster) {
this.cluster = cluster;
@@ -93,6 +90,9 @@ public class DefaultResourceAwareStrategy implements IStrategy {
List<ExecutorDetails> orderedExecutors = orderExecutors(td, unassignedExecutors);
Collection<ExecutorDetails> executorsNotScheduled = new HashSet<>(unassignedExecutors);
+ List<String> favoredNodes = (List<String>) td.getConf().get(Config.TOPOLOGY_SCHEDULER_FAVORED_NODES);
+ List<String> unFavoredNodes = (List<String>) td.getConf().get(Config.TOPOLOGY_SCHEDULER_UNFAVORED_NODES);
+ final List<ObjectResources> sortedNodes = sortAllNodes(td, favoredNodes, unFavoredNodes);
for (ExecutorDetails exec : orderedExecutors) {
LOG.debug(
@@ -100,14 +100,14 @@ public class DefaultResourceAwareStrategy implements IStrategy {
exec,
td.getExecutorToComponent().get(exec),
td.getTaskResourceReqList(exec));
- scheduleExecutor(exec, td, scheduledTasks);
+ scheduleExecutor(exec, td, scheduledTasks, sortedNodes);
}
executorsNotScheduled.removeAll(scheduledTasks);
LOG.debug("/* Scheduling left over task (most likely sys tasks) */");
// schedule left over system tasks
for (ExecutorDetails exec : executorsNotScheduled) {
- scheduleExecutor(exec, td, scheduledTasks);
+ scheduleExecutor(exec, td, scheduledTasks, sortedNodes);
}
SchedulingResult result;
@@ -128,6 +128,48 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return result;
}
+ private List<ObjectResources> sortAllNodes(TopologyDetails td, List<String> favoredNodes, List<String> unFavoredNodes) {
+ TreeSet<ObjectResources> sortedRacks = sortRacks(td.getId());
+ ArrayList<ObjectResources> totallySortedNodes = new ArrayList<>();
+ for (ObjectResources rack : sortedRacks) {
+ final String rackId = rack.id;
+ TreeSet<ObjectResources> sortedNodes =
+ sortNodes(getAvailableNodesFromRack(rackId), rackId, td.getId());
+ totallySortedNodes.addAll(sortedNodes);
+ }
+ //Now do some post processing to add make some nodes preferred over others.
+ if (favoredNodes != null || unFavoredNodes != null) {
+ HashMap<String, Integer> hostOrder = new HashMap<>();
+ if (favoredNodes != null) {
+ int size = favoredNodes.size();
+ for (int i = 0; i < size; i++) {
+ //First in the list is the most desired so gets the Lowest possible value
+ hostOrder.put(favoredNodes.get(i), -(size - i));
+ }
+ }
+ if (unFavoredNodes != null) {
+ int size = unFavoredNodes.size();
+ for (int i = 0; i < size; i++) {
+ //First in the list is the least desired so gets the highest value
+ hostOrder.put(unFavoredNodes.get(i), size - i);
+ }
+ }
+ //java guarantees a stable sort so we can just return 0 for values we don't want to move.
+ Collections.sort(totallySortedNodes, (o1, o2) -> {
+ RAS_Node n1 = nodes.getNodeById(o1.id);
+ String host1 = n1.getHostname();
+ int h1Value = hostOrder.getOrDefault(host1, 0);
+
+ RAS_Node n2 = nodes.getNodeById(o2.id);
+ String host2 = n2.getHostname();
+ int h2Value = hostOrder.getOrDefault(host2, 0);
+
+ return Integer.compare(h1Value, h2Value);
+ });
+ }
+ return totallySortedNodes;
+ }
+
/**
* Schedule executor exec from topology td.
*
@@ -136,8 +178,8 @@ public class DefaultResourceAwareStrategy implements IStrategy {
* @param scheduledTasks executors that have been scheduled
*/
private void scheduleExecutor(
- ExecutorDetails exec, TopologyDetails td, Collection<ExecutorDetails> scheduledTasks) {
- WorkerSlot targetSlot = findWorkerForExec(exec, td);
+ ExecutorDetails exec, TopologyDetails td, Collection<ExecutorDetails> scheduledTasks, List<ObjectResources> sortedNodes) {
+ WorkerSlot targetSlot = findWorkerForExec(exec, td, sortedNodes);
if (targetSlot != null) {
RAS_Node targetNode = idToNode(targetSlot.getNodeId());
targetNode.assignSingleExecutor(targetSlot, exec, td);
@@ -165,44 +207,7 @@ public class DefaultResourceAwareStrategy implements IStrategy {
* @param td the topology that the executor is a part of
* @return a worker to assign exec on. Returns null if a worker cannot be successfully found in cluster
*/
- private WorkerSlot findWorkerForExec(ExecutorDetails exec, TopologyDetails td) {
- WorkerSlot ws = null;
-
- // iterate through an ordered list of all racks available to make sure we cannot schedule
- // the first executor in any rack before we "give up" the list is ordered in decreasing order
- // of effective resources. With the rack in the front of the list having the most effective
- // resources.
- if (sortedRacks == null) {
- sortedRacks = sortRacks(td.getId());
- }
-
- for (ObjectResources rack : sortedRacks) {
- ws = this.getBestWorker(exec, td, rack.id);
- if (ws != null) {
- LOG.debug("best rack: {}", rack.id);
- break;
- }
- }
- return ws;
- }
-
- /**
- * Get the best worker to assign executor exec on a rack
- *
- * @param exec the executor to schedule
- * @param td the topology that the executor is a part of
- * @param rackId the rack id of the rack to find a worker on
- * @return a worker to assign executor exec to. Returns null if a worker cannot be successfully found on rack with
- * rackId
- */
- private WorkerSlot getBestWorker(ExecutorDetails exec, TopologyDetails td, String rackId) {
- if (!rackIdToSortedNodes.containsKey(rackId)) {
- rackIdToSortedNodes.put(
- rackId, sortNodes(this.getAvailableNodesFromRack(rackId), rackId, td.getId()));
- }
-
- TreeSet<ObjectResources> sortedNodes = rackIdToSortedNodes.get(rackId);
-
+ private WorkerSlot findWorkerForExec(ExecutorDetails exec, TopologyDetails td, List<ObjectResources> sortedNodes) {
for (ObjectResources nodeResources : sortedNodes) {
RAS_Node node = nodes.getNodeById(nodeResources.id);
for (WorkerSlot ws : node.getSlotsAvailbleTo(td)) {
http://git-wip-us.apache.org/repos/asf/storm/blob/9b4c30da/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
index fca26b9..16c0b8b 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
@@ -35,11 +35,13 @@ import org.apache.storm.scheduler.WorkerSlot;
import org.apache.storm.scheduler.resource.RAS_Node;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
import org.apache.storm.scheduler.resource.SchedulingResult;
+import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.ObjectResources;
import org.apache.storm.topology.SharedOffHeapWithinNode;
import org.apache.storm.topology.SharedOffHeapWithinWorker;
import org.apache.storm.topology.SharedOnHeap;
import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.utils.Utils;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@@ -324,4 +326,151 @@ public class TestDefaultResourceAwareStrategy {
}
Assert.assertEquals("All executors in topo-2 scheduled", 0, cluster.getUnassignedExecutors(topo1).size());
}
+
+ /**
+ * Test whether strategy will choose correct rack
+ */
+ @Test
+ public void testMultipleRacksWithFavoritism() {
+ final Map<String, SupervisorDetails> supMap = new HashMap<String, SupervisorDetails>();
+ final Map<String, SupervisorDetails> supMapRack1 = genSupervisors(10, 4, 0, 400, 8000);
+ //generate another rack of supervisors with less resources
+ final Map<String, SupervisorDetails> supMapRack2 = genSupervisors(10, 4, 10, 200, 4000);
+
+ //generate some supervisors that are depleted of one resource
+ final Map<String, SupervisorDetails> supMapRack3 = genSupervisors(10, 4, 20, 0, 8000);
+
+ //generate some that has alot of memory but little of cpu
+ final Map<String, SupervisorDetails> supMapRack4 = genSupervisors(10, 4, 30, 10, 8000 * 2 + 4000);
+
+ //generate some that has alot of cpu but little of memory
+ final Map<String, SupervisorDetails> supMapRack5 = genSupervisors(10, 4, 40, 400 + 200 + 10, 1000);
+
+ supMap.putAll(supMapRack1);
+ supMap.putAll(supMapRack2);
+ supMap.putAll(supMapRack3);
+ supMap.putAll(supMapRack4);
+ supMap.putAll(supMapRack5);
+
+ Config config = createClusterConfig(100, 500, 500, null);
+ config.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, Double.MAX_VALUE);
+ INimbus iNimbus = new INimbusTest();
+
+ //create test DNSToSwitchMapping plugin
+ DNSToSwitchMapping TestNetworkTopographyPlugin = new DNSToSwitchMapping() {
+ @Override
+ public Map<String, String> resolve(List<String> names) {
+ Map<String, String> ret = new HashMap<>();
+ for (SupervisorDetails sup : supMapRack1.values()) {
+ ret.put(sup.getHost(), "rack-0");
+ }
+ for (SupervisorDetails sup : supMapRack2.values()) {
+ ret.put(sup.getHost(), "rack-1");
+ }
+ for (SupervisorDetails sup : supMapRack3.values()) {
+ ret.put(sup.getHost(), "rack-2");
+ }
+ for (SupervisorDetails sup : supMapRack4.values()) {
+ ret.put(sup.getHost(), "rack-3");
+ }
+ for (SupervisorDetails sup : supMapRack5.values()) {
+ ret.put(sup.getHost(), "rack-4");
+ }
+ return ret;
+ }
+ };
+
+ Config t1Conf = new Config();
+ t1Conf.putAll(config);
+ final List<String> t1FavoredHostNames = Arrays.asList("host-41", "host-42", "host-43");
+ t1Conf.put(Config.TOPOLOGY_SCHEDULER_FAVORED_NODES, t1FavoredHostNames);
+ final List<String> t1UnfavoredHostIds = Arrays.asList("host-1", "host-2", "host-3");
+ t1Conf.put(Config.TOPOLOGY_SCHEDULER_UNFAVORED_NODES, t1UnfavoredHostIds);
+ //generate topologies
+ TopologyDetails topo1 = genTopology("topo-1", t1Conf, 8, 0, 2, 0, currentTime - 2, 10, "user");
+
+
+ Config t2Conf = new Config();
+ t2Conf.putAll(config);
+ t2Conf.put(Config.TOPOLOGY_SCHEDULER_FAVORED_NODES, Arrays.asList("host-31", "host-32", "host-33"));
+ t2Conf.put(Config.TOPOLOGY_SCHEDULER_UNFAVORED_NODES, Arrays.asList("host-11", "host-12", "host-13"));
+ TopologyDetails topo2 = genTopology("topo-2", t2Conf, 8, 0, 2, 0, currentTime - 2, 10, "user");
+
+ Topologies topologies = new Topologies(topo1, topo2);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
+
+ List<String> supHostnames = new LinkedList<>();
+ for (SupervisorDetails sup : supMap.values()) {
+ supHostnames.add(sup.getHost());
+ }
+ Map<String, List<String>> rackToNodes = new HashMap<>();
+ Map<String, String> resolvedSuperVisors = TestNetworkTopographyPlugin.resolve(supHostnames);
+ for (Map.Entry<String, String> entry : resolvedSuperVisors.entrySet()) {
+ String hostName = entry.getKey();
+ String rack = entry.getValue();
+ List<String> nodesForRack = rackToNodes.get(rack);
+ if (nodesForRack == null) {
+ nodesForRack = new ArrayList<String>();
+ rackToNodes.put(rack, nodesForRack);
+ }
+ nodesForRack.add(hostName);
+ }
+ cluster.setNetworkTopography(rackToNodes);
+
+ DefaultResourceAwareStrategy rs = new DefaultResourceAwareStrategy();
+
+ rs.prepare(cluster);
+ TreeSet<ObjectResources> sortedRacks= rs.sortRacks(topo1.getId());
+
+ Assert.assertEquals("# of racks sorted", 5, sortedRacks.size());
+ Iterator<ObjectResources> it = sortedRacks.iterator();
+ // Ranked first since rack-0 has the most balanced set of resources
+ Assert.assertEquals("rack-0 should be ordered first", "rack-0", it.next().id);
+ // Ranked second since rack-1 has a balanced set of resources but less than rack-0
+ Assert.assertEquals("rack-1 should be ordered second", "rack-1", it.next().id);
+ // Ranked third since rack-4 has a lot of cpu but not a lot of memory
+ Assert.assertEquals("rack-4 should be ordered third", "rack-4", it.next().id);
+ // Ranked fourth since rack-3 has alot of memory but not cpu
+ Assert.assertEquals("rack-3 should be ordered fourth", "rack-3", it.next().id);
+ //Ranked last since rack-2 has not cpu resources
+ Assert.assertEquals("rack-2 should be ordered fifth", "rack-2", it.next().id);
+
+ SchedulingResult schedulingResult = rs.schedule(cluster, topo1);
+ assert(schedulingResult.isSuccess());
+ SchedulerAssignment assignment = cluster.getAssignmentById(topo1.getId());
+ for (WorkerSlot ws : assignment.getSlotToExecutors().keySet()) {
+ String hostName = rs.idToNode(ws.getNodeId()).getHostname();
+ String rackId = resolvedSuperVisors.get(hostName);
+ Assert.assertTrue(ws + " is neither on a favored node " + t1FavoredHostNames + " nor the highest priority rack (rack-0)",
+ t1FavoredHostNames.contains(hostName) || "rack-0".equals(rackId));
+ Assert.assertFalse(ws + " is a part of an unfavored node " + t1UnfavoredHostIds,
+ t1UnfavoredHostIds.contains(hostName));
+ }
+ Assert.assertEquals("All executors in topo-1 scheduled", 0, cluster.getUnassignedExecutors(topo1).size());
+
+ //Test if topology is already partially scheduled on one rack
+ Iterator<ExecutorDetails> executorIterator = topo2.getExecutors().iterator();
+ List<String> nodeHostnames = rackToNodes.get("rack-1");
+ for (int i = 0; i< topo2.getExecutors().size()/2; i++) {
+ String nodeHostname = nodeHostnames.get(i % nodeHostnames.size());
+ RAS_Node node = rs.idToNode(rs.nodeHostnameToId(nodeHostname));
+ WorkerSlot targetSlot = node.getFreeSlots().iterator().next();
+ ExecutorDetails targetExec = executorIterator.next();
+ // to keep track of free slots
+ node.assign(targetSlot, topo2, Arrays.asList(targetExec));
+ }
+
+ rs = new DefaultResourceAwareStrategy();
+ // schedule topo2
+ schedulingResult = rs.schedule(cluster, topo2);
+ assert(schedulingResult.isSuccess());
+ assignment = cluster.getAssignmentById(topo2.getId());
+ for (WorkerSlot ws : assignment.getSlotToExecutors().keySet()) {
+ //make sure all workers on scheduled in rack-1
+ // The favored nodes would have put it on a different rack, but because that rack does not have free space to run the
+ // topology it falls back to this rack
+ Assert.assertEquals("assert worker scheduled on rack-1", "rack-1", resolvedSuperVisors.get(rs.idToNode(ws.getNodeId()).getHostname()));
+ }
+ Assert.assertEquals("All executors in topo-2 scheduled", 0, cluster.getUnassignedExecutors(topo1).size());
+ }
}
[2/2] storm git commit: Merge branch 'STORM-2673' of
https://github.com/revans2/incubator-storm into STORM-2673
Posted by bo...@apache.org.
Merge branch 'STORM-2673' of https://github.com/revans2/incubator-storm into STORM-2673
STORM-2673: add support for prioritizing nodes in scheduling
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ee109142
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ee109142
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ee109142
Branch: refs/heads/master
Commit: ee109142d6b4faa18b68edc03a6f2171a7628694
Parents: b87dcc0 9b4c30d
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Tue Aug 22 12:05:25 2017 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Tue Aug 22 12:05:25 2017 -0500
----------------------------------------------------------------------
.../src/jvm/org/apache/storm/Config.java | 14 ++
.../org/apache/storm/scheduler/Cluster.java | 12 ++
.../DefaultResourceAwareStrategy.java | 97 ++++++------
.../TestDefaultResourceAwareStrategy.java | 149 +++++++++++++++++++
4 files changed, 226 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ee109142/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------