You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by je...@apache.org on 2016/10/13 20:36:09 UTC
[1/6] storm git commit: [STORM-2134] - improving the current
scheduling strategy for RAS
Repository: storm
Updated Branches:
refs/heads/1.x-branch e7fab41a1 -> 20b612c25
[STORM-2134] - improving the current scheduling strategy for RAS
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0d755eeb
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0d755eeb
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0d755eeb
Branch: refs/heads/1.x-branch
Commit: 0d755eeb71c33a5ced9d603f11d74f1546e5a544
Parents: e7fab41
Author: Boyang Jerry Peng <je...@yahoo-inc.com>
Authored: Thu Oct 6 13:45:55 2016 -0500
Committer: Boyang Jerry Peng <je...@yahoo-inc.com>
Committed: Thu Oct 13 15:25:39 2016 -0500
----------------------------------------------------------------------
pom.xml | 6 +
storm-core/pom.xml | 16 +
.../storm/scheduler/resource/ResourceUtils.java | 22 +
.../DefaultResourceAwareStrategy.java | 673 ++++++++++++-------
.../TestDefaultResourceAwareStrategy.java | 8 +-
5 files changed, 472 insertions(+), 253 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0d755eeb/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b60a1fa..562b8b5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -206,6 +206,7 @@
<commons-io.version>2.5</commons-io.version>
<commons-lang.version>2.5</commons-lang.version>
<commons-exec.version>1.1</commons-exec.version>
+ <commons-collections.version>3.2.2</commons-collections.version>
<commons-fileupload.version>1.3.2</commons-fileupload.version>
<commons-codec.version>1.6</commons-codec.version>
<commons-cli.version>1.3.1</commons-cli.version>
@@ -568,6 +569,11 @@
<version>${commons-exec.version}</version>
</dependency>
<dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ <version>${commons-collections.version}</version>
+ </dependency>
+ <dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>${commons-lang.version}</version>
http://git-wip-us.apache.org/repos/asf/storm/blob/0d755eeb/storm-core/pom.xml
----------------------------------------------------------------------
diff --git a/storm-core/pom.xml b/storm-core/pom.xml
index 0b5ea02..b05fd35 100644
--- a/storm-core/pom.xml
+++ b/storm-core/pom.xml
@@ -161,6 +161,10 @@
<scope>compile</scope>
</dependency>
<dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ </dependency>
+ <dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<scope>compile</scope>
@@ -502,6 +506,7 @@
<include>org.jgrapht:jgrapht-core</include>
<include>org.apache.commons:commons-exec</include>
<include>org.apache.commons:commons-compress</include>
+ <include>commons-collections:commons-collections</include>
<include>org.apache.hadoop:hadoop-auth</include>
<include>commons-io:commons-io</include>
<include>commons-codec:commons-codec</include>
@@ -664,6 +669,10 @@
<shadedPattern>org.apache.storm.shade.org.apache.commons.lang</shadedPattern>
</relocation>
<relocation>
+ <pattern>org.apache.commons.collections</pattern>
+ <shadedPattern>org.apache.storm.shade.org.apache.commons.collections</shadedPattern>
+ </relocation>
+ <relocation>
<pattern>org.json.simple</pattern>
<shadedPattern>org.apache.storm.shade.org.json.simple</shadedPattern>
</relocation>
@@ -780,6 +789,13 @@
</excludes>
</filter>
<filter>
+ <artifact>commons-collections:commons-collections</artifact>
+ <excludes>
+ <exclude>META-INF/LICENSE.txt</exclude>
+ <exclude>META-INF/NOTICE.txt</exclude>
+ </excludes>
+ </filter>
+ <filter>
<artifact>commons-io:commons-io</artifact>
<excludes>
<exclude>META-INF/LICENSE.txt</exclude>
http://git-wip-us.apache.org/repos/asf/storm/blob/0d755eeb/storm-core/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java b/storm-core/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
index 8700746..f3799af 100644
--- a/storm-core/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
+++ b/storm-core/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
@@ -181,4 +181,26 @@ public class ResourceUtils {
}
return str.toString();
}
+
+ /**
+ * Calculate the sum of a collection of doubles
+ * @param list collection of doubles
+ * @return the sum of of collection of doubles
+ */
+ public static double sum(Collection<Double> list) {
+ double sum = 0.0;
+ for (Double elem : list) {
+ sum += elem;
+ }
+ return sum;
+ }
+
+ /**
+ * Caculate the average of a collection of doubles
+ * @param list a collection of doubles
+ * @return the average of collection of doubles
+ */
+ public static double avg(Collection<Double> list) {
+ return sum(list) / list.size();
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0d755eeb/storm-core/src/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java b/storm-core/src/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
index 8957dc0..df3300c 100644
--- a/storm-core/src/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
+++ b/storm-core/src/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
@@ -20,25 +20,26 @@ package org.apache.storm.scheduler.resource.strategies.scheduling;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Queue;
-import java.util.TreeMap;
import java.util.HashSet;
-import java.util.Iterator;
+import java.util.Set;
import java.util.TreeSet;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.resource.RAS_Node;
import org.apache.storm.scheduler.resource.RAS_Nodes;
+import org.apache.storm.scheduler.resource.ResourceUtils;
import org.apache.storm.scheduler.resource.SchedulingResult;
import org.apache.storm.scheduler.resource.SchedulingState;
import org.apache.storm.scheduler.resource.SchedulingStatus;
+import org.apache.commons.collections.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,13 +52,11 @@ public class DefaultResourceAwareStrategy implements IStrategy {
private static final Logger LOG = LoggerFactory.getLogger(DefaultResourceAwareStrategy.class);
private Cluster _cluster;
private Topologies _topologies;
- private RAS_Node refNode = null;
private Map<String, List<String>> _clusterInfo;
private RAS_Nodes _nodes;
- private final double CPU_WEIGHT = 1.0;
- private final double MEM_WEIGHT = 1.0;
- private final double NETWORK_WEIGHT = 1.0;
+ private TreeSet<ObjectResources> _sortedRacks = null;
+ private Map<String, TreeSet<ObjectResources>> _rackIdToSortedNodes = new HashMap<String, TreeSet<ObjectResources>>();
public void prepare (SchedulingState schedulingState) {
_cluster = schedulingState.cluster;
@@ -67,29 +66,12 @@ public class DefaultResourceAwareStrategy implements IStrategy {
LOG.debug(this.getClusterInfo());
}
- //the returned TreeMap keeps the Components sorted
- private TreeMap<Integer, List<ExecutorDetails>> getPriorityToExecutorDetailsListMap(
- Queue<Component> ordered__Component_list, Collection<ExecutorDetails> unassignedExecutors) {
- TreeMap<Integer, List<ExecutorDetails>> retMap = new TreeMap<>();
- Integer rank = 0;
- for (Component ras_comp : ordered__Component_list) {
- retMap.put(rank, new ArrayList<ExecutorDetails>());
- for(ExecutorDetails exec : ras_comp.execs) {
- if(unassignedExecutors.contains(exec)) {
- retMap.get(rank).add(exec);
- }
- }
- rank++;
- }
- return retMap;
- }
-
public SchedulingResult schedule(TopologyDetails td) {
if (_nodes.getNodes().size() <= 0) {
LOG.warn("No available nodes to schedule tasks on!");
return SchedulingResult.failure(SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES, "No available nodes to schedule tasks on!");
}
- Collection<ExecutorDetails> unassignedExecutors = _cluster.getUnassignedExecutors(td);
+ Collection<ExecutorDetails> unassignedExecutors = new HashSet<ExecutorDetails>(_cluster.getUnassignedExecutors(td));
Map<WorkerSlot, Collection<ExecutorDetails>> schedulerAssignmentMap = new HashMap<>();
LOG.debug("ExecutorsNeedScheduling: {}", unassignedExecutors);
Collection<ExecutorDetails> scheduledTasks = new ArrayList<>();
@@ -100,25 +82,16 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return SchedulingResult.failure(SchedulingStatus.FAIL_INVALID_TOPOLOGY, "Cannot find a Spout!");
}
- Queue<Component> ordered__Component_list = bfs(td, spouts);
+ //order executors to be scheduled
+ List<ExecutorDetails> orderedExecutors = orderExecutors(td, unassignedExecutors);
- Map<Integer, List<ExecutorDetails>> priorityToExecutorMap = getPriorityToExecutorDetailsListMap(ordered__Component_list, unassignedExecutors);
Collection<ExecutorDetails> executorsNotScheduled = new HashSet<>(unassignedExecutors);
- Integer longestPriorityListSize = this.getLongestPriorityListSize(priorityToExecutorMap);
- //Pick the first executor with priority one, then the 1st exec with priority 2, so on an so forth.
- //Once we reach the last priority, we go back to priority 1 and schedule the second task with priority 1.
- for (int i = 0; i < longestPriorityListSize; i++) {
- for (Entry<Integer, List<ExecutorDetails>> entry : priorityToExecutorMap.entrySet()) {
- Iterator<ExecutorDetails> it = entry.getValue().iterator();
- if (it.hasNext()) {
- ExecutorDetails exec = it.next();
- LOG.debug("\n\nAttempting to schedule: {} of component {}[ REQ {} ] with rank {}",
- new Object[] { exec, td.getExecutorToComponent().get(exec),
- td.getTaskResourceReqList(exec), entry.getKey() });
- scheduleExecutor(exec, td, schedulerAssignmentMap, scheduledTasks);
- it.remove();
- }
- }
+
+ for (ExecutorDetails exec : orderedExecutors) {
+ LOG.debug("\n\nAttempting to schedule: {} of component {}[ REQ {} ]",
+ exec, td.getExecutorToComponent().get(exec),
+ td.getTaskResourceReqList(exec));
+ scheduleExecutor(exec, td, schedulerAssignmentMap, scheduledTasks);
}
executorsNotScheduled.removeAll(scheduledTasks);
@@ -146,6 +119,14 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return result;
}
+ /**
+ * Schedule executor exec from topology td
+ *
+ * @param exec the executor to schedule
+ * @param td the topology executor exec is a part of
+ * @param schedulerAssignmentMap the assignments already calculated
+ * @param scheduledTasks executors that have been scheduled
+ */
private void scheduleExecutor(ExecutorDetails exec, TopologyDetails td, Map<WorkerSlot,
Collection<ExecutorDetails>> schedulerAssignmentMap, Collection<ExecutorDetails> scheduledTasks) {
WorkerSlot targetSlot = this.findWorkerForExec(exec, td, schedulerAssignmentMap);
@@ -167,69 +148,59 @@ public class DefaultResourceAwareStrategy implements IStrategy {
}
}
+ /**
+ * Find a worker to schedule executor exec on
+ *
+ * @param exec the executor to schedule
+ * @param td the topology that the executor is a part of
+ * @param scheduleAssignmentMap already calculated assignments
+ * @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, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
WorkerSlot ws = null;
- // first scheduling
- if (this.refNode == 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.
- for (RackResources rack : sortRacks(td.getId())) {
- ws = this.getBestWorker(exec, td, rack.id, scheduleAssignmentMap);
- if (ws != null) {
- LOG.debug("best rack: {}", rack.id);
- break;
- }
- }
- } else {
- ws = this.getBestWorker(exec, td, scheduleAssignmentMap);
+
+ // 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(), scheduleAssignmentMap);
}
- if (ws != null) {
- this.refNode = this.idToNode(ws.getNodeId());
+
+ for (ObjectResources rack : _sortedRacks) {
+ ws = this.getBestWorker(exec, td, rack.id, scheduleAssignmentMap);
+ if (ws != null) {
+ LOG.debug("best rack: {}", rack.id);
+ break;
+ }
}
- LOG.debug("reference node for the resource aware scheduler is: {}", this.refNode);
return ws;
}
- private WorkerSlot getBestWorker(ExecutorDetails exec, TopologyDetails td, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
- return this.getBestWorker(exec, td, null, scheduleAssignmentMap);
- }
-
+ /**
+ * 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
+ * @param scheduleAssignmentMap already calculated assignments
+ * @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, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
- double taskMem = td.getTotalMemReqTask(exec);
- double taskCPU = td.getTotalCpuReqTask(exec);
- List<RAS_Node> nodes;
- if(rackId != null) {
- nodes = this.getAvailableNodesFromRack(rackId);
- } else {
- nodes = this.getAvailableNodes();
+ if (!_rackIdToSortedNodes.containsKey(rackId)) {
+ _rackIdToSortedNodes.put(rackId, sortNodes(this.getAvailableNodesFromRack(rackId), rackId, td.getId(), scheduleAssignmentMap));
}
- //First sort nodes by distance
- TreeMap<Double, RAS_Node> nodeRankMap = new TreeMap<>();
- for (RAS_Node n : nodes) {
- if(n.getFreeSlots().size()>0) {
- if (n.getAvailableMemoryResources() >= taskMem
- && n.getAvailableCpuResources() >= taskCPU) {
- double a = Math.pow(((taskCPU - n.getAvailableCpuResources())/(n.getAvailableCpuResources() + 1))
- * this.CPU_WEIGHT, 2);
- double b = Math.pow(((taskMem - n.getAvailableMemoryResources())/(n.getAvailableMemoryResources() + 1))
- * this.MEM_WEIGHT, 2);
- double c = 0.0;
- if(this.refNode != null) {
- c = Math.pow(this.distToNode(this.refNode, n)
- * this.NETWORK_WEIGHT, 2);
+
+ TreeSet<ObjectResources> sortedNodes = _rackIdToSortedNodes.get(rackId);
+
+ double taskMem = td.getTotalMemReqTask(exec);
+ double taskCPU = td.getTotalCpuReqTask(exec);
+ for (ObjectResources nodeResources : sortedNodes) {
+ RAS_Node n = _nodes.getNodeById(nodeResources.id);
+ if (n.getAvailableCpuResources() >= taskCPU && n.getAvailableMemoryResources() >= taskMem && n.getFreeSlots().size() > 0) {
+ for (WorkerSlot ws : n.getFreeSlots()) {
+ if (checkWorkerConstraints(exec, ws, td, scheduleAssignmentMap)) {
+ return ws;
}
- double distance = Math.sqrt(a + b + c);
- nodeRankMap.put(distance, n);
- }
- }
- }
- //Then, pick worker from closest node that satisfy constraints
- for(Map.Entry<Double, RAS_Node> entry : nodeRankMap.entrySet()) {
- RAS_Node n = entry.getValue();
- for(WorkerSlot ws : n.getFreeSlots()) {
- if(checkWorkerConstraints(exec, ws, td, scheduleAssignmentMap)) {
- return ws;
}
}
}
@@ -237,18 +208,40 @@ public class DefaultResourceAwareStrategy implements IStrategy {
}
/**
- * class to keep track of resources on a rack
+ * interface for calculating the number of existing executors scheduled on a object (rack or node)
+ */
+ private interface ExistingScheduleFunc {
+ int getNumExistingSchedule(String objectId);
+ }
+
+ /**
+ * a class to contain individual object resources as well as cumulative stats
+ */
+ static class AllResources {
+ List<ObjectResources> objectResources = new LinkedList<ObjectResources>();
+ double availMemResourcesOverall = 0.0;
+ double totalMemResourcesOverall = 0.0;
+ double availCpuResourcesOverall = 0.0;
+ double totalCpuResourcesOverall = 0.0;
+ String identifier;
+
+ public AllResources(String identifier) {
+ this.identifier = identifier;
+ }
+ }
+
+ /**
+ * class to keep track of resources on a rack or node
*/
- class RackResources {
+ static class ObjectResources {
String id;
double availMem = 0.0;
double totalMem = 0.0;
double availCpu = 0.0;
double totalCpu = 0.0;
- int freeSlots = 0;
- int totalSlots = 0;
double effectiveResources = 0.0;
- public RackResources(String id) {
+
+ public ObjectResources(String id) {
this.id = id;
}
@@ -259,8 +252,87 @@ public class DefaultResourceAwareStrategy implements IStrategy {
}
/**
+ * Sorted Nodes
*
- * @param topoId
+ * @param availNodes a list of all the nodes we want to sort
+ * @param rackId the rack id availNodes are a part of
+ * @param topoId the topology that we are trying to schedule
+ * @param scheduleAssignmentMap calculated assignments so far
+ * @return a sorted list of nodes
+ * <p>
+ * Nodes are sorted by two criteria. 1) the number executors of the topology that needs to be scheduled is already on the node in descending order.
+ * The reasoning to sort based on criterion 1 is so we schedule the rest of a topology on the same node as the existing executors of the topology.
+ * 2) the subordinate/subservient resource availability percentage of a node in descending order
+ * We calculate the resource availability percentage by dividing the resource availability on the node by the resource availability of the entire rack
+ * By doing this calculation, nodes that have exhausted or little of one of the resources mentioned above will be ranked after nodes that have more balanced resource availability.
+ * So we will be less likely to pick a node that have a lot of one resource but a low amount of another.
+ */
+ private TreeSet<ObjectResources> sortNodes(List<RAS_Node> availNodes, String rackId, final String topoId, final Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
+ AllResources allResources = new AllResources("RACK");
+ List<ObjectResources> nodes = allResources.objectResources;
+ final Map<String, String> nodeIdToRackId = new HashMap<String, String>();
+
+ for (RAS_Node ras_node : availNodes) {
+ String nodeId = ras_node.getId();
+ ObjectResources node = new ObjectResources(nodeId);
+
+ double availMem = ras_node.getAvailableMemoryResources();
+ double availCpu = ras_node.getAvailableCpuResources();
+ int freeSlots = ras_node.totalSlotsFree();
+ double totalMem = ras_node.getTotalMemoryResources();
+ double totalCpu = ras_node.getTotalCpuResources();
+ int totalSlots = ras_node.totalSlots();
+
+ node.availMem = availMem;
+ node.totalMem = totalMem;
+ node.availCpu = availCpu;
+ node.totalCpu = totalCpu;
+ nodes.add(node);
+
+ allResources.availMemResourcesOverall += availMem;
+ allResources.availCpuResourcesOverall += availCpu;
+
+ allResources.totalMemResourcesOverall += totalMem;
+ allResources.totalCpuResourcesOverall += totalCpu;
+ }
+
+
+ LOG.debug("Rack {}: Overall Avail [ CPU {} MEM {} ] Total [ CPU {} MEM {} ]",
+ rackId, allResources.availCpuResourcesOverall, allResources.availMemResourcesOverall, allResources.totalCpuResourcesOverall, allResources.totalMemResourcesOverall);
+
+ return sortObjectResources(allResources, new ExistingScheduleFunc() {
+ @Override
+ public int getNumExistingSchedule(String objectId) {
+
+ //Get execs already assigned in rack
+ Collection<ExecutorDetails> execs = new LinkedList<ExecutorDetails>();
+ if (_cluster.getAssignmentById(topoId) != null) {
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : _cluster.getAssignmentById(topoId).getExecutorToSlot().entrySet()) {
+ WorkerSlot workerSlot = entry.getValue();
+ ExecutorDetails exec = entry.getKey();
+ if (workerSlot.getNodeId().equals(objectId)) {
+ execs.add(exec);
+ }
+ }
+ }
+ // get execs already scheduled in the current scheduling
+ for (Map.Entry<WorkerSlot, Collection<ExecutorDetails>> entry : scheduleAssignmentMap.entrySet()) {
+
+ WorkerSlot workerSlot = entry.getKey();
+ if (workerSlot.getNodeId().equals(objectId)) {
+ execs.addAll(entry.getValue());
+ }
+ }
+ return execs.size();
+ }
+ });
+ }
+
+ /**
+ * Sort racks
+ *
+ * @param topoId topology id
+ * @param scheduleAssignmentMap calculated assignments so far
* @return a sorted list of racks
* Racks are sorted by two criteria. 1) the number executors of the topology that needs to be scheduled is already on the rack in descending order.
* The reasoning to sort based on criterion 1 is so we schedule the rest of a topology on the same rack as the existing executors of the topology.
@@ -269,117 +341,161 @@ public class DefaultResourceAwareStrategy implements IStrategy {
* By doing this calculation, racks that have exhausted or little of one of the resources mentioned above will be ranked after racks that have more balanced resource availability.
* So we will be less likely to pick a rack that have a lot of one resource but a low amount of another.
*/
- TreeSet<RackResources> sortRacks(final String topoId) {
- List<RackResources> racks = new LinkedList<RackResources>();
- final Map<String, String> nodeIdToRackId = new HashMap<String, String>();
- double availMemResourcesOverall = 0.0;
- double totalMemResourcesOverall = 0.0;
-
- double availCpuResourcesOverall = 0.0;
- double totalCpuResourcesOverall = 0.0;
+ TreeSet<ObjectResources> sortRacks(final String topoId, final Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
+ AllResources allResources = new AllResources("Cluster");
+ List<ObjectResources> racks = allResources.objectResources;
- int freeSlotsOverall = 0;
- int totalSlotsOverall = 0;
+ final Map<String, String> nodeIdToRackId = new HashMap<String, String>();
- for (Entry<String, List<String>> entry : _clusterInfo.entrySet()) {
+ for (Map.Entry<String, List<String>> entry : _clusterInfo.entrySet()) {
String rackId = entry.getKey();
List<String> nodeIds = entry.getValue();
- RackResources rack = new RackResources(rackId);
+ ObjectResources rack = new ObjectResources(rackId);
racks.add(rack);
for (String nodeId : nodeIds) {
RAS_Node node = _nodes.getNodeById(this.NodeHostnameToId(nodeId));
double availMem = node.getAvailableMemoryResources();
double availCpu = node.getAvailableCpuResources();
- double freeSlots = node.totalSlotsFree();
double totalMem = node.getTotalMemoryResources();
double totalCpu = node.getTotalCpuResources();
- double totalSlots = node.totalSlots();
rack.availMem += availMem;
rack.totalMem += totalMem;
rack.availCpu += availCpu;
rack.totalCpu += totalCpu;
- rack.freeSlots += freeSlots;
- rack.totalSlots += totalSlots;
nodeIdToRackId.put(nodeId, rack.id);
- availMemResourcesOverall += availMem;
- availCpuResourcesOverall += availCpu;
- freeSlotsOverall += freeSlots;
+ allResources.availMemResourcesOverall += availMem;
+ allResources.availCpuResourcesOverall += availCpu;
- totalMemResourcesOverall += totalMem;
- totalCpuResourcesOverall += totalCpu;
- totalSlotsOverall += totalSlots;
+ allResources.totalMemResourcesOverall += totalMem;
+ allResources.totalCpuResourcesOverall += totalCpu;
}
}
+ LOG.debug("Cluster Overall Avail [ CPU {} MEM {} ] Total [ CPU {} MEM {} ]",
+ allResources.availCpuResourcesOverall, allResources.availMemResourcesOverall, allResources.totalCpuResourcesOverall, allResources.totalMemResourcesOverall);
+
+ return sortObjectResources(allResources, new ExistingScheduleFunc() {
+ @Override
+ public int getNumExistingSchedule(String objectId) {
+
+ String rackId = objectId;
+ //Get execs already assigned in rack
+ Collection<ExecutorDetails> execs = new LinkedList<ExecutorDetails>();
+ if (_cluster.getAssignmentById(topoId) != null) {
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : _cluster.getAssignmentById(topoId).getExecutorToSlot().entrySet()) {
+ String nodeId = entry.getValue().getNodeId();
+ String hostname = idToNode(nodeId).getHostname();
+ ExecutorDetails exec = entry.getKey();
+ if (nodeIdToRackId.get(hostname) != null && nodeIdToRackId.get(hostname).equals(rackId)) {
+ execs.add(exec);
+ }
+ }
+ }
+ // get execs already scheduled in the current scheduling
+ for (Map.Entry<WorkerSlot, Collection<ExecutorDetails>> entry : scheduleAssignmentMap.entrySet()) {
+ WorkerSlot workerSlot = entry.getKey();
+ String nodeId = workerSlot.getNodeId();
+ String hostname = idToNode(nodeId).getHostname();
+ if (nodeIdToRackId.get(hostname).equals(rackId)) {
+ execs.addAll(entry.getValue());
+ }
+ }
+ return execs.size();
+ }
+ });
+ }
+
+ /**
+ * Sort objects by the following two criteria. 1) the number executors of the topology that needs to be scheduled is already on the object (node or rack) in descending order.
+ * The reasoning to sort based on criterion 1 is so we schedule the rest of a topology on the same object (node or rack) as the existing executors of the topology.
+ * 2) the subordinate/subservient resource availability percentage of a rack in descending order
+ * We calculate the resource availability percentage by dividing the resource availability of the object (node or rack) by the resource availability of the entire rack or cluster depending on if object
+ * references a node or a rack.
+ * By doing this calculation, objects (node or rack) that have exhausted or little of one of the resources mentioned above will be ranked after racks that have more balanced resource availability.
+ * So we will be less likely to pick a rack that have a lot of one resource but a low amount of another.
+ *
+ * @param allResources contains all individual ObjectResources as well as cumulative stats
+ * @param existingScheduleFunc a function to get existing executors already scheduled on this object
+ * @return a sorted list of ObjectResources
+ */
+ private TreeSet<ObjectResources> sortObjectResources(final AllResources allResources, final ExistingScheduleFunc existingScheduleFunc) {
- LOG.debug("Cluster Overall Avail [ CPU {} MEM {} Slots {} ] Total [ CPU {} MEM {} Slots {} ]",
- availCpuResourcesOverall, availMemResourcesOverall, freeSlotsOverall, totalCpuResourcesOverall, totalMemResourcesOverall, totalSlotsOverall);
- for (RackResources rack : racks) {
- if (availCpuResourcesOverall <= 0.0 || availMemResourcesOverall <= 0.0 || freeSlotsOverall <= 0.0) {
- rack.effectiveResources = 0.0;
+ for (ObjectResources objectResources : allResources.objectResources) {
+ StringBuilder sb = new StringBuilder();
+ if (allResources.availCpuResourcesOverall <= 0.0 || allResources.availMemResourcesOverall <= 0.0) {
+ objectResources.effectiveResources = 0.0;
} else {
- rack.effectiveResources = Math.min(Math.min((rack.availCpu / availCpuResourcesOverall), (rack.availMem / availMemResourcesOverall)), ((double) rack.freeSlots / (double) freeSlotsOverall));
+ List<Double> values = new LinkedList<Double>();
+
+ //add cpu
+ double cpuPercent = (objectResources.availCpu / allResources.availCpuResourcesOverall) * 100.0;
+ values.add(cpuPercent);
+ sb.append(String.format("CPU %f(%f%%) ", objectResources.availCpu, cpuPercent));
+
+ //add memory
+ double memoryPercent = (objectResources.availMem / allResources.availMemResourcesOverall) * 100.0;
+ values.add(memoryPercent);
+ sb.append(String.format("MEM %f(%f%%) ", objectResources.availMem, memoryPercent));
+
+ objectResources.effectiveResources = Collections.min(values);
}
- LOG.debug("rack: {} Avail [ CPU {}({}%) MEM {}({}%) Slots {}({}%) ] Total [ CPU {} MEM {} Slots {} ] effective resources: {}",
- rack.id, rack.availCpu, rack.availCpu/availCpuResourcesOverall * 100.0, rack.availMem, rack.availMem/availMemResourcesOverall * 100.0,
- rack.freeSlots, ((double) rack.freeSlots / (double) freeSlotsOverall) * 100.0, rack.totalCpu, rack.totalMem, rack.totalSlots, rack.effectiveResources);
+ LOG.debug("{}: Avail [ {} ] Total [ CPU {} MEM {}] effective resources: {}",
+ objectResources.id, sb.toString(),
+ objectResources.totalCpu, objectResources.totalMem, objectResources.effectiveResources);
}
- TreeSet<RackResources> sortedRacks = new TreeSet<RackResources>(new Comparator<RackResources>() {
+ TreeSet<ObjectResources> sortedObjectResources = new TreeSet<ObjectResources>(new Comparator<ObjectResources>() {
@Override
- public int compare(RackResources o1, RackResources o2) {
+ public int compare(ObjectResources o1, ObjectResources o2) {
- int execsScheduledInRack1 = getExecutorsScheduledinRack(topoId, o1.id, nodeIdToRackId).size();
- int execsScheduledInRack2 = getExecutorsScheduledinRack(topoId, o2.id, nodeIdToRackId).size();
- if (execsScheduledInRack1 > execsScheduledInRack2) {
+ int execsScheduled1 = existingScheduleFunc.getNumExistingSchedule(o1.id);
+ int execsScheduled2 = existingScheduleFunc.getNumExistingSchedule(o2.id);
+ if (execsScheduled1 > execsScheduled2) {
return -1;
- } else if (execsScheduledInRack1 < execsScheduledInRack2) {
+ } else if (execsScheduled1 < execsScheduled2) {
return 1;
} else {
if (o1.effectiveResources > o2.effectiveResources) {
return -1;
} else if (o1.effectiveResources < o2.effectiveResources) {
return 1;
- }
- else {
- return o1.id.compareTo(o2.id);
+ } else {
+ List<Double> o1_values = new LinkedList<Double>();
+ List<Double> o2_values = new LinkedList<Double>();
+ o1_values.add((o1.availCpu / allResources.availCpuResourcesOverall) * 100.0);
+ o2_values.add((o2.availCpu / allResources.availCpuResourcesOverall) * 100.0);
+
+ o1_values.add((o1.availMem / allResources.availMemResourcesOverall) * 100.0);
+ o2_values.add((o2.availMem / allResources.availMemResourcesOverall) * 100.0);
+
+ double o1_avg = ResourceUtils.avg(o1_values);
+ double o2_avg = ResourceUtils.avg(o2_values);
+
+ if (o1_avg > o2_avg) {
+ return -1;
+ } else if (o1_avg < o2_avg) {
+ return 1;
+ } else {
+ return o1.id.compareTo(o2.id);
+ }
}
}
}
});
- sortedRacks.addAll(racks);
- LOG.debug("Sorted rack: {}", sortedRacks);
- return sortedRacks;
- }
-
- private Collection<ExecutorDetails> getExecutorsScheduledinRack(String topoId, String rackId, Map<String, String> nodeToRack) {
- Collection<ExecutorDetails> execs = new LinkedList<ExecutorDetails>();
- if (_cluster.getAssignmentById(topoId) != null) {
- for (Entry<ExecutorDetails, WorkerSlot> entry : _cluster.getAssignmentById(topoId).getExecutorToSlot().entrySet()) {
- String nodeId = entry.getValue().getNodeId();
- String hostname = idToNode(nodeId).getHostname();
- ExecutorDetails exec = entry.getKey();
- if (nodeToRack.get(hostname) != null && nodeToRack.get(hostname).equals(rackId)) {
- execs.add(exec);
- }
- }
- }
- return execs;
- }
-
- private Double distToNode(RAS_Node src, RAS_Node dest) {
- if (src.getId().equals(dest.getId())) {
- return 0.0;
- } else if (this.nodeToRack(src).equals(this.nodeToRack(dest))) {
- return 0.5;
- } else {
- return 1.0;
- }
+ sortedObjectResources.addAll(allResources.objectResources);
+ LOG.debug("Sorted Object Resources: {}", sortedObjectResources);
+ return sortedObjectResources;
}
+ /**
+ * Get the rack on which a node is a part of
+ *
+ * @param node the node to find out which rack its on
+ * @return the rack id
+ */
private String nodeToRack(RAS_Node node) {
- for (Entry<String, List<String>> entry : _clusterInfo
+ for (Map.Entry<String, List<String>> entry : _clusterInfo
.entrySet()) {
if (entry.getValue().contains(node.getHostname())) {
return entry.getKey();
@@ -389,14 +505,12 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return null;
}
- private List<RAS_Node> getAvailableNodes() {
- LinkedList<RAS_Node> nodes = new LinkedList<>();
- for (String rackId : _clusterInfo.keySet()) {
- nodes.addAll(this.getAvailableNodesFromRack(rackId));
- }
- return nodes;
- }
-
+ /**
+ * get a list nodes from a rack
+ *
+ * @param rackId the rack id of the rack to get nodes from
+ * @return a list of nodes
+ */
private List<RAS_Node> getAvailableNodesFromRack(String rackId) {
List<RAS_Node> retList = new ArrayList<>();
for (String node_id : _clusterInfo.get(rackId)) {
@@ -406,59 +520,127 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return retList;
}
- private List<WorkerSlot> getAvailableWorkersFromRack(String rackId) {
- List<RAS_Node> nodes = this.getAvailableNodesFromRack(rackId);
- List<WorkerSlot> workers = new LinkedList<>();
- for(RAS_Node node : nodes) {
- workers.addAll(node.getFreeSlots());
- }
- return workers;
+ /**
+ * sort components by the number of in and out connections that need to be made
+ *
+ * @param componentMap The components that need to be sorted
+ * @return a sorted set of components
+ */
+ private Set<Component> sortComponents(final Map<String, Component> componentMap) {
+ Set<Component> sortedComponents = new TreeSet<Component>(new Comparator<Component>() {
+ @Override
+ public int compare(Component o1, Component o2) {
+ int connections1 = 0;
+ int connections2 = 0;
+
+ for (String childId : (List<String>) ListUtils.union(o1.children, o1.parents)) {
+ connections1 += (componentMap.get(childId).execs.size() * o1.execs.size());
+ }
+
+ for (String childId : (List<String>) ListUtils.union(o2.children, o2.parents)) {
+ connections2 += (componentMap.get(childId).execs.size() * o2.execs.size());
+ }
+
+ if (connections1 > connections1) {
+ return -1;
+ } else if (connections1 < connections2) {
+ return 1;
+ } else {
+ return o1.id.compareTo(o2.id);
+ }
+ }
+ });
+ sortedComponents.addAll(componentMap.values());
+ return sortedComponents;
}
- private List<WorkerSlot> getAvailableWorker() {
- List<WorkerSlot> workers = new LinkedList<>();
- for (String rackId : _clusterInfo.keySet()) {
- workers.addAll(this.getAvailableWorkersFromRack(rackId));
- }
- return workers;
+ /**
+ * Sort a component's neighbors by the number of connections it needs to make with this component
+ *
+ * @param thisComp the component that we need to sort its neighbors
+ * @param componentMap all the components to sort
+ * @return a sorted set of components
+ */
+ private Set<Component> sortNeighbors(final Component thisComp, final Map<String, Component> componentMap) {
+ Set<Component> sortedComponents = new TreeSet<Component>(new Comparator<Component>() {
+ @Override
+ public int compare(Component o1, Component o2) {
+ int connections1 = o1.execs.size() * thisComp.execs.size();
+ int connections2 = o2.execs.size() * thisComp.execs.size();
+ if (connections1 > connections2) {
+ return -1;
+ } else if (connections1 < connections2) {
+ return 1;
+ } else {
+ return o1.id.compareTo(o2.id);
+ }
+ }
+ });
+ sortedComponents.addAll(componentMap.values());
+ return sortedComponents;
}
/**
- * Breadth first traversal of the topology DAG
- * @param td
- * @param spouts
- * @return A partial ordering of components
+ * Order executors based on how many in and out connections it will potentially need to make.
+ * First order components by the number of in and out connections it will have. Then iterate through the sorted list of components.
+ * For each component sort the neighbors of that component by how many connections it will have to make with that component.
+ * Add an executor from this component and then from each neighboring component in sorted order. Do this until there is nothing left to schedule
+ *
+ * @param td The topology the executors belong to
+ * @param unassignedExecutors a collection of unassigned executors that need to be unassigned. Should only try to assign executors from this list
+ * @return a list of executors in sorted order
*/
- private Queue<Component> bfs(TopologyDetails td, List<Component> spouts) {
- // Since queue is a interface
- Queue<Component> ordered__Component_list = new LinkedList<Component>();
- HashSet<String> visited = new HashSet<>();
-
- /* start from each spout that is not visited, each does a breadth-first traverse */
- for (Component spout : spouts) {
- if (!visited.contains(spout.id)) {
- Queue<Component> queue = new LinkedList<>();
- visited.add(spout.id);
- queue.offer(spout);
- while (!queue.isEmpty()) {
- Component comp = queue.poll();
- ordered__Component_list.add(comp);
- List<String> neighbors = new ArrayList<>();
- neighbors.addAll(comp.children);
- neighbors.addAll(comp.parents);
- for (String nbID : neighbors) {
- if (!visited.contains(nbID)) {
- Component child = td.getComponents().get(nbID);
- visited.add(nbID);
- queue.offer(child);
- }
- }
+ private List<ExecutorDetails> orderExecutors(TopologyDetails td, Collection<ExecutorDetails> unassignedExecutors) {
+ Map<String, Component> componentMap = td.getComponents();
+ List<ExecutorDetails> execsScheduled = new LinkedList<>();
+
+ Map<String, Queue<ExecutorDetails>> compToExecsToSchedule = new HashMap<>();
+ for (Component component : componentMap.values()) {
+ compToExecsToSchedule.put(component.id, new LinkedList<ExecutorDetails>());
+ for (ExecutorDetails exec : component.execs) {
+ if (unassignedExecutors.contains(exec)) {
+ compToExecsToSchedule.get(component.id).add(exec);
}
}
}
- return ordered__Component_list;
+
+ Set<Component> sortedComponents = sortComponents(componentMap);
+ sortedComponents.addAll(componentMap.values());
+
+ for (Component currComp : sortedComponents) {
+ Map<String, Component> neighbors = new HashMap<String, Component>();
+ for (String compId : (List<String>) ListUtils.union(currComp.children, currComp.parents)) {
+ neighbors.put(compId, componentMap.get(compId));
+ }
+ Set<Component> sortedNeighbors = sortNeighbors(currComp, neighbors);
+ Queue<ExecutorDetails> currCompExesToSched = compToExecsToSchedule.get(currComp.id);
+
+ boolean flag = false;
+ do {
+ flag = false;
+ if (!currCompExesToSched.isEmpty()) {
+ execsScheduled.add(currCompExesToSched.poll());
+ flag = true;
+ }
+
+ for (Component neighborComp : sortedNeighbors) {
+ Queue<ExecutorDetails> neighborCompExesToSched = compToExecsToSchedule.get(neighborComp.id);
+ if (!neighborCompExesToSched.isEmpty()) {
+ execsScheduled.add(neighborCompExesToSched.poll());
+ flag = true;
+ }
+ }
+ } while (flag);
+ }
+ return execsScheduled;
}
+ /**
+ * Get a list of all the spouts in the topology
+ *
+ * @param td topology to get spouts from
+ * @return a list of spouts
+ */
private List<Component> getSpouts(TopologyDetails td) {
List<Component> spouts = new ArrayList<>();
@@ -470,22 +652,12 @@ public class DefaultResourceAwareStrategy implements IStrategy {
return spouts;
}
- private Integer getLongestPriorityListSize(Map<Integer, List<ExecutorDetails>> priorityToExecutorMap) {
- Integer mostNum = 0;
- for (List<ExecutorDetails> execs : priorityToExecutorMap.values()) {
- Integer numExecs = execs.size();
- if (mostNum < numExecs) {
- mostNum = numExecs;
- }
- }
- return mostNum;
- }
-
/**
* Get the remaining amount memory that can be assigned to a worker given the set worker max heap size
- * @param ws
- * @param td
- * @param scheduleAssignmentMap
+ *
+ * @param ws the worker to get the remaining amount of memory that can be assigned to it
+ * @param td the topology that has executors running on the worker
+ * @param scheduleAssignmentMap the schedulings calculated so far
* @return The remaining amount of memory
*/
private Double getWorkerScheduledMemoryAvailable(WorkerSlot ws, TopologyDetails td, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
@@ -495,16 +667,17 @@ public class DefaultResourceAwareStrategy implements IStrategy {
/**
* Get the amount of memory already assigned to a worker
- * @param ws
- * @param td
- * @param scheduleAssignmentMap
+ *
+ * @param ws the worker to get the amount of memory assigned to a worker
+ * @param td the topology that has executors running on the worker
+ * @param scheduleAssignmentMap the schedulings calculated so far
* @return the amount of memory
*/
private Double getWorkerScheduledMemoryUse(WorkerSlot ws, TopologyDetails td, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
Double totalMem = 0.0;
Collection<ExecutorDetails> execs = scheduleAssignmentMap.get(ws);
- if(execs != null) {
- for(ExecutorDetails exec : execs) {
+ if (execs != null) {
+ for (ExecutorDetails exec : execs) {
totalMem += td.getTotalMemReqTask(exec);
}
}
@@ -514,15 +687,16 @@ public class DefaultResourceAwareStrategy implements IStrategy {
/**
* Checks whether we can schedule an Executor exec on the worker slot ws
* Only considers memory currently. May include CPU in the future
- * @param exec
- * @param ws
- * @param td
- * @param scheduleAssignmentMap
+ *
+ * @param exec the executor to check whether it can be asssigned to worker ws
+ * @param ws the worker to check whether executor exec can be assigned to it
+ * @param td the topology that the exec is from
+ * @param scheduleAssignmentMap the schedulings calculated so far
* @return a boolean: True denoting the exec can be scheduled on ws and false if it cannot
*/
private boolean checkWorkerConstraints(ExecutorDetails exec, WorkerSlot ws, TopologyDetails td, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) {
boolean retVal = false;
- if(this.getWorkerScheduledMemoryAvailable(ws, td, scheduleAssignmentMap) >= td.getTotalMemReqTask(exec)) {
+ if (this.getWorkerScheduledMemoryAvailable(ws, td, scheduleAssignmentMap) >= td.getTotalMemReqTask(exec)) {
retVal = true;
}
return retVal;
@@ -530,14 +704,15 @@ public class DefaultResourceAwareStrategy implements IStrategy {
/**
* Get the amount of resources available and total for each node
+ *
* @return a String with cluster resource info for debug
*/
private String getClusterInfo() {
String retVal = "Cluster info:\n";
- for(Entry<String, List<String>> clusterEntry : _clusterInfo.entrySet()) {
+ for (Map.Entry<String, List<String>> clusterEntry : _clusterInfo.entrySet()) {
String clusterId = clusterEntry.getKey();
retVal += "Rack: " + clusterId + "\n";
- for(String nodeHostname : clusterEntry.getValue()) {
+ for (String nodeHostname : clusterEntry.getValue()) {
RAS_Node node = this.idToNode(this.NodeHostnameToId(nodeHostname));
retVal += "-> Node: " + node.getHostname() + " " + node.getId() + "\n";
retVal += "--> Avail Resources: {Mem " + node.getAvailableMemoryResources() + ", CPU " + node.getAvailableCpuResources() + " Slots: " + node.totalSlotsFree() + "}\n";
@@ -549,7 +724,8 @@ public class DefaultResourceAwareStrategy implements IStrategy {
/**
* hostname to Id
- * @param hostname
+ *
+ * @param hostname the hostname to convert to node id
* @return the id of a node
*/
public String NodeHostnameToId(String hostname) {
@@ -567,12 +743,13 @@ public class DefaultResourceAwareStrategy implements IStrategy {
/**
* Find RAS_Node for specified node id
- * @param id
+ *
+ * @param id the node/supervisor id to lookup
* @return a RAS_Node object
*/
public RAS_Node idToNode(String id) {
RAS_Node ret = _nodes.getNodeById(id);
- if(ret == null) {
+ if (ret == null) {
LOG.error("Cannot find Node with Id: {}", id);
}
return ret;
http://git-wip-us.apache.org/repos/asf/storm/blob/0d755eeb/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
index 77f23aa..f533543 100644
--- a/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
+++ b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/scheduling/TestDefaultResourceAwareStrategy.java
@@ -33,7 +33,7 @@ import org.apache.storm.scheduler.resource.RAS_Node;
import org.apache.storm.scheduler.resource.RAS_Nodes;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
import org.apache.storm.scheduler.resource.SchedulingResult;
-import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.RackResources;
+import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.ObjectResources;
import org.apache.storm.scheduler.resource.SchedulingState;
import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
import org.apache.storm.scheduler.resource.User;
@@ -268,10 +268,10 @@ public class TestDefaultResourceAwareStrategy {
DefaultResourceAwareStrategy rs = new DefaultResourceAwareStrategy();
rs.prepare(new SchedulingState(new HashMap<String, User>(), cluster, topologies, config));
- TreeSet<RackResources> sortedRacks= rs.sortRacks(topo1.getId());
+ TreeSet<ObjectResources> sortedRacks= rs.sortRacks(topo1.getId(), new HashMap<WorkerSlot, Collection<ExecutorDetails>>());
Assert.assertEquals("# of racks sorted", 5, sortedRacks.size());
- Iterator<RackResources> it = sortedRacks.iterator();
+ 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
@@ -312,8 +312,6 @@ public class TestDefaultResourceAwareStrategy {
cluster.assign(targetSlot, topo2.getId(), Arrays.asList(targetExec));
}
- topologies.getById(topo2.getId()).getTotalMemoryResourceList();
-
rs = new DefaultResourceAwareStrategy();
rs.prepare(new SchedulingState(new HashMap<String, User>(), cluster, topologies, config));
// schedule topo2
[5/6] storm git commit: adding to documentation
Posted by je...@apache.org.
adding to documentation
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/667fc5e1
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/667fc5e1
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/667fc5e1
Branch: refs/heads/1.x-branch
Commit: 667fc5e177f916844fcc6171c316818054d4af19
Parents: 5e4c6c4
Author: Boyang Jerry Peng <je...@yahoo-inc.com>
Authored: Sun Oct 9 22:44:32 2016 -0500
Committer: Boyang Jerry Peng <je...@yahoo-inc.com>
Committed: Thu Oct 13 15:26:03 2016 -0500
----------------------------------------------------------------------
docs/Resource_Aware_Scheduler_overview.md | 253 +++++++++++++++----
.../ras_new_strategy_network_cdf_random.png | Bin 0 -> 157859 bytes
...tegy_network_metric_cdf_yahoo_topologies.png | Bin 0 -> 152034 bytes
...rategy_network_metric_improvement_random.png | Bin 0 -> 68976 bytes
.../ras_new_strategy_network_metric_random.png | Bin 0 -> 79004 bytes
...strategy_network_metric_yahoo_topologies.png | Bin 0 -> 75195 bytes
docs/images/ras_new_strategy_runtime_random.png | Bin 0 -> 76534 bytes
docs/images/ras_new_strategy_runtime_yahoo.png | Bin 0 -> 71727 bytes
8 files changed, 201 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/Resource_Aware_Scheduler_overview.md
----------------------------------------------------------------------
diff --git a/docs/Resource_Aware_Scheduler_overview.md b/docs/Resource_Aware_Scheduler_overview.md
index 4dc26f5..e3e2b56 100644
--- a/docs/Resource_Aware_Scheduler_overview.md
+++ b/docs/Resource_Aware_Scheduler_overview.md
@@ -3,23 +3,45 @@ title: Resource Aware Scheduler
layout: documentation
documentation: true
---
-# Introduction
-The purpose of this document is to provide a description of the Resource Aware Scheduler for the Storm distributed real-time computation system. This document will provide you with both a high level description of the resource aware scheduler in Storm
+# Introduction
+The purpose of this document is to provide a description of the Resource Aware Scheduler for the Storm distributed real-time computation system. This document will provide you with both a high level description of the resource aware scheduler in Storm. Some of the benefits are using a resource aware scheduler on top of Storm is outlined in the following presentation at Hadoop Summit 2016:
+
+http://www.slideshare.net/HadoopSummit/resource-aware-scheduling-in-apache-storm
+
+# Table of Contents
+1. [Using Resource Aware Scheduler](#Using-Resource-Aware-Scheduler)
+2. [API Overview](#API-Overview)
+ 1. [Setting Memory Requirement](#Setting-Memory-Requirement)
+ 2. [Setting CPU Requirement](#Setting-CPU-Requirement)
+ 3. [Limiting the Heap Size per Worker (JVM) Process](#Limiting-the-Heap-Size-per-Worker-(JVM)Process)
+ 4. [Setting Available Resources on Node](#Setting-Available-Resources-on-Node)
+ 5. [Other Configurations](#Other-Configurations)
+3. [Topology Priorities and Per User Resource](#Topology-Priorities-and-Per-User-Resource)
+ 1. [Setup](#Setup)
+ 2. [Specifying Topology Priority](#Specifying-Topology-Priority)
+ 3. [Specifying Scheduling Strategy](#Specifying-Scheduling-Strategy)
+ 4. [Specifying Topology Prioritization Strategy](#Specifying-Topology-Prioritization-Strategy)
+ 5. [Specifying Eviction Strategy](#Specifying-Eviction-Strategy)
+4. [Profiling Resource Usage](#Profiling-Resource-Usage)
+5. [Enhancements on original DefaultResourceAwareStrategy](#Enhancements-on-original-DefaultResourceAwareStrategy)
+
+<div id='Using-Resource-Aware-Scheduler'/>
## Using Resource Aware Scheduler
The user can switch to using the Resource Aware Scheduler by setting the following in *conf/storm.yaml*
storm.scheduler: \u201corg.apache.storm.scheduler.resource.ResourceAwareScheduler\u201d
-
-
+
+<div id='API-Overview'/>
## API Overview
For use with Trident, please see the [Trident RAS API](Trident-RAS-API.html)
For a Storm Topology, the user can now specify the amount of resources a topology component (i.e. Spout or Bolt) is required to run a single instance of the component. The user can specify the resource requirement for a topology component by using the following API calls.
+<div id='Setting-Memory-Requirement'/>
### Setting Memory Requirement
API to set component memory requirement:
@@ -48,9 +70,9 @@ Example of Usage:
The entire memory requested for this topology is 16.5 GB. That is from 10 spouts with 1GB on heap memory and 0.5 GB off heap memory each and 3 bolts with 0.5 GB on heap memory each.
+<div id='Setting-CPU-Requirement'/>
### Setting CPU Requirement
-
API to set component CPU requirement:
public T setCPULoad(Double amount)
@@ -71,9 +93,9 @@ Example of Usage:
builder.setBolt("exclaim2", new HeavyBolt(), 1)
.shuffleGrouping("exclaim1").setCPULoad(450.0);
+<div id='Limiting-the-Heap-Size-per-Worker-(JVM)Process'/>
### Limiting the Heap Size per Worker (JVM) Process
-
public void setTopologyWorkerMaxHeapSize(Number size)
Parameters:
@@ -85,7 +107,8 @@ Example of Usage:
Config conf = new Config();
conf.setTopologyWorkerMaxHeapSize(512.0);
-
+
+<div id='Setting-Available-Resources-on-Node'/>
### Setting Available Resources on Node
A storm administrator can specify node resource availability by modifying the *conf/storm.yaml* file located in the storm home directory of that node.
@@ -106,7 +129,7 @@ Example of Usage:
supervisor.memory.capacity.mb: 20480.0
supervisor.cpu.capacity: 100.0
-
+<div id='Other-Configurations'/>
### Other Configurations
The user can set some default configurations for the Resource Aware Scheduler in *conf/storm.yaml*:
@@ -123,11 +146,13 @@ The user can set some default configurations for the Resource Aware Scheduler in
//default value for the max heap size for a worker
topology.worker.max.heap.size.mb: 768.0
-# Topology Priorities and Per User Resource
+<div id='Topology-Priorities-and-Per-User-Resource'/>
+## Topology Priorities and Per User Resource
The Resource Aware Scheduler or RAS also has multitenant capabilities since many Storm users typically share a Storm cluster. Resource Aware Scheduler can allocate resources on a per user basis. Each user can be guaranteed a certain amount of resources to run his or her topologies and the Resource Aware Scheduler will meet those guarantees when possible. When the Storm cluster has extra free resources, Resource Aware Scheduler will to be able allocate additional resources to user in a fair manner. The importance of topologies can also vary. Topologies can be used for actual production or just experimentation, thus Resource Aware Scheduler will take into account the importance of a topology when determining the order in which to schedule topologies or when to evict topologies
-## Setup
+<div id='Setup'/>
+### Setup
The resource guarantees of a user can be specified *conf/user-resource-pools.yaml*. Specify the resource guarantees of a user in the following format:
@@ -151,7 +176,7 @@ An example of what *user-resource-pools.yaml* can look like:
Please note that the specified amount of Guaranteed CPU and Memory can be either a integer or double
-## API Overview
+<div id='Specifying-Topology-Priority'/>
### Specifying Topology Priority
The range of topology priorities can range form 0-29. The topologies priorities will be partitioned into several priority levels that may contain a range of priorities.
For example we can create a priority level mapping:
@@ -169,7 +194,8 @@ Parameters:
Please note that the 0-29 range is not a hard limit. Thus, a user can set a priority number that is higher than 29. However, the property of higher the priority number, lower the importance still holds
-### Specifying Scheduling Strategy:
+<div id='Specifying-Scheduling-Strategy'/>
+### Specifying Scheduling Strategy
A user can specify on a per topology basis what scheduling strategy to use. Users can implement the IStrategy interface and define new strategies to schedule specific topologies. This pluggable interface was created since we realize different topologies might have different scheduling needs. A user can set the topology strategy within the topology definition by using the API:
@@ -184,13 +210,144 @@ Example Usage:
A default scheduling is provided. The DefaultResourceAwareStrategy is implemented based off the scheduling algorithm in the original paper describing resource aware scheduling in Storm:
-http://web.engr.illinois.edu/~bpeng/files/r-storm.pdf
+Peng, Boyang, Mohammad Hosseini, Zhihao Hong, Reza Farivar, and Roy Campbell. "R-storm: Resource-aware scheduling in storm." In Proceedings of the 16th Annual Middleware Conference, pp. 149-161. ACM, 2015.
+
+http://dl.acm.org/citation.cfm?id=2814808
+
+**Please Note: Enhancements have to made on top of the original scheduling strategy as described in the paper. Please see section "Enhancements on original DefaultResourceAwareStrategy"**
+
+<div id='Specifying-Topology-Prioritization-Strategy'/>
+### Specifying Topology Prioritization Strategy
+
+The order of scheduling is a pluggable interface in which a user could define a strategy that prioritizes topologies. For a user to define his or her own prioritization strategy, he or she needs to implement the ISchedulingPriorityStrategy interface. A user can set the scheduling priority strategy by setting the *Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY* to point to the class that implements the strategy. For instance:
+
+ resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy"
+
+A default strategy will be provided. The following explains how the default scheduling priority strategy works.
+
+**DefaultSchedulingPriorityStrategy**
+
+The order of scheduling should be based on the distance between a user\u2019s current resource allocation and his or her guaranteed allocation. We should prioritize the users who are the furthest away from their resource guarantee. The difficulty of this problem is that a user may have multiple resource guarantees, and another user can have another set of resource guarantees, so how can we compare them in a fair manner? Let's use the average percentage of resource guarantees satisfied as a method of comparison.
+
+For example:
+
+|User|Resource Guarantee|Resource Allocated|
+|----|------------------|------------------|
+|A|<10 CPU, 50GB>|<2 CPU, 40 GB>|
+|B|< 20 CPU, 25GB>|<15 CPU, 10 GB>|
+
+User A\u2019s average percentage satisfied of resource guarantee:
+
+(2/10+40/50)/2 = 0.5
+
+User B\u2019s average percentage satisfied of resource guarantee:
+
+(15/20+10/25)/2 = 0.575
+
+Thus, in this example User A has a smaller average percentage of his or her resource guarantee satisfied than User B. Thus, User A should get priority to be allocated more resource, i.e., schedule a topology submitted by User A.
+
+When scheduling, RAS sorts users by the average percentage satisfied of resource guarantee and schedule topologies from users based on that ordering starting from the users with the lowest average percentage satisfied of resource guarantee. When a user\u2019s resource guarantee is completely satisfied, the user\u2019s average percentage satisfied of resource guarantee will be greater than or equal to 1.
+
+<div id='Specifying-Eviction-Strategy'/>
+### Specifying Eviction Strategy
+The eviction strategy is used when there are not enough free resources in the cluster to schedule new topologies. If the cluster is full, we need a mechanism to evict topologies so that user resource guarantees can be met and additional resource can be shared fairly among users. The strategy for evicting topologies is also a pluggable interface in which the user can implement his or her own topology eviction strategy. For a user to implement his or her own eviction strategy, he or she needs to implement the IEvictionStrategy Interface and set *Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY* to point to the implemented strategy class. For instance:
+
+ resource.aware.scheduler.eviction.strategy: "org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy"
+
+A default eviction strategy is provided. The following explains how the default topology eviction strategy works
+
+**DefaultEvictionStrategy**
+
+To determine if topology eviction should occur we should take into account the priority of the topology that we are trying to schedule and whether the resource guarantees for the owner of the topology have been met.
+
+We should never evict a topology from a user that does not have his or her resource guarantees satisfied. The following flow chart should describe the logic for the eviction process.
+
+
+
+<div id='Profiling-Resource-Usage'/>
+## Profiling Resource Usage
+
+Figuring out resource usage for your topology:
+
+To get an idea of how much memory/CPU your topology is actually using you can add the following to your topology launch code.
+
+ //Log all storm metrics
+ conf.registerMetricsConsumer(backtype.storm.metric.LoggingMetricsConsumer.class);
+
+ //Add in per worker CPU measurement
+ Map<String, String> workerMetrics = new HashMap<String, String>();
+ workerMetrics.put("CPU", "org.apache.storm.metrics.sigar.CPUMetric");
+ conf.put(Config.TOPOLOGY_WORKER_METRICS, workerMetrics);
+
+The CPU metrics will require you to add
+
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-metrics</artifactId>
+ <version>1.0.0</version>
+ </dependency>
+
+as a topology dependency (1.0.0 or higher).
+
+You can then go to your topology on the UI, turn on the system metrics, and find the log that the LoggingMetricsConsumer is writing to. It will output results in the log like.
+
+ 1454526100 node1.nodes.com:6707 -1:__system CPU {user-ms=74480, sys-ms=10780}
+ 1454526100 node1.nodes.com:6707 -1:__system memory/nonHeap {unusedBytes=2077536, virtualFreeBytes=-64621729, initBytes=2555904, committedBytes=66699264, maxBytes=-1, usedBytes=64621728}
+ 1454526100 node1.nodes.com:6707 -1:__system memory/heap {unusedBytes=573861408, virtualFreeBytes=694644256, initBytes=805306368, committedBytes=657719296, maxBytes=778502144, usedBytes=83857888}
+
+The metrics with -1:__system are generally metrics for the entire worker. In the example above that worker is running on node1.nodes.com:6707. These metrics are collected every 60 seconds. For the CPU you can see that over the 60 seconds this worker used 74480 + 10780 = 85260 ms of CPU time. This is equivalent to 85260/60000 or about 1.5 cores.
+
+The Memory usage is similar but look at the usedBytes. offHeap is 64621728 or about 62MB, and onHeap is 83857888 or about 80MB, but you should know what you set your heap to in each of your workers already. How do you divide this up per bolt/spout? That is a bit harder and may require some trial and error from your end.
+
+<div id='Enhancements-on-original-DefaultResourceAwareStrategy'/>
+## * Enhancements on original DefaultResourceAwareStrategy *
+
+The default resource aware scheduling strategy as described in the paper above has two main scheduling phases:
+
+1. Task Selection - Calculate the order task/executors in a topology should be scheduled
+2. Node Selection - Given a task/executor, find a node to schedule the task/executor on.
+
+Enhancements have been made for both scheduling phases
+
+### Task Selection Enhancements
+
+Instead of using a breadth first traversal of the topology graph to create a ordering of components and its executors, a new heuristic is used that orders components by the number of in and out edges (potential connections) of the component. This is discovered to be a more effective way to colocate executors that communicate with each other and reduce the network latency.
+
+
+### Node Selection Enhancements
+
+Node selection comes down first selecting which rack (server rack) and then which node on that rack to choose. The gist of strategy in choosing a rack and node is finding the rack that has the "most" resource available and in that rack find the node with the "most" free resources. The assumption we are making for this strategy is that the node or rack with the most free resources will have the highest probability that allows us to schedule colocate the most number of executors on the node or rack to reduce network communication latency
+
+Racks and nodes will be sorted from best choice to worst choice. When finding an executor, the strategy will iterate through all racks and nodes, starting from best to worst, before giving up. Racks and nodes will be sorted in the following matter:
-#### * Enhancements on original DefaultResourceAwareStrategy *
+1. How many executors are already scheduled on the rack or node
+ -- This is done so we move executors to schedule closer to executors that are already scheduled and running. If a topology partially crashed and a subset of the topology's executors need to be rescheduled, we want to reschedule these executors as close (network wise) as possible to the executors that healthy and running.
-Originally the getBestClustering algorithm for RAS finds the "Best" cluster/rack based on which rack has the "most available" resources by finding the rack with the biggest sum of available memory + available across all nodes in the rack. This method is not very accurate since memory and cpu usage aree values on a different scale and the values are not normailized. This method is also not effective since it does not consider the number of slots available and it fails to identifying racks that are not schedulable due to the exhaustion of one of the resources either memory, cpu, or slots. Also the current method does not consider failures of workers. When executors of a topology gets unassigned and needs to be scheduled again, the current logic in getBestClustering may be inadequate since it will likely return a cluster that is different from where the majority of executors from the topology is originally scheduling in.
+2. Subordinate resource availability or the amount "effective" resources on the rack or node
+ -- Please refer the section on Subordinate Resource Availability
-The new strategy/algorithm to find the "best" cluster, I dub subordinate resource availability ordering (inspired by Dominant Resource Fairness), sorts racks by the subordinate (not dominant) resource availability.
+3. Average of the all the resource availability
+ -- This is simply taking the average of the percent available (available resources on node or rack divied by theavailable resources on rack or cluster, repectively). This situation will only be used when "effective resources" for two objects (rack or node) are the same. Then we consider the average of all the percentages of resources as a metric for sorting. For example:
+
+ Avail Resources:
+ node 1: CPU = 50 Memory = 1024 Slots = 20
+ node 2: CPU = 50 Memory = 8192 Slots = 40
+ node 3: CPU = 1000 Memory = 0 Slots = 0
+
+ Effective resources for nodes:
+ node 1 = 50 / (50+50+1000) = 0.045 (CPU bound)
+ node 2 = 50 / (50+50+1000) = 0.045 (CPU bound)
+ node 3 = 0 (memory and slots are 0)
+
+ode 1 and node 2 have the same effective resources but clearly node 2 has more resources (memory and slots) than node 1 and we would want to pick node 2 first since there is a higher probability we will be able to schedule more executors on it. This is what the phase 2 averaging does
+
+Thus the sorting follows the following progression. Compare based on 1) and if equal then compare based on 2) and if equal compare based on 3) and if equal we break ties by arbitrarly assigning ordering based on comparing the ids of the node or rack.
+
+**Subordinate Resource Availability**
+
+Originally the getBestClustering algorithm for RAS finds the "Best" rack based on which rack has the "most available" resources by finding the rack with the biggest sum of available memory + available across all nodes in the rack. This method is not very accurate since memory and cpu usage aree values on a different scale and the values are not normailized. This method is also not effective since it does not consider the number of slots available and it fails to identifying racks that are not schedulable due to the exhaustion of one of the resources either memory, cpu, or slots. Also the previous method does not consider failures of workers. When executors of a topology gets unassigned and needs to be scheduled again, the current logic in getBestClustering may be inadequate since it will likely return a cluster that is different from where the majority of executors from the topology is originally scheduling in.
+
+The new strategy/algorithm to find the "best" rack or node, I dub subordinate resource availability ordering (inspired by Dominant Resource Fairness), sorts racks and nodes by the subordinate (not dominant) resource availability.
For example given 4 racks with the following resource availabilities
@@ -231,56 +388,48 @@ Then we order the racks by the effective resource.
Thus for our example:
Sorted rack: [rack-0, rack-1, rack-4, rack-3, rack-2]
-
-Also to deal with the presence of failures, if a topology is partially scheduled, we find the rack with the most scheduled executors for the topology and we try to schedule on that rack first.
-
-Thus we first sort by the number of executors already scheduled on the rack and then by the subordinate resource availability.
-
-Original Jira for this enhancement: [STORM-1766](https://issues.apache.org/jira/browse/STORM-1766)
-
-### Specifying Topology Prioritization Strategy
-
-The order of scheduling is a pluggable interface in which a user could define a strategy that prioritizes topologies. For a user to define his or her own prioritization strategy, he or she needs to implement the ISchedulingPriorityStrategy interface. A user can set the scheduling priority strategy by setting the *Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY* to point to the class that implements the strategy. For instance:
-
- resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy"
-A default strategy will be provided. The following explains how the default scheduling priority strategy works.
-
-**DefaultSchedulingPriorityStrategy**
-
-The order of scheduling should be based on the distance between a user\u2019s current resource allocation and his or her guaranteed allocation. We should prioritize the users who are the furthest away from their resource guarantee. The difficulty of this problem is that a user may have multiple resource guarantees, and another user can have another set of resource guarantees, so how can we compare them in a fair manner? Let's use the average percentage of resource guarantees satisfied as a method of comparison.
+This metric is used in sorting for both nodes and racks. When sorting racks, we consider resources available on the rack and in the whole cluster (containing all racks). When sorting nodes, we consider resources available on the node and the resources available in the rack (sum of all resources available for all nodes in rack)
-For example:
+Original Jira for this enhancement: [STORM-1766](https://issues.apache.org/jira/browse/STORM-1766)
-|User|Resource Guarantee|Resource Allocated|
-|----|------------------|------------------|
-|A|<10 CPU, 50GB>|<2 CPU, 40 GB>|
-|B|< 20 CPU, 25GB>|<15 CPU, 10 GB>|
+### Improvements in Scheduling
+This section provides some experimental results on the performance benefits with the enhancements on top of the original scheduling strategy. The experiments are based off of running simulations using:
-User A\u2019s average percentage satisfied of resource guarantee:
+https://github.com/jerrypeng/storm-scheduler-test-framework
-(2/10+40/50)/2 = 0.5
+Random topologies and clusters are used in the simulation as well as a comprehensive dataset consisting of all real topologies running in all the storm clusters at Yahoo.
-User B\u2019s average percentage satisfied of resource guarantee:
+The below graphs provides a comparison of how well the various strategies schedule topologies to minimize network latency. A network metric is calculated for each scheduling of a topology by each scheduling strategy. The network metric is calculated based on how many connections each executor in a topology has to make to another executor residing in the same worker (JVM process), in different worker but same host, different host, different rack. The assumption we are making is the following
-(15/20+10/25)/2 = 0.575
+1. Intra-worker communication is the fastest
+2. Inter-worker communication is fast
+3. Inter-node communication is slower
+4. Inter-rack communication is the slowest
-Thus, in this example User A has a smaller average percentage of his or her resource guarantee satisfied than User B. Thus, User A should get priority to be allocated more resource, i.e., schedule a topology submitted by User A.
+For this network metric, the larger the number is number is the more potential network latency the topology will have for this scheduling. Two types of experiments are performed. One set experiments are performed with randomly generated topologies and randomly generate clusters. The other set of experiments are performed with a dataset containing all of the running topologies at yahoo and semi-randomly generated clusters based on the size of the topology. Both set of experiments are run millions of iterations until results converge.
-When scheduling, RAS sorts users by the average percentage satisfied of resource guarantee and schedule topologies from users based on that ordering starting from the users with the lowest average percentage satisfied of resource guarantee. When a user\u2019s resource guarantee is completely satisfied, the user\u2019s average percentage satisfied of resource guarantee will be greater than or equal to 1.
+For the experiments involving randomly generated topologies, an optimal strategy is implemented that exhausively finds the most optimal solution if a solution exists. The topologies and clusters used in this experiment are relatively small so that the optimal strategy traverse to solution space to find a optimal solution in a reasonable amount of time. This strategy is not run with the Yahoo topologies since the topologies are large and would take unreasonable amount of time to run, since the solutions space is W^N (ordering doesn't matter within a worker) where W is the number of workers and N is the number of executors. The NextGenStrategy represents the scheduling strategy with these enhancements. The DefaultResourceAwareStrategy represents the original scheduling strategy. The RoundRobinStrategy represents a naive strategy that simply schedules executors in a round robin fashion while respecting the resource constraints. The graph below presents averages of the network metr
ic. A CDF graph is also presented further down.
-### Specifying Eviction Strategy
-The eviction strategy is used when there are not enough free resources in the cluster to schedule new topologies. If the cluster is full, we need a mechanism to evict topologies so that user resource guarantees can be met and additional resource can be shared fairly among users. The strategy for evicting topologies is also a pluggable interface in which the user can implement his or her own topology eviction strategy. For a user to implement his or her own eviction strategy, he or she needs to implement the IEvictionStrategy Interface and set *Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY* to point to the implemented strategy class. For instance:
+| Random Topologies | Yahoo topologies |
+|-------------------|------------------|
+|| |
- resource.aware.scheduler.eviction.strategy: "org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy"
+The next graph displays how close the schedulings from the respectively scheduling strategies are to the schedulings of the optimal strategy. As explained earlier, this is only done for the random generated topologies and clusters.
-A default eviction strategy is provided. The following explains how the default topology eviction strategy works
+| Random Topologies |
+|-------------------|
+||
-**DefaultEvictionStrategy**
+The below graph is a CDF of the network metric:
+| Random Topologies | Yahoo topologies |
+|-------------------|------------------|
+|| |
-To determine if topology eviction should occur we should take into account the priority of the topology that we are trying to schedule and whether the resource guarantees for the owner of the topology have been met.
+Below is a comparison of the how long the strategies take to run:
-We should never evict a topology from a user that does not have his or her resource guarantees satisfied. The following flow chart should describe the logic for the eviction process.
+| Random Topologies | Yahoo topologies |
+|-------------------|------------------|
+|| |
-
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_network_cdf_random.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_network_cdf_random.png b/docs/images/ras_new_strategy_network_cdf_random.png
new file mode 100644
index 0000000..8b47f36
Binary files /dev/null and b/docs/images/ras_new_strategy_network_cdf_random.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_network_metric_cdf_yahoo_topologies.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_network_metric_cdf_yahoo_topologies.png b/docs/images/ras_new_strategy_network_metric_cdf_yahoo_topologies.png
new file mode 100644
index 0000000..6e5a04d
Binary files /dev/null and b/docs/images/ras_new_strategy_network_metric_cdf_yahoo_topologies.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_network_metric_improvement_random.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_network_metric_improvement_random.png b/docs/images/ras_new_strategy_network_metric_improvement_random.png
new file mode 100644
index 0000000..3fd4029
Binary files /dev/null and b/docs/images/ras_new_strategy_network_metric_improvement_random.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_network_metric_random.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_network_metric_random.png b/docs/images/ras_new_strategy_network_metric_random.png
new file mode 100644
index 0000000..11d95d9
Binary files /dev/null and b/docs/images/ras_new_strategy_network_metric_random.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_network_metric_yahoo_topologies.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_network_metric_yahoo_topologies.png b/docs/images/ras_new_strategy_network_metric_yahoo_topologies.png
new file mode 100644
index 0000000..60ed7c1
Binary files /dev/null and b/docs/images/ras_new_strategy_network_metric_yahoo_topologies.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_runtime_random.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_runtime_random.png b/docs/images/ras_new_strategy_runtime_random.png
new file mode 100644
index 0000000..0ba1c73
Binary files /dev/null and b/docs/images/ras_new_strategy_runtime_random.png differ
http://git-wip-us.apache.org/repos/asf/storm/blob/667fc5e1/docs/images/ras_new_strategy_runtime_yahoo.png
----------------------------------------------------------------------
diff --git a/docs/images/ras_new_strategy_runtime_yahoo.png b/docs/images/ras_new_strategy_runtime_yahoo.png
new file mode 100644
index 0000000..0a51089
Binary files /dev/null and b/docs/images/ras_new_strategy_runtime_yahoo.png differ
[2/6] storm git commit: re-adding accidentally deleted RAS tests
Posted by je...@apache.org.
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c6c44/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java b/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
index fbc0421..bc69725 100644
--- a/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
+++ b/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestUtilsForResourceAwareScheduler.java
@@ -22,9 +22,11 @@ import org.apache.storm.Config;
import org.apache.storm.generated.Bolt;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
+import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
import org.apache.storm.scheduler.IScheduler;
+import org.apache.storm.scheduler.SchedulerAssignment;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
@@ -45,6 +47,7 @@ import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
@@ -112,12 +115,11 @@ public class TestUtilsForResourceAwareScheduler {
public static Map<ExecutorDetails, String> genExecsAndComps(StormTopology topology) {
Map<ExecutorDetails, String> retMap = new HashMap<ExecutorDetails, String>();
int startTask = 0;
- int endTask = 1;
+ int endTask = 0;
for (Map.Entry<String, SpoutSpec> entry : topology.get_spouts().entrySet()) {
SpoutSpec spout = entry.getValue();
String spoutId = entry.getKey();
int spoutParallelism = spout.get_common().get_parallelism_hint();
-
for (int i = 0; i < spoutParallelism; i++) {
retMap.put(new ExecutorDetails(startTask, endTask), spoutId);
startTask++;
@@ -296,4 +298,72 @@ public class TestUtilsForResourceAwareScheduler {
}
return ret;
}
+
+ public static Map<SupervisorDetails, Double> getSupervisorToMemoryUsage(Cluster cluster, Topologies topologies) {
+ Map<SupervisorDetails, Double> superToMem = new HashMap<>();
+ Collection<SchedulerAssignment> assignments = cluster.getAssignments().values();
+ Collection<SupervisorDetails> supervisors = cluster.getSupervisors().values();
+ for (SupervisorDetails supervisor : supervisors) {
+ superToMem.put(supervisor, 0.0);
+ }
+
+ for (SchedulerAssignment assignment : assignments) {
+ Map<ExecutorDetails, SupervisorDetails> executorToSupervisor = new HashMap<>();
+ Map<SupervisorDetails, List<ExecutorDetails>> supervisorToExecutors = new HashMap<>();
+ TopologyDetails topology = topologies.getById(assignment.getTopologyId());
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : assignment.getExecutorToSlot().entrySet()) {
+ executorToSupervisor.put(entry.getKey(), cluster.getSupervisorById(entry.getValue().getNodeId()));
+ }
+ for (Map.Entry<ExecutorDetails, SupervisorDetails> entry : executorToSupervisor.entrySet()) {
+ List<ExecutorDetails> executorsOnSupervisor = supervisorToExecutors.get(entry.getValue());
+ if (executorsOnSupervisor == null) {
+ executorsOnSupervisor = new ArrayList<>();
+ supervisorToExecutors.put(entry.getValue(), executorsOnSupervisor);
+ }
+ executorsOnSupervisor.add(entry.getKey());
+ }
+ for (Map.Entry<SupervisorDetails, List<ExecutorDetails>> entry : supervisorToExecutors.entrySet()) {
+ Double supervisorUsedMemory = 0.0;
+ for (ExecutorDetails executor: entry.getValue()) {
+ supervisorUsedMemory += topology.getTotalMemReqTask(executor);
+ }
+ superToMem.put(entry.getKey(), superToMem.get(entry.getKey()) + supervisorUsedMemory);
+ }
+ }
+ return superToMem;
+ }
+
+ public static Map<SupervisorDetails, Double> getSupervisorToCpuUsage(Cluster cluster, Topologies topologies) {
+ Map<SupervisorDetails, Double> superToCpu = new HashMap<>();
+ Collection<SchedulerAssignment> assignments = cluster.getAssignments().values();
+ Collection<SupervisorDetails> supervisors = cluster.getSupervisors().values();
+ for (SupervisorDetails supervisor : supervisors) {
+ superToCpu.put(supervisor, 0.0);
+ }
+
+ for (SchedulerAssignment assignment : assignments) {
+ Map<ExecutorDetails, SupervisorDetails> executorToSupervisor = new HashMap<>();
+ Map<SupervisorDetails, List<ExecutorDetails>> supervisorToExecutors = new HashMap<>();
+ TopologyDetails topology = topologies.getById(assignment.getTopologyId());
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : assignment.getExecutorToSlot().entrySet()) {
+ executorToSupervisor.put(entry.getKey(), cluster.getSupervisorById(entry.getValue().getNodeId()));
+ }
+ for (Map.Entry<ExecutorDetails, SupervisorDetails> entry : executorToSupervisor.entrySet()) {
+ List<ExecutorDetails> executorsOnSupervisor = supervisorToExecutors.get(entry.getValue());
+ if (executorsOnSupervisor == null) {
+ executorsOnSupervisor = new ArrayList<>();
+ supervisorToExecutors.put(entry.getValue(), executorsOnSupervisor);
+ }
+ executorsOnSupervisor.add(entry.getKey());
+ }
+ for (Map.Entry<SupervisorDetails, List<ExecutorDetails>> entry : supervisorToExecutors.entrySet()) {
+ Double supervisorUsedCpu = 0.0;
+ for (ExecutorDetails executor: entry.getValue()) {
+ supervisorUsedCpu += topology.getTotalCpuReqTask(executor);
+ }
+ superToCpu.put(entry.getKey(), superToCpu.get(entry.getKey()) + supervisorUsedCpu);
+ }
+ }
+ return superToCpu;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c6c44/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
new file mode 100644
index 0000000..1d199b1
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
@@ -0,0 +1,736 @@
+package org.apache.storm.scheduler.resource.strategies.eviction;
+
+import org.apache.storm.Config;
+import org.apache.storm.scheduler.Cluster;
+import org.apache.storm.scheduler.INimbus;
+import org.apache.storm.scheduler.SchedulerAssignmentImpl;
+import org.apache.storm.scheduler.SupervisorDetails;
+import org.apache.storm.scheduler.Topologies;
+import org.apache.storm.scheduler.TopologyDetails;
+import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
+import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
+import org.apache.storm.utils.Utils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Created by jerrypeng on 10/6/16.
+ */
+public class TestDefaultEvictionStrategy {
+
+ private static int currentTime = 1450418597;
+
+ /**
+ * The resources in the cluster are limited. In the first round of scheduling, all resources in the cluster is used.
+ * User jerry submits another toploogy. Since user jerry has his resource guarantees satisfied, and user bobby
+ * has exceeded his resource guarantee, topo-3 from user bobby should be evicted.
+ */
+ @Test
+ public void testEviction() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<String, Number>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
+ Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
+ resourceUserPool.put("jerry", new HashMap<String, Number>());
+ resourceUserPool.get("jerry").put("cpu", 200.0);
+ resourceUserPool.get("jerry").put("memory", 2000.0);
+
+ resourceUserPool.put("bobby", new HashMap<String, Number>());
+ resourceUserPool.get("bobby").put("cpu", 100.0);
+ resourceUserPool.get("bobby").put("memory", 1000.0);
+
+ resourceUserPool.put("derek", new HashMap<String, Number>());
+ resourceUserPool.get("derek").put("cpu", 200.0);
+ resourceUserPool.get("derek").put("memory", 2000.0);
+
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
+
+ TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 20);
+
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
+
+ TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
+
+ TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
+
+ Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
+ topoMap.put(topo1.getId(), topo1);
+ topoMap.put(topo2.getId(), topo2);
+ topoMap.put(topo3.getId(), topo3);
+ topoMap.put(topo4.getId(), topo4);
+
+ Topologies topologies = new Topologies(topoMap);
+
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ //user jerry submits another topology
+ topoMap.put(topo6.getId(), topo6);
+ topologies = new Topologies(topoMap);
+
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
+ Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("correct topology to evict", "topo-3", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName());
+ }
+
+ @Test
+ public void TestEvictMultipleTopologies() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<String, Number>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
+ Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
+ resourceUserPool.put("jerry", new HashMap<String, Number>());
+ resourceUserPool.get("jerry").put("cpu", 200.0);
+ resourceUserPool.get("jerry").put("memory", 2000.0);
+
+ resourceUserPool.put("derek", new HashMap<String, Number>());
+ resourceUserPool.get("derek").put("cpu", 100.0);
+ resourceUserPool.get("derek").put("memory", 1000.0);
+
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
+
+ TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 2, 0, 1, 0, currentTime - 2, 10);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
+
+ TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
+
+ TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
+ TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29);
+
+ Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
+ topoMap.put(topo2.getId(), topo2);
+ topoMap.put(topo3.getId(), topo3);
+ topoMap.put(topo4.getId(), topo4);
+ topoMap.put(topo5.getId(), topo5);
+
+ Topologies topologies = new Topologies(topoMap);
+
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ //user jerry submits another topology
+ topoMap.put(topo1.getId(), topo1);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
+ Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
+ }
+
+ /**
+ * Eviction order:
+ * topo-3: since user bobby don't have any resource guarantees and topo-3 is the lowest priority for user bobby
+ * topo-2: since user bobby don't have any resource guarantees and topo-2 is the next lowest priority for user bobby
+ * topo-5: since user derek has exceeded his resource guarantee while user jerry has not. topo-5 and topo-4 has the same priority
+ * but topo-4 was submitted earlier thus we choose that one to evict
+ */
+ @Test
+ public void TestEvictMultipleTopologiesFromMultipleUsersInCorrectOrder() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<String, Number>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
+ Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
+ resourceUserPool.put("jerry", new HashMap<String, Number>());
+ resourceUserPool.get("jerry").put("cpu", 300.0);
+ resourceUserPool.get("jerry").put("memory", 3000.0);
+
+ resourceUserPool.put("derek", new HashMap<String, Number>());
+ resourceUserPool.get("derek").put("cpu", 100.0);
+ resourceUserPool.get("derek").put("memory", 1000.0);
+
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
+
+ TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo7 = TestUtilsForResourceAwareScheduler.getTopology("topo-7", config, 1, 0, 1, 0, currentTime - 2, 10);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
+
+ TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
+
+ TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
+ TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 15, 29);
+
+ Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
+ topoMap.put(topo2.getId(), topo2);
+ topoMap.put(topo3.getId(), topo3);
+ topoMap.put(topo4.getId(), topo4);
+ topoMap.put(topo5.getId(), topo5);
+
+ Topologies topologies = new Topologies(topoMap);
+
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ //user jerry submits another topology
+ topoMap.put(topo1.getId(), topo1);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
+ Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("correct topology to evict", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName(), "topo-3");
+
+ topoMap.put(topo6.getId(), topo6);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
+ Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
+
+ Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
+ Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
+
+ topoMap.put(topo7.getId(), topo7);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 3, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesAttempted()) {
+ Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 1, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+ Assert.assertEquals("correct topology to evict", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName(), "topo-4");
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
+ Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
+
+ Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
+ Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
+ }
+
+ /**
+ * If topologies from other users cannot be evicted to make space
+ * check if there is a topology with lower priority that can be evicted from the current user
+ */
+ @Test
+ public void TestEvictTopologyFromItself() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<String, Number>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
+ Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
+ resourceUserPool.put("jerry", new HashMap<String, Number>());
+ resourceUserPool.get("jerry").put("cpu", 200.0);
+ resourceUserPool.get("jerry").put("memory", 2000.0);
+
+ resourceUserPool.put("bobby", new HashMap<String, Number>());
+ resourceUserPool.get("bobby").put("cpu", 100.0);
+ resourceUserPool.get("bobby").put("memory", 1000.0);
+
+ resourceUserPool.put("derek", new HashMap<String, Number>());
+ resourceUserPool.get("derek").put("cpu", 100.0);
+ resourceUserPool.get("derek").put("memory", 1000.0);
+
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
+
+ TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20);
+ TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20);
+ TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 29);
+ TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
+
+ TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 10);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
+
+ TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 29);
+
+ Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
+ topoMap.put(topo1.getId(), topo1);
+ topoMap.put(topo2.getId(), topo2);
+ topoMap.put(topo5.getId(), topo5);
+ topoMap.put(topo6.getId(), topo6);
+
+ Topologies topologies = new Topologies(topoMap);
+
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ //user jerry submits another topology into a full cluster
+ // topo3 should not be able to scheduled
+ topoMap.put(topo3.getId(), topo3);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 1, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+ //make sure that topo-3 didn't get scheduled.
+ Assert.assertEquals("correct topology in attempted queue", rs.getUser("jerry").getTopologiesAttempted().iterator().next().getName(), "topo-3");
+
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ //user jerry submits another topology but this one should be scheduled since it has higher priority than than the
+ //rest of jerry's running topologies
+ topoMap.put(topo4.getId(), topo4);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 2, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+ Assert.assertTrue("correct topology in attempted queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("jerry").getTopologiesAttempted()) != null);
+ //Either topo-1 or topo-2 should have gotten evicted
+ Assert.assertTrue("correct topology in attempted queue", ((TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-1", rs.getUser("jerry").getTopologiesAttempted())) != null)
+ || (TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("jerry").getTopologiesAttempted()) != null));
+ //assert that topo-4 got scheduled
+ Assert.assertTrue("correct topology in running queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-4", rs.getUser("jerry").getTopologiesRunning()) != null);
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ }
+
+ /**
+ * If users are above his or her guarantee, check if topology eviction works correct
+ */
+ @Test
+ public void TestOverGuaranteeEviction() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<String, Number>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
+ Config config = new Config();
+ config.putAll(Utils.readDefaultConfig());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
+ config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
+ Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
+ resourceUserPool.put("jerry", new HashMap<String, Number>());
+ resourceUserPool.get("jerry").put("cpu", 70.0);
+ resourceUserPool.get("jerry").put("memory", 700.0);
+
+ resourceUserPool.put("bobby", new HashMap<String, Number>());
+ resourceUserPool.get("bobby").put("cpu", 100.0);
+ resourceUserPool.get("bobby").put("memory", 1000.0);
+
+ resourceUserPool.put("derek", new HashMap<String, Number>());
+ resourceUserPool.get("derek").put("cpu", 25.0);
+ resourceUserPool.get("derek").put("memory", 250.0);
+
+ config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
+
+ TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20);
+ TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
+
+ TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 10);
+ TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10);
+
+ config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
+
+ TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29);
+ TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10);
+
+ Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
+ topoMap.put(topo1.getId(), topo1);
+ topoMap.put(topo3.getId(), topo3);
+ topoMap.put(topo4.getId(), topo4);
+ topoMap.put(topo5.getId(), topo5);
+
+ Topologies topologies = new Topologies(topoMap);
+
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+
+ //user derek submits another topology into a full cluster
+ //topo6 should not be able to scheduled intially, but since topo6 has higher priority than topo5
+ //topo5 will be evicted so that topo6 can be scheduled
+ topoMap.put(topo6.getId(), topo6);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 1, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+ //topo5 will be evicted since topo6 has higher priority
+ Assert.assertEquals("correct topology in attempted queue", "topo-5", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+
+ //user jerry submits topo2
+ topoMap.put(topo2.getId(), topo2);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster);
+
+ for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
+
+ for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 0, rs.getUser("derek").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 2, rs.getUser("derek").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
+ Assert.assertEquals("correct topology in attempted queue", "topo-6", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName());
+
+ for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
+ Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
+ }
+ Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
+ Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
+ Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
+ Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
+ }
+}
[4/6] storm git commit: re-adding accidentally deleted RAS tests
Posted by je...@apache.org.
re-adding accidentally deleted RAS tests
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5e4c6c44
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5e4c6c44
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5e4c6c44
Branch: refs/heads/1.x-branch
Commit: 5e4c6c44d456bf60258835424e1a16192ef8658b
Parents: 0d755ee
Author: Boyang Jerry Peng <je...@yahoo-inc.com>
Authored: Thu Oct 6 14:25:51 2016 -0500
Committer: Boyang Jerry Peng <je...@yahoo-inc.com>
Committed: Thu Oct 13 15:25:52 2016 -0500
----------------------------------------------------------------------
.../resource/TestResourceAwareScheduler.java | 1394 +++++++++---------
.../TestUtilsForResourceAwareScheduler.java | 74 +-
.../eviction/TestDefaultEvictionStrategy.java | 736 +++++++++
3 files changed, 1491 insertions(+), 713 deletions(-)
----------------------------------------------------------------------
[6/6] storm git commit: addressing comments
Posted by je...@apache.org.
addressing comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/20b612c2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/20b612c2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/20b612c2
Branch: refs/heads/1.x-branch
Commit: 20b612c25f28057a26f0355acc289bcef3543333
Parents: 667fc5e
Author: Boyang Jerry Peng <je...@yahoo-inc.com>
Authored: Tue Oct 11 10:16:00 2016 -0500
Committer: Boyang Jerry Peng <je...@yahoo-inc.com>
Committed: Thu Oct 13 15:26:14 2016 -0500
----------------------------------------------------------------------
.../storm/scheduler/resource/RAS_Node.java | 12 +++++++++++
.../eviction/TestDefaultEvictionStrategy.java | 21 +++++++++++++++++---
2 files changed, 30 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/20b612c2/storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java b/storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
index 79502d7..0bb36e8 100644
--- a/storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
+++ b/storm-core/src/jvm/org/apache/storm/scheduler/resource/RAS_Node.java
@@ -514,4 +514,16 @@ public class RAS_Node {
consumeCPU(taskCpuReq);
consumeMemory(taskMemReq);
}
+
+ /**
+ * frees the amount of resources for a executor in a topology.
+ * @param exec is the executor for which the resources are freed for
+ * @param topo the topology the executor is a part
+ */
+ public void freeResourcesForTask(ExecutorDetails exec, TopologyDetails topo) {
+ Double taskMemReq = topo.getTotalMemReqTask(exec);
+ Double taskCpuReq = topo.getTotalCpuReqTask(exec);
+ freeCPU(taskCpuReq);
+ freeMemory(taskMemReq);
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/20b612c2/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
index 1d199b1..f2d65bd 100644
--- a/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
+++ b/storm-core/test/jvm/org/apache/storm/scheduler/resource/strategies/eviction/TestDefaultEvictionStrategy.java
@@ -1,3 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.storm.scheduler.resource.strategies.eviction;
import org.apache.storm.Config;
@@ -16,9 +34,6 @@ import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
-/**
- * Created by jerrypeng on 10/6/16.
- */
public class TestDefaultEvictionStrategy {
private static int currentTime = 1450418597;
[3/6] storm git commit: re-adding accidentally deleted RAS tests
Posted by je...@apache.org.
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c6c44/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java b/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
index 054ecc2..44f2136 100644
--- a/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
+++ b/storm-core/test/jvm/org/apache/storm/scheduler/resource/TestResourceAwareScheduler.java
@@ -19,6 +19,7 @@
package org.apache.storm.scheduler.resource;
import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
@@ -29,6 +30,8 @@ import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
+import org.apache.storm.testing.TestWordCounter;
+import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.topology.BoltDeclarer;
import org.apache.storm.topology.SpoutDeclarer;
import org.apache.storm.topology.TopologyBuilder;
@@ -36,14 +39,21 @@ import org.apache.storm.utils.Utils;
import org.apache.storm.validation.ConfigValidation;
import org.junit.Assert;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Collection;
+import java.util.Collections;
import static org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler.genExecsAndComps;
@@ -55,6 +65,679 @@ public class TestResourceAwareScheduler {
private static int currentTime = 1450418597;
+ private static final Config defaultTopologyConf = new Config();
+
+ @BeforeClass
+ public static void initConf() {
+ defaultTopologyConf.put(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN, "org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping");
+ defaultTopologyConf.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
+ defaultTopologyConf.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+
+ defaultTopologyConf.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ defaultTopologyConf.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10.0);
+ defaultTopologyConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 128.0);
+ defaultTopologyConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 0.0);
+ defaultTopologyConf.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 8192.0);
+ defaultTopologyConf.put(Config.TOPOLOGY_PRIORITY, 0);
+ defaultTopologyConf.put(Config.TOPOLOGY_SUBMITTER_USER, "zhuo");
+ }
+
+ @Test
+ public void testRASNodeSlotAssign() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(5, 4, resourceMap);
+ Topologies topologies = new Topologies(new HashMap<String, TopologyDetails>());
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), new HashMap());
+ Map<String, RAS_Node> nodes = RAS_Nodes.getAllNodesFrom(cluster, topologies);
+ Assert.assertEquals(5, nodes.size());
+ RAS_Node node = nodes.get("sup-0");
+
+ Assert.assertEquals("sup-0", node.getId());
+ Assert.assertTrue(node.isAlive());
+ Assert.assertEquals(0, node.getRunningTopologies().size());
+ Assert.assertTrue(node.isTotallyFree());
+ Assert.assertEquals(4, node.totalSlotsFree());
+ Assert.assertEquals(0, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+
+ TopologyDetails topology1 = TestUtilsForResourceAwareScheduler.getTopology("topology1", new HashMap(), 1, 0, 2, 0, 0, 0);
+
+ List<ExecutorDetails> executors11 = new ArrayList<>();
+ executors11.add(new ExecutorDetails(1, 1));
+ node.assign(node.getFreeSlots().iterator().next(), topology1, executors11);
+ Assert.assertEquals(1, node.getRunningTopologies().size());
+ Assert.assertFalse(node.isTotallyFree());
+ Assert.assertEquals(3, node.totalSlotsFree());
+ Assert.assertEquals(1, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+
+ List<ExecutorDetails> executors12 = new ArrayList<>();
+ executors12.add(new ExecutorDetails(2, 2));
+ node.assign(node.getFreeSlots().iterator().next(), topology1, executors12);
+ Assert.assertEquals(1, node.getRunningTopologies().size());
+ Assert.assertFalse(node.isTotallyFree());
+ Assert.assertEquals(2, node.totalSlotsFree());
+ Assert.assertEquals(2, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+
+ TopologyDetails topology2 = TestUtilsForResourceAwareScheduler.getTopology("topology2", new HashMap(), 1, 0, 2, 0, 0, 0);
+
+ List<ExecutorDetails> executors21 = new ArrayList<>();
+ executors21.add(new ExecutorDetails(1, 1));
+ node.assign(node.getFreeSlots().iterator().next(), topology2, executors21);
+ Assert.assertEquals(2, node.getRunningTopologies().size());
+ Assert.assertFalse(node.isTotallyFree());
+ Assert.assertEquals(1, node.totalSlotsFree());
+ Assert.assertEquals(3, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+
+ List<ExecutorDetails> executors22 = new ArrayList<>();
+ executors22.add(new ExecutorDetails(2, 2));
+ node.assign(node.getFreeSlots().iterator().next(), topology2, executors22);
+ Assert.assertEquals(2, node.getRunningTopologies().size());
+ Assert.assertFalse(node.isTotallyFree());
+ Assert.assertEquals(0, node.totalSlotsFree());
+ Assert.assertEquals(4, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+
+ node.freeAllSlots();
+ Assert.assertEquals(0, node.getRunningTopologies().size());
+ Assert.assertTrue(node.isTotallyFree());
+ Assert.assertEquals(4, node.totalSlotsFree());
+ Assert.assertEquals(0, node.totalSlotsUsed());
+ Assert.assertEquals(4, node.totalSlots());
+ }
+
+ @Test
+ public void sanityTestOfScheduling() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(1, 2, resourceMap);
+
+ Config config = new Config();
+ config.putAll(defaultTopologyConf);
+
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ TopologyDetails topology1 = TestUtilsForResourceAwareScheduler.getTopology("topology1", config, 1, 1, 1, 1, 0, 0);
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ Topologies topologies = new Topologies(topoMap);
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ SchedulerAssignment assignment = cluster.getAssignmentById(topology1.getId());
+ Set<WorkerSlot> assignedSlots = assignment.getSlots();
+ Set<String> nodesIDs = new HashSet<>();
+ for (WorkerSlot slot : assignedSlots) {
+ nodesIDs.add(slot.getNodeId());
+ }
+ Collection<ExecutorDetails> executors = assignment.getExecutors();
+
+ Assert.assertEquals(1, assignedSlots.size());
+ Assert.assertEquals(1, nodesIDs.size());
+ Assert.assertEquals(2, executors.size());
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ }
+
+ @Test
+ public void testTopologyWithMultipleSpouts() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(2, 4, resourceMap);
+
+ TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
+ builder1.setSpout("wordSpout1", new TestWordSpout(), 1);
+ builder1.setSpout("wordSpout2", new TestWordSpout(), 1);
+ builder1.setBolt("wordCountBolt1", new TestWordCounter(), 1).shuffleGrouping("wordSpout1").shuffleGrouping("wordSpout2");
+ builder1.setBolt("wordCountBolt2", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt1");
+ builder1.setBolt("wordCountBolt3", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt1");
+ builder1.setBolt("wordCountBolt4", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt2");
+ builder1.setBolt("wordCountBolt5", new TestWordCounter(), 1).shuffleGrouping("wordSpout2");
+ StormTopology stormTopology1 = builder1.createTopology();
+
+ Config config = new Config();
+ config.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 0, executorMap1, 0);
+
+ TopologyBuilder builder2 = new TopologyBuilder(); // a topology with two unconnected partitions
+ builder2.setSpout("wordSpoutX", new TestWordSpout(), 1);
+ builder2.setSpout("wordSpoutY", new TestWordSpout(), 1);
+ StormTopology stormTopology2 = builder2.createTopology();
+ Map<ExecutorDetails, String> executorMap2 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology2);
+ TopologyDetails topology2 = new TopologyDetails("topology2", config, stormTopology2, 0, executorMap2, 0);
+
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topoMap.put(topology2.getId(), topology2);
+ Topologies topologies = new Topologies(topoMap);
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
+ Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
+ Set<String> nodesIDs1 = new HashSet<>();
+ for (WorkerSlot slot : assignedSlots1) {
+ nodesIDs1.add(slot.getNodeId());
+ }
+ Collection<ExecutorDetails> executors1 = assignment1.getExecutors();
+
+ Assert.assertEquals(1, assignedSlots1.size());
+ Assert.assertEquals(1, nodesIDs1.size());
+ Assert.assertEquals(7, executors1.size());
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+
+ SchedulerAssignment assignment2 = cluster.getAssignmentById(topology2.getId());
+ Set<WorkerSlot> assignedSlots2 = assignment2.getSlots();
+ Set<String> nodesIDs2 = new HashSet<>();
+ for (WorkerSlot slot : assignedSlots2) {
+ nodesIDs2.add(slot.getNodeId());
+ }
+ Collection<ExecutorDetails> executors2 = assignment2.getExecutors();
+
+ Assert.assertEquals(1, assignedSlots2.size());
+ Assert.assertEquals(1, nodesIDs2.size());
+ Assert.assertEquals(2, executors2.size());
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology2.getId()));
+ }
+
+ @Test
+ public void testTopologySetCpuAndMemLoad() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(2, 2, resourceMap);
+
+ TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
+ builder1.setSpout("wordSpout", new TestWordSpout(), 1).setCPULoad(20.0).setMemoryLoad(200.0);
+ builder1.setBolt("wordCountBolt", new TestWordCounter(), 1).shuffleGrouping("wordSpout").setCPULoad(20.0).setMemoryLoad(200.0);
+ StormTopology stormTopology1 = builder1.createTopology();
+
+ Config config = new Config();
+ config.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 0, executorMap1, 0);
+
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ Topologies topologies = new Topologies(topoMap);
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
+ Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
+ double assignedMemory = 0.0;
+ double assignedCpu = 0.0;
+ Set<String> nodesIDs1 = new HashSet<>();
+ for (WorkerSlot slot : assignedSlots1) {
+ nodesIDs1.add(slot.getNodeId());
+ assignedMemory += slot.getAllocatedMemOnHeap() + slot.getAllocatedMemOffHeap();
+ assignedCpu += slot.getAllocatedCpu();
+
+ }
+ Collection<ExecutorDetails> executors1 = assignment1.getExecutors();
+
+ Assert.assertEquals(1, assignedSlots1.size());
+ Assert.assertEquals(1, nodesIDs1.size());
+ Assert.assertEquals(2, executors1.size());
+ Assert.assertEquals(400.0, assignedMemory, 0.001);
+ Assert.assertEquals(40.0, assignedCpu, 0.001);
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ }
+
+ @Test
+ public void testResourceLimitation() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(2, 2, resourceMap);
+
+ TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
+ builder1.setSpout("wordSpout", new TestWordSpout(), 2).setCPULoad(250.0).setMemoryLoad(1000.0, 200.0);
+ builder1.setBolt("wordCountBolt", new TestWordCounter(), 1).shuffleGrouping("wordSpout").setCPULoad(100.0).setMemoryLoad(500.0, 100.0);
+ StormTopology stormTopology1 = builder1.createTopology();
+
+ Config config = new Config();
+ config.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 2, executorMap1, 0);
+
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ Topologies topologies = new Topologies(topoMap);
+
+ rs.prepare(config);
+ rs.schedule(topologies, cluster);
+
+ SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
+ Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
+ Set<String> nodesIDs1 = new HashSet<>();
+ for (WorkerSlot slot : assignedSlots1) {
+ nodesIDs1.add(slot.getNodeId());
+ }
+ Collection<ExecutorDetails> executors1 = assignment1.getExecutors();
+ List<Double> assignedExecutorMemory = new ArrayList<>();
+ List<Double> assignedExecutorCpu = new ArrayList<>();
+ for (ExecutorDetails executor : executors1) {
+ assignedExecutorMemory.add(topology1.getTotalMemReqTask(executor));
+ assignedExecutorCpu.add(topology1.getTotalCpuReqTask(executor));
+ }
+ Collections.sort(assignedExecutorCpu);
+ Collections.sort(assignedExecutorMemory);
+
+ Map<ExecutorDetails, SupervisorDetails> executorToSupervisor = new HashMap<>();
+ Map<SupervisorDetails, List<ExecutorDetails>> supervisorToExecutors = new HashMap<>();
+ Map<Double, Double> cpuAvailableToUsed = new HashMap();
+ Map<Double, Double> memoryAvailableToUsed = new HashMap();
+
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : assignment1.getExecutorToSlot().entrySet()) {
+ executorToSupervisor.put(entry.getKey(), cluster.getSupervisorById(entry.getValue().getNodeId()));
+ }
+ for (Map.Entry<ExecutorDetails, SupervisorDetails> entry : executorToSupervisor.entrySet()) {
+ List<ExecutorDetails> executorsOnSupervisor = supervisorToExecutors.get(entry.getValue());
+ if (executorsOnSupervisor == null) {
+ executorsOnSupervisor = new ArrayList<>();
+ supervisorToExecutors.put(entry.getValue(), executorsOnSupervisor);
+ }
+ executorsOnSupervisor.add(entry.getKey());
+ }
+ for (Map.Entry<SupervisorDetails, List<ExecutorDetails>> entry : supervisorToExecutors.entrySet()) {
+ Double supervisorTotalCpu = entry.getKey().getTotalCPU();
+ Double supervisorTotalMemory = entry.getKey().getTotalMemory();
+ Double supervisorUsedCpu = 0.0;
+ Double supervisorUsedMemory = 0.0;
+ for (ExecutorDetails executor: entry.getValue()) {
+ supervisorUsedMemory += topology1.getTotalCpuReqTask(executor);
+ supervisorTotalCpu += topology1.getTotalMemReqTask(executor);
+ }
+ cpuAvailableToUsed.put(supervisorTotalCpu, supervisorUsedCpu);
+ memoryAvailableToUsed.put(supervisorTotalMemory, supervisorUsedMemory);
+ }
+ // executor0 resides one one worker (on one), executor1 and executor2 on another worker (on the other node)
+ Assert.assertEquals(2, assignedSlots1.size());
+ Assert.assertEquals(2, nodesIDs1.size());
+ Assert.assertEquals(3, executors1.size());
+
+ Assert.assertEquals(100.0, assignedExecutorCpu.get(0), 0.001);
+ Assert.assertEquals(250.0, assignedExecutorCpu.get(1), 0.001);
+ Assert.assertEquals(250.0, assignedExecutorCpu.get(2), 0.001);
+ Assert.assertEquals(600.0, assignedExecutorMemory.get(0), 0.001);
+ Assert.assertEquals(1200.0, assignedExecutorMemory.get(1), 0.001);
+ Assert.assertEquals(1200.0, assignedExecutorMemory.get(2), 0.001);
+
+ for (Map.Entry<Double, Double> entry : memoryAvailableToUsed.entrySet()) {
+ Assert.assertTrue(entry.getKey()- entry.getValue() >= 0);
+ }
+ for (Map.Entry<Double, Double> entry : cpuAvailableToUsed.entrySet()) {
+ Assert.assertTrue(entry.getKey()- entry.getValue() >= 0);
+ }
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ }
+
+ @Test
+ public void testScheduleResilience() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(2, 2, resourceMap);
+
+ TopologyBuilder builder1 = new TopologyBuilder();
+ builder1.setSpout("wordSpout1", new TestWordSpout(), 3);
+ StormTopology stormTopology1 = builder1.createTopology();
+ Config config1 = new Config();
+ config1.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 3, executorMap1, 0);
+
+ TopologyBuilder builder2 = new TopologyBuilder();
+ builder2.setSpout("wordSpout2", new TestWordSpout(), 2);
+ StormTopology stormTopology2 = builder2.createTopology();
+ Config config2 = new Config();
+ config2.putAll(defaultTopologyConf);
+ // memory requirement is large enough so that two executors can not be fully assigned to one node
+ config2.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 1280.0);
+ Map<ExecutorDetails, String> executorMap2 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology2);
+ TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 2, executorMap2, 0);
+
+ // Test1: When a worker fails, RAS does not alter existing assignments on healthy workers
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology2.getId(), topology2);
+ Topologies topologies = new Topologies(topoMap);
+
+ rs.prepare(config1);
+ rs.schedule(topologies, cluster);
+
+ SchedulerAssignmentImpl assignment = (SchedulerAssignmentImpl)cluster.getAssignmentById(topology2.getId());
+ // pick a worker to mock as failed
+ WorkerSlot failedWorker = new ArrayList<WorkerSlot>(assignment.getSlots()).get(0);
+ Map<ExecutorDetails, WorkerSlot> executorToSlot = assignment.getExecutorToSlot();
+ List<ExecutorDetails> failedExecutors = new ArrayList<>();
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : executorToSlot.entrySet()) {
+ if (entry.getValue().equals(failedWorker)) {
+ failedExecutors.add(entry.getKey());
+ }
+ }
+ for (ExecutorDetails executor : failedExecutors) {
+ executorToSlot.remove(executor); // remove executor details assigned to the failed worker
+ }
+ Map<ExecutorDetails, WorkerSlot> copyOfOldMapping = new HashMap<>(executorToSlot);
+ Set<ExecutorDetails> healthyExecutors = copyOfOldMapping.keySet();
+
+ rs.schedule(topologies, cluster);
+ SchedulerAssignment newAssignment = cluster.getAssignmentById(topology2.getId());
+ Map<ExecutorDetails, WorkerSlot> newExecutorToSlot = newAssignment.getExecutorToSlot();
+
+ for (ExecutorDetails executor : healthyExecutors) {
+ Assert.assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
+ }
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology2.getId()));
+ // end of Test1
+
+ // Test2: When a supervisor fails, RAS does not alter existing assignments
+ executorToSlot = new HashMap<>();
+ executorToSlot.put(new ExecutorDetails(0, 0), new WorkerSlot("sup-0", 0));
+ executorToSlot.put(new ExecutorDetails(1, 1), new WorkerSlot("sup-0", 1));
+ executorToSlot.put(new ExecutorDetails(2, 2), new WorkerSlot("sup-1", 1));
+ Map<String, SchedulerAssignmentImpl> existingAssignments = new HashMap<>();
+ assignment = new SchedulerAssignmentImpl(topology1.getId(), executorToSlot);
+ existingAssignments.put(topology1.getId(), assignment);
+ copyOfOldMapping = new HashMap<>(executorToSlot);
+ Set<ExecutorDetails> existingExecutors = copyOfOldMapping.keySet();
+ Map<String, SupervisorDetails> supMap1 = new HashMap<>(supMap);
+ supMap1.remove("sup-0"); // mock the supervisor sup-0 as a failed supervisor
+ Cluster cluster1 = new Cluster(iNimbus, supMap1, existingAssignments, config1);
+
+ topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster1);
+
+ newAssignment = cluster1.getAssignmentById(topology1.getId());
+ newExecutorToSlot = newAssignment.getExecutorToSlot();
+
+ for (ExecutorDetails executor : existingExecutors) {
+ Assert.assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
+ }
+ Assert.assertEquals("Fully Scheduled", cluster1.getStatusMap().get(topology1.getId()));
+ // end of Test2
+
+ // Test3: When a supervisor and a worker on it fails, RAS does not alter existing assignments
+ executorToSlot = new HashMap<>();
+ executorToSlot.put(new ExecutorDetails(0, 0), new WorkerSlot("sup-0", 1)); // the worker to orphan
+ executorToSlot.put(new ExecutorDetails(1, 1), new WorkerSlot("sup-0", 2)); // the worker that fails
+ executorToSlot.put(new ExecutorDetails(2, 2), new WorkerSlot("sup-1", 1)); // the healthy worker
+ existingAssignments = new HashMap<>();
+ assignment = new SchedulerAssignmentImpl(topology1.getId(), executorToSlot);
+ existingAssignments.put(topology1.getId(), assignment);
+ // delete one worker of sup-0 (failed) from topo1 assignment to enable actual schedule for testing
+ executorToSlot.remove(new ExecutorDetails(1, 1));
+
+ copyOfOldMapping = new HashMap<>(executorToSlot);
+ existingExecutors = copyOfOldMapping.keySet(); // namely the two eds on the orphaned worker and the healthy worker
+ supMap1 = new HashMap<>(supMap);
+ supMap1.remove("sup-0"); // mock the supervisor sup-0 as a failed supervisor
+ cluster1 = new Cluster(iNimbus, supMap1, existingAssignments, config1);
+
+ topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster1);
+
+ newAssignment = cluster1.getAssignmentById(topology1.getId());
+ newExecutorToSlot = newAssignment.getExecutorToSlot();
+
+ for (ExecutorDetails executor : existingExecutors) {
+ Assert.assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
+ }
+ Assert.assertEquals("Fully Scheduled", cluster1.getStatusMap().get(topology1.getId()));
+ // end of Test3
+
+ // Test4: Scheduling a new topology does not disturb other assignments unnecessarily
+ cluster1 = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster1);
+ assignment = (SchedulerAssignmentImpl)cluster1.getAssignmentById(topology1.getId());
+ executorToSlot = assignment.getExecutorToSlot();
+ copyOfOldMapping = new HashMap<>(executorToSlot);
+
+ topoMap.put(topology2.getId(), topology2);
+ topologies = new Topologies(topoMap);
+ rs.schedule(topologies, cluster1);
+
+ newAssignment = (SchedulerAssignmentImpl)cluster1.getAssignmentById(topology1.getId());
+ newExecutorToSlot = newAssignment.getExecutorToSlot();
+
+ for (ExecutorDetails executor : copyOfOldMapping.keySet()) {
+ Assert.assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
+ }
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster1.getStatusMap().get(topology1.getId()));
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster1.getStatusMap().get(topology2.getId()));
+ }
+
+ @Test
+ public void testHeterogeneousCluster() {
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap1 = new HashMap<>(); // strong supervisor node
+ resourceMap1.put(Config.SUPERVISOR_CPU_CAPACITY, 800.0);
+ resourceMap1.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 4096.0);
+ Map<String, Number> resourceMap2 = new HashMap<>(); // weak supervisor node
+ resourceMap2.put(Config.SUPERVISOR_CPU_CAPACITY, 200.0);
+ resourceMap2.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1024.0);
+
+ Map<String, SupervisorDetails> supMap = new HashMap<String, SupervisorDetails>();
+ for (int i = 0; i < 2; i++) {
+ List<Number> ports = new LinkedList<Number>();
+ for (int j = 0; j < 4; j++) {
+ ports.add(j);
+ }
+ SupervisorDetails sup = new SupervisorDetails("sup-" + i, "host-" + i, null, ports, (Map)(i == 0 ? resourceMap1 : resourceMap2));
+ supMap.put(sup.getId(), sup);
+ }
+
+ // topo1 has one single huge task that can not be handled by the small-super
+ TopologyBuilder builder1 = new TopologyBuilder();
+ builder1.setSpout("wordSpout1", new TestWordSpout(), 1).setCPULoad(300.0).setMemoryLoad(2000.0, 48.0);
+ StormTopology stormTopology1 = builder1.createTopology();
+ Config config1 = new Config();
+ config1.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 1, executorMap1, 0);
+
+ // topo2 has 4 large tasks
+ TopologyBuilder builder2 = new TopologyBuilder();
+ builder2.setSpout("wordSpout2", new TestWordSpout(), 4).setCPULoad(100.0).setMemoryLoad(500.0, 12.0);
+ StormTopology stormTopology2 = builder2.createTopology();
+ Config config2 = new Config();
+ config2.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap2 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology2);
+ TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 1, executorMap2, 0);
+
+ // topo3 has 4 large tasks
+ TopologyBuilder builder3 = new TopologyBuilder();
+ builder3.setSpout("wordSpout3", new TestWordSpout(), 4).setCPULoad(20.0).setMemoryLoad(200.0, 56.0);
+ StormTopology stormTopology3 = builder3.createTopology();
+ Config config3 = new Config();
+ config3.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap3 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology3);
+ TopologyDetails topology3 = new TopologyDetails("topology3", config2, stormTopology3, 1, executorMap3, 0);
+
+ // topo4 has 12 small tasks, whose mem usage does not exactly divide a node's mem capacity
+ TopologyBuilder builder4 = new TopologyBuilder();
+ builder4.setSpout("wordSpout4", new TestWordSpout(), 12).setCPULoad(30.0).setMemoryLoad(100.0, 0.0);
+ StormTopology stormTopology4 = builder4.createTopology();
+ Config config4 = new Config();
+ config4.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap4 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology4);
+ TopologyDetails topology4 = new TopologyDetails("topology4", config4, stormTopology4, 1, executorMap4, 0);
+
+ // topo5 has 40 small tasks, it should be able to exactly use up both the cpu and mem in the cluster
+ TopologyBuilder builder5 = new TopologyBuilder();
+ builder5.setSpout("wordSpout5", new TestWordSpout(), 40).setCPULoad(25.0).setMemoryLoad(100.0, 28.0);
+ StormTopology stormTopology5 = builder5.createTopology();
+ Config config5 = new Config();
+ config5.putAll(defaultTopologyConf);
+ Map<ExecutorDetails, String> executorMap5 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology5);
+ TopologyDetails topology5 = new TopologyDetails("topology5", config5, stormTopology5, 1, executorMap5, 0);
+
+ // Test1: Launch topo 1-3 together, it should be able to use up either mem or cpu resource due to exact division
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topoMap.put(topology2.getId(), topology2);
+ topoMap.put(topology3.getId(), topology3);
+ Topologies topologies = new Topologies(topoMap);
+ rs.prepare(config1);
+ rs.schedule(topologies, cluster);
+
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology2.getId()));
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology3.getId()));
+
+ Map<SupervisorDetails, Double> superToCpu = TestUtilsForResourceAwareScheduler.getSupervisorToCpuUsage(cluster, topologies);
+ Map<SupervisorDetails, Double> superToMem = TestUtilsForResourceAwareScheduler.getSupervisorToMemoryUsage(cluster, topologies);
+
+ final Double EPSILON = 0.0001;
+ for (SupervisorDetails supervisor : supMap.values()) {
+ Double cpuAvailable = supervisor.getTotalCPU();
+ Double memAvailable = supervisor.getTotalMemory();
+ Double cpuUsed = superToCpu.get(supervisor);
+ Double memUsed = superToMem.get(supervisor);
+ Assert.assertTrue((Math.abs(memAvailable - memUsed) < EPSILON) || (Math.abs(cpuAvailable - cpuUsed) < EPSILON));
+ }
+ // end of Test1
+
+ // Test2: Launch topo 1, 2 and 4, they together request a little more mem than available, so one of the 3 topos will not be scheduled
+ cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ topoMap.put(topology2.getId(), topology2);
+ topoMap.put(topology4.getId(), topology4);
+ topologies = new Topologies(topoMap);
+ rs.prepare(config1);
+ rs.schedule(topologies, cluster);
+ int numTopologiesAssigned = 0;
+ if (cluster.getStatusMap().get(topology1.getId()).equals("Running - Fully Scheduled by DefaultResourceAwareStrategy")) {
+ numTopologiesAssigned++;
+ }
+ if (cluster.getStatusMap().get(topology2.getId()).equals("Running - Fully Scheduled by DefaultResourceAwareStrategy")) {
+ numTopologiesAssigned++;
+ }
+ if (cluster.getStatusMap().get(topology4.getId()).equals("Running - Fully Scheduled by DefaultResourceAwareStrategy")) {
+ numTopologiesAssigned++;
+ }
+ Assert.assertEquals(2, numTopologiesAssigned);
+ //end of Test2
+
+ //Test3: "Launch topo5 only, both mem and cpu should be exactly used up"
+ cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ topoMap = new HashMap<>();
+ topoMap.put(topology5.getId(), topology5);
+ topologies = new Topologies(topoMap);
+ rs.prepare(config1);
+ rs.schedule(topologies, cluster);
+ superToCpu = TestUtilsForResourceAwareScheduler.getSupervisorToCpuUsage(cluster, topologies);
+ superToMem = TestUtilsForResourceAwareScheduler.getSupervisorToMemoryUsage(cluster, topologies);
+ for (SupervisorDetails supervisor : supMap.values()) {
+ Double cpuAvailable = supervisor.getTotalCPU();
+ Double memAvailable = supervisor.getTotalMemory();
+ Double cpuUsed = superToCpu.get(supervisor);
+ Double memUsed = superToMem.get(supervisor);
+ Assert.assertEquals(cpuAvailable, cpuUsed, 0.0001);
+ Assert.assertEquals(memAvailable, memUsed, 0.0001);
+ }
+ //end of Test3
+ }
+
+ @Test
+ public void testTopologyWorkerMaxHeapSize() {
+ // Test1: If RAS spreads executors across multiple workers based on the set limit for a worker used by the topology
+ INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
+ Map<String, Number> resourceMap = new HashMap<>();
+ resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 400.0);
+ resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 2000.0);
+ Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(2, 2, resourceMap);
+
+ TopologyBuilder builder1 = new TopologyBuilder();
+ builder1.setSpout("wordSpout1", new TestWordSpout(), 4);
+ StormTopology stormTopology1 = builder1.createTopology();
+ Config config1 = new Config();
+ config1.putAll(defaultTopologyConf);
+ config1.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 128.0);
+ Map<ExecutorDetails, String> executorMap1 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology1);
+ TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 1, executorMap1, 0);
+ Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config1);
+ ResourceAwareScheduler rs = new ResourceAwareScheduler();
+ Map<String, TopologyDetails> topoMap = new HashMap<>();
+ topoMap.put(topology1.getId(), topology1);
+ Topologies topologies = new Topologies(topoMap);
+ rs.prepare(config1);
+ rs.schedule(topologies, cluster);
+ Assert.assertEquals("Running - Fully Scheduled by DefaultResourceAwareStrategy", cluster.getStatusMap().get(topology1.getId()));
+ Assert.assertEquals(4, cluster.getAssignedNumWorkers(topology1));
+
+ // Test2: test when no more workers are available due to topology worker max heap size limit but there is memory is still available
+ // wordSpout2 is going to contain 5 executors that needs scheduling. Each of those executors has a memory requirement of 128.0 MB
+ // The cluster contains 4 free WorkerSlots. For this topolology each worker is limited to a max heap size of 128.0
+ // Thus, one executor not going to be able to get scheduled thus failing the scheduling of this topology and no executors of this topology will be scheduleded
+ TopologyBuilder builder2 = new TopologyBuilder();
+ builder2.setSpout("wordSpout2", new TestWordSpout(), 5);
+ StormTopology stormTopology2 = builder2.createTopology();
+ Config config2 = new Config();
+ config2.putAll(defaultTopologyConf);
+ config2.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 128.0);
+ Map<ExecutorDetails, String> executorMap2 = TestUtilsForResourceAwareScheduler.genExecsAndComps(stormTopology2);
+ TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 1, executorMap2, 0);
+ cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config2);
+ topoMap = new HashMap<>();
+ topoMap.put(topology2.getId(), topology2);
+ topologies = new Topologies(topoMap);
+ rs.prepare(config2);
+ rs.schedule(topologies, cluster);
+ Assert.assertEquals("Not enough resources to schedule - 0/5 executors scheduled", cluster.getStatusMap().get(topology2.getId()));
+ Assert.assertEquals(5, cluster.getUnassignedExecutors(topology2).size());
+ }
+
+ @Test(expected=IllegalArgumentException.class)
+ public void testMemoryLoadLargerThanMaxHeapSize() throws Exception {
+ // Topology will not be able to be successfully scheduled: Config TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB=128.0 < 129.0,
+ // Largest memory requirement of a component in the topology).
+ TopologyBuilder builder1 = new TopologyBuilder();
+ builder1.setSpout("wordSpout1", new TestWordSpout(), 4);
+ StormTopology stormTopology1 = builder1.createTopology();
+ Config config1 = new Config();
+ config1.putAll(defaultTopologyConf);
+ config1.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 128.0);
+ config1.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 129.0);
+ StormSubmitter.submitTopologyWithProgressBar("test", config1, stormTopology1);
+ }
+
@Test
public void TestReadInResourceAwareSchedulerUserPools() {
Map fromFile = Utils.findAndReadConfigFile("user-resource-pools.yaml", false);
@@ -386,717 +1069,6 @@ public class TestResourceAwareScheduler {
}
/**
- * The resources in the cluster are limited. In the first round of scheduling, all resources in the cluster is used.
- * User jerry submits another toploogy. Since user jerry has his resource guarantees satisfied, and user bobby
- * has exceeded his resource guarantee, topo-3 from user bobby should be evicted.
- */
- @Test
- public void testEviction() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 200.0);
- resourceUserPool.get("derek").put("memory", 2000.0);
-
- config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 20);
-
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
-
- Topologies topologies = new Topologies(topoMap);
-
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
-
- //user jerry submits another topology
- topoMap.put(topo6.getId(), topo6);
- topologies = new Topologies(topoMap);
-
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("correct topology to evict", "topo-3", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName());
- }
-
- @Test
- public void TestEvictMultipleTopologies() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 2, 0, 1, 0, currentTime - 2, 10);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29);
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
-
- //user jerry submits another topology
- topoMap.put(topo1.getId(), topo1);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("assert topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
- }
-
- /**
- * Eviction order:
- * topo-3: since user bobby don't have any resource guarantees and topo-3 is the lowest priority for user bobby
- * topo-2: since user bobby don't have any resource guarantees and topo-2 is the next lowest priority for user bobby
- * topo-5: since user derek has exceeded his resource guarantee while user jerry has not. topo-5 and topo-4 has the same priority
- * but topo-4 was submitted earlier thus we choose that one to evict
- */
- @Test
- public void TestEvictMultipleTopologiesFromMultipleUsersInCorrectOrder() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 300.0);
- resourceUserPool.get("jerry").put("memory", 3000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo7 = TestUtilsForResourceAwareScheduler.getTopology("topo-7", config, 1, 0, 1, 0, currentTime - 2, 10);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
-
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
-
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 29);
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 15, 29);
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
-
- //user jerry submits another topology
- topoMap.put(topo1.getId(), topo1);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("correct topology to evict", rs.getUser("bobby").getTopologiesAttempted().iterator().next().getName(), "topo-3");
-
- topoMap.put(topo6.getId(), topo6);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
-
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
-
- topoMap.put(topo7.getId(), topo7);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 3, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
- Assert.assertEquals("correct topology to evict", rs.getUser("derek").getTopologiesAttempted().iterator().next().getName(), "topo-4");
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesAttempted()) {
- Assert.assertFalse("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of running topologies", 0, rs.getUser("bobby").getTopologiesRunning().size());
-
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("bobby").getTopologiesAttempted()) != null);
- Assert.assertTrue("correct topology to evict", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("bobby").getTopologiesAttempted()) != null);
- }
-
- /**
- * If topologies from other users cannot be evicted to make space
- * check if there is a topology with lower priority that can be evicted from the current user
- */
- @Test
- public void TestEvictTopologyFromItself() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 200.0);
- resourceUserPool.get("jerry").put("memory", 2000.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 100.0);
- resourceUserPool.get("derek").put("memory", 1000.0);
-
- config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20);
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20);
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 29);
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
-
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 10);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
-
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 29);
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo2.getId(), topo2);
- topoMap.put(topo5.getId(), topo5);
- topoMap.put(topo6.getId(), topo6);
-
- Topologies topologies = new Topologies(topoMap);
-
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
-
- //user jerry submits another topology into a full cluster
- // topo3 should not be able to scheduled
- topoMap.put(topo3.getId(), topo3);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 1, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
- //make sure that topo-3 didn't get scheduled.
- Assert.assertEquals("correct topology in attempted queue", rs.getUser("jerry").getTopologiesAttempted().iterator().next().getName(), "topo-3");
-
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
-
- //user jerry submits another topology but this one should be scheduled since it has higher priority than than the
- //rest of jerry's running topologies
- topoMap.put(topo4.getId(), topo4);
- topologies = new Topologies(topoMap);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 2, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 2, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
- Assert.assertTrue("correct topology in attempted queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-3", rs.getUser("jerry").getTopologiesAttempted()) != null);
- //Either topo-1 or topo-2 should have gotten evicted
- Assert.assertTrue("correct topology in attempted queue", ((TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-1", rs.getUser("jerry").getTopologiesAttempted())) != null)
- || (TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-2", rs.getUser("jerry").getTopologiesAttempted()) != null));
- //assert that topo-4 got scheduled
- Assert.assertTrue("correct topology in running queue", TestUtilsForResourceAwareScheduler.findTopologyInSetFromName("topo-4", rs.getUser("jerry").getTopologiesRunning()) != null);
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("bobby").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("bobby").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("bobby").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("bobby").getTopologiesInvalid().size());
- }
-
- /**
- * If users are above his or her guarantee, check if topology eviction works correct
- */
- @Test
- public void TestOverGuaranteeEviction() {
- INimbus iNimbus = new TestUtilsForResourceAwareScheduler.INimbusTest();
- Map<String, Number> resourceMap = new HashMap<String, Number>();
- resourceMap.put(Config.SUPERVISOR_CPU_CAPACITY, 100.0);
- resourceMap.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1000.0);
- Map<String, SupervisorDetails> supMap = TestUtilsForResourceAwareScheduler.genSupervisors(4, 4, resourceMap);
- Config config = new Config();
- config.putAll(Utils.readDefaultConfig());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config.put(Config.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
- config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
- config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 100.0);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 500);
- config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 500);
- Map<String, Map<String, Number>> resourceUserPool = new HashMap<String, Map<String, Number>>();
- resourceUserPool.put("jerry", new HashMap<String, Number>());
- resourceUserPool.get("jerry").put("cpu", 70.0);
- resourceUserPool.get("jerry").put("memory", 700.0);
-
- resourceUserPool.put("bobby", new HashMap<String, Number>());
- resourceUserPool.get("bobby").put("cpu", 100.0);
- resourceUserPool.get("bobby").put("memory", 1000.0);
-
- resourceUserPool.put("derek", new HashMap<String, Number>());
- resourceUserPool.get("derek").put("cpu", 25.0);
- resourceUserPool.get("derek").put("memory", 250.0);
-
- config.put(Config.RESOURCE_AWARE_SCHEDULER_USER_POOLS, resourceUserPool);
- Cluster cluster = new Cluster(iNimbus, supMap, new HashMap<String, SchedulerAssignmentImpl>(), config);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
-
- TopologyDetails topo1 = TestUtilsForResourceAwareScheduler.getTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 20);
- TopologyDetails topo2 = TestUtilsForResourceAwareScheduler.getTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "bobby");
-
- TopologyDetails topo3 = TestUtilsForResourceAwareScheduler.getTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 10);
- TopologyDetails topo4 = TestUtilsForResourceAwareScheduler.getTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10);
-
- config.put(Config.TOPOLOGY_SUBMITTER_USER, "derek");
-
- TopologyDetails topo5 = TestUtilsForResourceAwareScheduler.getTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29);
- TopologyDetails topo6 = TestUtilsForResourceAwareScheduler.getTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10);
-
- Map<String, TopologyDetails> topoMap = new HashMap<String, TopologyDetails>();
- topoMap.put(topo1.getId(), topo1);
- topoMap.put(topo3.getId(), topo3);
- topoMap.put(topo4.getId(), topo4);
- topoMap.put(topo5.getId(), topo5);
-
- Topologies topologies = new Topologies(topoMap);
-
- ResourceAwareScheduler rs = new ResourceAwareScheduler();
-
- rs.prepare(config);
- rs.schedule(topologies, cluster);
-
- for (TopologyDetails topo : rs.getUser("jerry").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("jerry").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("jerry").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("jerry").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("jerry").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("derek").getTopologiesRunning()) {
- Assert.assertTrue("Assert scheduling topology success", TestUtilsForResourceAwareScheduler.assertStatusSuccess(cluster.getStatusMap().get(topo.getId())));
- }
- Assert.assertEquals("# of running topologies", 1, rs.getUser("derek").getTopologiesRunning().size());
- Assert.assertEquals("# of pending topologies", 0, rs.getUser("derek").getTopologiesPending().size());
- Assert.assertEquals("# of attempted topologies", 0, rs.getUser("derek").getTopologiesAttempted().size());
- Assert.assertEquals("# of invalid topologies", 0, rs.getUser("derek").getTopologiesInvalid().size());
-
- for (TopologyDetails topo : rs.getUser("bobby").getTopologiesRunning()) {
<TRUNCATED>