You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/11/07 16:38:46 UTC
[2/5] storm git commit: STORM-2792: Remove RAS EvictionPolicy and
cleanup
STORM-2792: Remove RAS EvictionPolicy and cleanup
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5f0abb21
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5f0abb21
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5f0abb21
Branch: refs/heads/master
Commit: 5f0abb21c83c129af5b4acce2d0bb3bbb0f41887
Parents: 5d95e71
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Mon Oct 30 16:09:45 2017 -0500
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Fri Nov 3 08:16:49 2017 -0500
----------------------------------------------------------------------
conf/defaults.yaml | 1 -
docs/Resource_Aware_Scheduler_overview.md | 80 ++++--
storm-client/pom.xml | 4 +
.../java/org/apache/storm/DaemonConfig.java | 17 +-
.../org/apache/storm/daemon/nimbus/Nimbus.java | 4 +-
.../org/apache/storm/scheduler/Cluster.java | 70 +----
.../storm/scheduler/ISchedulingState.java | 5 +-
.../storm/scheduler/SupervisorResources.java | 79 ++++++
.../apache/storm/scheduler/TopologyDetails.java | 4 +-
.../resource/ResourceAwareScheduler.java | 211 ++++++++-------
.../storm/scheduler/resource/ResourceUtils.java | 45 ++--
.../apache/storm/scheduler/resource/User.java | 40 +--
.../eviction/DefaultEvictionStrategy.java | 110 --------
.../strategies/eviction/IEvictionStrategy.java | 37 ---
.../DefaultSchedulingPriorityStrategy.java | 165 +++++++++---
.../FIFOSchedulingPriorityStrategy.java | 79 ++++++
.../priority/ISchedulingPriorityStrategy.java | 8 +-
.../DefaultResourceAwareStrategy.java | 9 +-
.../java/org/apache/storm/TestRebalance.java | 2 -
.../apache/storm/daemon/nimbus/NimbusTest.java | 7 +-
.../org/apache/storm/scheduler/ClusterTest.java | 10 +-
.../resource/TestResourceAwareScheduler.java | 263 +++++++++----------
.../TestUtilsForResourceAwareScheduler.java | 195 +++++++-------
.../eviction/TestDefaultEvictionStrategy.java | 32 ++-
.../TestFIFOSchedulingPriorityStrategy.java | 89 +++++++
.../TestDefaultResourceAwareStrategy.java | 4 +-
26 files changed, 876 insertions(+), 694 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/conf/defaults.yaml
----------------------------------------------------------------------
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index ad34054..d931b21 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -273,7 +273,6 @@ topology.component.resources.offheap.memory.mb: 0.0
topology.component.cpu.pcore.percent: 10.0
topology.worker.max.heap.size.mb: 768.0
topology.scheduler.strategy: "org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy"
-resource.aware.scheduler.eviction.strategy: "org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy"
resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy"
blacklist.scheduler.tolerance.time.secs: 300
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/docs/Resource_Aware_Scheduler_overview.md
----------------------------------------------------------------------
diff --git a/docs/Resource_Aware_Scheduler_overview.md b/docs/Resource_Aware_Scheduler_overview.md
index 94d4106..fc0aaf6 100644
--- a/docs/Resource_Aware_Scheduler_overview.md
+++ b/docs/Resource_Aware_Scheduler_overview.md
@@ -24,7 +24,6 @@ http://www.slideshare.net/HadoopSummit/resource-aware-scheduling-in-apache-storm
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)
@@ -157,6 +156,7 @@ Example of Usage:
### Other Configurations
The user can set some default configurations for the Resource Aware Scheduler in *conf/storm.yaml*:
+
```
//default value if on heap memory requirement is not specified for a component
topology.component.resources.onheap.memory.mb: 128.0
@@ -243,50 +243,72 @@ http://dl.acm.org/citation.cfm?id=2814808
<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:
+The order of scheduling and eviction is determined by a pluggable interface in which the cluster owner can define how topologies should be scheduled. For the owner to define his or her own prioritization strategy, she or he needs to implement the ISchedulingPriorityStrategy interface. A user can set the scheduling priority strategy by setting the `DaemonConfig.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.
+
+Topologies are scheduled starting at the beginning of the list returned by this plugin. If there are not enough resources to schedule the topology others are evicted starting at the end of the list. Eviction stops when there are no lower priority topologies left to evict.
**DefaultSchedulingPriorityStrategy**
-The order of scheduling should be based on the distance between a user’s 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.
+In the past the order of scheduling was based on the distance between a user’s current resource allocation and his or her guaranteed allocation.
+
+We currently use a slightly different approach. We simulate scheduling the highest priority topology for each user and score the topology for each of the resources using the formula
+
+```
+(Requested + Assigned - Guaranteed)/Available
+```
+
+Where
+
+ * `Requested` is the resource requested by this topology (or a approximation of it for complex requests like shared memory)
+ * `Assigned` is the resources already assigned by the simulation.
+ * `Guaranteed` is the resource guarantee for this user
+ * `Available` is the amount of that resource currently available in the cluster.
-For example:
+This gives a score that is negative for guaranteed requests and a score that is positive for requests that are not within the guarantee.
-|User|Resource Guarantee|Resource Allocated|
-|----|------------------|------------------|
-|A|<10 CPU, 50GB>|<2 CPU, 40 GB>|
-|B|< 20 CPU, 25GB>|<15 CPU, 10 GB>|
+To combine different resources the maximum of all the indavidual resource scores is used. This guarantees that if a user would go over a guarantee for a single resource it would not be offset by being under guarantee on any other resources.
-User A’s average percentage satisfied of resource guarantee:
+For Example:
-(2/10+40/50)/2 = 0.5
+Assume we have to schedule the following topologies.
-User B’s average percentage satisfied of resource guarantee:
+|ID|User|CPU|Memory|Priority|
+|---|----|---|------|-------|
+|A-1|A|100|1,000|1|
+|A-2|A|100|1,000|10|
+|B-1|B|100|1,000|1|
+|B-2|B|100|1,000|10|
-(15/20+10/25)/2 = 0.575
+The cluster as a whole has 300 CPU and 4,000 Memory.
-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.
+User A is guaranteed 100 CPU and 1,000 Memory. User B is guaranteed 200 CPU and 1,500 Memory. The scores for the most important, lowest priority number, topologies for each user would be.
+
+```
+A-1 Score = max(CPU: (100 + 0 - 100)/300, MEM: (1,000 + 0 - 1,000)/4,000) = 0
+B-1 Score = max(CPU: (100 + 0 - 200)/300, MEM: (1,000 + 0 - 1,500)/4,000) = -0.125
+```
-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’s resource guarantee is completely satisfied, the user’s average percentage satisfied of resource guarantee will be greater than or equal to 1.
+`B-1` has the lowest score so it would be the highest priority topology to schedule. In the next round the scores would be.
-<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-1 Score = max(CPU: (100 + 0 - 100)/200, MEM: (1,000 + 0 - 1,000)/3,000) = 0
+B-2 Score = max(CPU: (100 + 100 - 200)/200, MEM: (1,000 + 1,000 - 1,500)/3,000) = 0.167
```
-A default eviction strategy is provided. The following explains how the default topology eviction strategy works
-**DefaultEvictionStrategy**
+`A-1` has the lowest score now so it would be the next higest priority topology to schedule.
+
+This process would be repeated until all of the topologies are ordered, even if there are no resources left on the cluster to schedule a topology.
+
+**FIFOSchedulingPriorityStrategy**
-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.
+The FIFO strategy is intended more for test or staging clusters where users are running integration tests or doing development work. Topologies in these situations tend to be short lived and at times a user may forget that they are running a topology at all.
-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.
+To try and be as fair as possible to users running short lived topologies the `FIFOSchedulingPriorityStrategy` extends the `DefaultSchedulingPriorityStrategy` so that any negative score (a.k.a. a topology that fits within a user's guarantees) would remain unchanged, but positive scores are replaced with the uptime of the topology.
-
+This respects the guarantees of a user, but at the same time it gives priority for the rest of the resources to the most recently launched topology. Older topologies, that have probably been forgotten about, are then least likely to get resources.
<div id='Profiling-Resource-Usage'/>
## Profiling Resource Usage
@@ -294,7 +316,8 @@ We should never evict a topology from a user that does not have his or her resou
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);
@@ -303,7 +326,9 @@ To get an idea of how much memory/CPU your topology is actually using you can ad
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>
@@ -311,20 +336,23 @@ The CPU metrics will require you to add
<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 *
+## *Enhancements on original DefaultResourceAwareStrategy*
The default resource aware scheduling strategy as described in the paper above has two main scheduling phases:
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-client/pom.xml
----------------------------------------------------------------------
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index d0cbe0e..c4accca 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -193,6 +193,10 @@
<groupId>org.hamcrest</groupId>
<artifactId>java-hamcrest</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-client</artifactId>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index 79b44ef..4d22299 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -23,7 +23,6 @@ import static org.apache.storm.validation.ConfigValidationAnnotations.isString;
import static org.apache.storm.validation.ConfigValidationAnnotations.isStringList;
import static org.apache.storm.validation.ConfigValidationAnnotations.isStringOrStringList;
import static org.apache.storm.validation.ConfigValidationAnnotations.isPositiveNumber;
-import static org.apache.storm.validation.ConfigValidationAnnotations.isType;
import static org.apache.storm.validation.ConfigValidationAnnotations.NotNull;
import static org.apache.storm.validation.ConfigValidationAnnotations.isListEntryCustom;
import static org.apache.storm.validation.ConfigValidationAnnotations.isBoolean;
@@ -37,7 +36,6 @@ import org.apache.storm.container.ResourceIsolationInterface;
import org.apache.storm.nimbus.ITopologyActionNotifierPlugin;
import org.apache.storm.scheduler.blacklist.reporters.IReporter;
import org.apache.storm.scheduler.blacklist.strategies.IBlacklistStrategy;
-import org.apache.storm.scheduler.resource.strategies.eviction.IEvictionStrategy;
import org.apache.storm.scheduler.resource.strategies.priority.ISchedulingPriorityStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.IStrategy;
import org.apache.storm.validation.ConfigValidation;
@@ -879,13 +877,6 @@ public class DaemonConfig implements Validated {
public static final String RESOURCE_AWARE_SCHEDULER_USER_POOLS = "resource.aware.scheduler.user.pools";
/**
- * The class that specifies the eviction strategy to use in ResourceAwareScheduler.
- */
- @NotNull
- @isImplementationOfClass(implementsClass = IEvictionStrategy.class)
- public static final String RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY = "resource.aware.scheduler.eviction.strategy";
-
- /**
* the class that specifies the scheduling priority strategy to use in ResourceAwareScheduler.
*/
@NotNull
@@ -893,6 +884,14 @@ public class DaemonConfig implements Validated {
public static final String RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY = "resource.aware.scheduler.priority.strategy";
/**
+ * The maximum number of times that the RAS will attempt to schedule a topology. The default is 5.
+ */
+ @isInteger
+ @isPositiveNumber
+ public static final String RESOURCE_AWARE_SCHEDULER_MAX_TOPOLOGY_SCHEDULING_ATTEMPTS =
+ "resource.aware.scheduler.max.topology.scheduling.attempts";
+
+ /**
* How often nimbus's background thread to sync code for missing topologies should run.
*/
@isInteger
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index d4f0ee8..d9020e2 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -134,7 +134,7 @@ import org.apache.storm.nimbus.ITopologyActionNotifierPlugin;
import org.apache.storm.nimbus.ITopologyValidator;
import org.apache.storm.nimbus.NimbusInfo;
import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.Cluster.SupervisorResources;
+import org.apache.storm.scheduler.SupervisorResources;
import org.apache.storm.scheduler.DefaultScheduler;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
@@ -971,7 +971,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
if (resourcesMap == null) {
resourcesMap = new HashMap<>();
}
- ResourceUtils.checkIntialization(resourcesMap, compId, topoConf);
+ ResourceUtils.checkInitialization(resourcesMap, compId, topoConf);
return resourcesMap;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
index 6cb53e4..1b30e8d 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.scheduler;
import com.google.common.annotations.VisibleForTesting;
@@ -44,64 +45,6 @@ import org.slf4j.LoggerFactory;
public class Cluster implements ISchedulingState {
private static final Logger LOG = LoggerFactory.getLogger(Cluster.class);
- public static class SupervisorResources {
- private final double totalMem;
- private final double totalCpu;
- private final double usedMem;
- private final double usedCpu;
-
- /**
- * Constructor for a Supervisor's resources.
- *
- * @param totalMem the total mem on the supervisor
- * @param totalCpu the total CPU on the supervisor
- * @param usedMem the used mem on the supervisor
- * @param usedCpu the used CPU on the supervisor
- */
- public SupervisorResources(double totalMem, double totalCpu, double usedMem, double usedCpu) {
- this.totalMem = totalMem;
- this.totalCpu = totalCpu;
- this.usedMem = usedMem;
- this.usedCpu = usedCpu;
- }
-
- public double getUsedMem() {
- return usedMem;
- }
-
- public double getUsedCpu() {
- return usedCpu;
- }
-
- public double getTotalMem() {
- return totalMem;
- }
-
- public double getTotalCpu() {
- return totalCpu;
- }
-
- public double getAvailableCpu() {
- return totalCpu - usedCpu;
- }
-
- public double getAvailableMem() {
- return totalMem - usedMem;
- }
-
- private SupervisorResources add(WorkerResources wr) {
- return new SupervisorResources(
- totalMem,
- totalCpu,
- usedMem + wr.get_mem_off_heap() + wr.get_mem_on_heap(),
- usedCpu + wr.get_cpu());
- }
-
- public SupervisorResources addMem(Double value) {
- return new SupervisorResources(totalMem, totalCpu, usedMem + value, usedCpu);
- }
- }
-
/**
* key: supervisor id, value: supervisor details.
*/
@@ -839,8 +782,8 @@ public class Cluster implements ISchedulingState {
* @param topConf - the topology config
* @return the assigned memory (in MB)
*/
- public static Double getAssignedMemoryForSlot(final Map<String, Object> topConf) {
- Double totalWorkerMemory = 0.0;
+ public static double getAssignedMemoryForSlot(final Map<String, Object> topConf) {
+ double totalWorkerMemory = 0.0;
final Integer topologyWorkerDefaultMemoryAllocation = 768;
List<String> topologyWorkerGcChildopts = ConfigUtils.getValueAsList(
@@ -890,6 +833,13 @@ public class Cluster implements ISchedulingState {
/**
* set scheduler status for a topology.
*/
+ public void setStatus(TopologyDetails td, String statusMessage) {
+ setStatus(td.getId(), statusMessage);
+ }
+
+ /**
+ * set scheduler status for a topology.
+ */
public void setStatus(String topologyId, String statusMessage) {
assertValidTopologyForModification(topologyId);
LOG.info("STATUS - {} {}", topologyId, statusMessage);
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java b/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
index 95e0c86..e37e4b0 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/ISchedulingState.java
@@ -25,7 +25,6 @@ import java.util.Set;
import org.apache.storm.daemon.nimbus.TopologyResources;
import org.apache.storm.generated.WorkerResources;
-import org.apache.storm.scheduler.Cluster.SupervisorResources;
/** An interface that provides access to the current scheduling state. */
public interface ISchedulingState {
@@ -43,7 +42,7 @@ public interface ISchedulingState {
List<TopologyDetails> needsSchedulingTopologies();
/**
- * Does the topology need scheduling?
+ * Does the topology need scheduling.
*
* <p>A topology needs scheduling if one of the following conditions holds:
*
@@ -103,12 +102,14 @@ public interface ISchedulingState {
Collection<ExecutorDetails> getUnassignedExecutors(TopologyDetails topology);
/**
+ * Get the executor to component name map for executors that need to be scheduled.
* @param topology the topology this is for
* @return a executor -> component-id map which needs scheduling in this topology.
*/
Map<ExecutorDetails, String> getNeedsSchedulingExecutorToComponents(TopologyDetails topology);
/**
+ * Get the component name to executor list for executors that need to be scheduled.
* @param topology the topology this is for
* @return a component-id -> executors map which needs scheduling in this topology.
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorResources.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorResources.java b/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorResources.java
new file mode 100644
index 0000000..ac2a506
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/SupervisorResources.java
@@ -0,0 +1,79 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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;
+
+import org.apache.storm.generated.WorkerResources;
+
+public class SupervisorResources {
+ private final double totalMem;
+ private final double totalCpu;
+ private final double usedMem;
+ private final double usedCpu;
+
+ /**
+ * Constructor for a Supervisor's resources.
+ *
+ * @param totalMem the total mem on the supervisor
+ * @param totalCpu the total CPU on the supervisor
+ * @param usedMem the used mem on the supervisor
+ * @param usedCpu the used CPU on the supervisor
+ */
+ public SupervisorResources(double totalMem, double totalCpu, double usedMem, double usedCpu) {
+ this.totalMem = totalMem;
+ this.totalCpu = totalCpu;
+ this.usedMem = usedMem;
+ this.usedCpu = usedCpu;
+ }
+
+ public double getUsedMem() {
+ return usedMem;
+ }
+
+ public double getUsedCpu() {
+ return usedCpu;
+ }
+
+ public double getTotalMem() {
+ return totalMem;
+ }
+
+ public double getTotalCpu() {
+ return totalCpu;
+ }
+
+ public double getAvailableCpu() {
+ return totalCpu - usedCpu;
+ }
+
+ public double getAvailableMem() {
+ return totalMem - usedMem;
+ }
+
+ SupervisorResources add(WorkerResources wr) {
+ return new SupervisorResources(
+ totalMem,
+ totalCpu,
+ usedMem + wr.get_mem_off_heap() + wr.get_mem_on_heap(),
+ usedCpu + wr.get_cpu());
+ }
+
+ public SupervisorResources addMem(Double value) {
+ return new SupervisorResources(totalMem, totalCpu, usedMem + value, usedCpu);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
index 005df7d..326440e 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
@@ -139,7 +139,7 @@ public class TopologyDetails {
//the json_conf is populated by TopologyBuilder (e.g. boltDeclarer.setMemoryLoad)
Map<String, Double> topologyResources =
ResourceUtils.parseResources(bolt.getValue().get_common().get_json_conf());
- ResourceUtils.checkIntialization(topologyResources, bolt.getKey(), topologyConf);
+ ResourceUtils.checkInitialization(topologyResources, bolt.getKey(), topologyConf);
for (Map.Entry<ExecutorDetails, String> anExecutorToComponent :
executorToComponent.entrySet()) {
if (bolt.getKey().equals(anExecutorToComponent.getValue())) {
@@ -153,7 +153,7 @@ public class TopologyDetails {
for (Map.Entry<String, SpoutSpec> spout : topology.get_spouts().entrySet()) {
Map<String, Double> topologyResources =
ResourceUtils.parseResources(spout.getValue().get_common().get_json_conf());
- ResourceUtils.checkIntialization(topologyResources, spout.getKey(), this.topologyConf);
+ ResourceUtils.checkInitialization(topologyResources, spout.getKey(), this.topologyConf);
for (Map.Entry<ExecutorDetails, String> anExecutorToComponent :
executorToComponent.entrySet()) {
if (spout.getKey().equals(anExecutorToComponent.getValue())) {
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
index 669079b..fa7763e 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
@@ -18,22 +18,30 @@
package org.apache.storm.scheduler.resource;
+import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.IScheduler;
+import org.apache.storm.scheduler.SchedulerAssignment;
+import org.apache.storm.scheduler.SchedulerAssignmentImpl;
import org.apache.storm.scheduler.SingleTopologyCluster;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
-import org.apache.storm.scheduler.resource.strategies.eviction.IEvictionStrategy;
+import org.apache.storm.scheduler.WorkerSlot;
import org.apache.storm.scheduler.resource.strategies.priority.ISchedulingPriorityStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.IStrategy;
import org.apache.storm.scheduler.utils.ConfigLoaderFactoryService;
-import org.apache.storm.utils.ReflectionUtils;
-import org.apache.storm.utils.DisallowedStrategyException;
import org.apache.storm.scheduler.utils.IConfigLoader;
+import org.apache.storm.utils.DisallowedStrategyException;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,18 +49,18 @@ import org.slf4j.LoggerFactory;
public class ResourceAwareScheduler implements IScheduler {
private static final Logger LOG = LoggerFactory.getLogger(ResourceAwareScheduler.class);
private Map<String, Object> conf;
- private ISchedulingPriorityStrategy schedulingPrioritystrategy;
- private IEvictionStrategy evictionStrategy;
+ private ISchedulingPriorityStrategy schedulingPriorityStrategy;
private IConfigLoader configLoader;
+ private int maxSchedulingAttempts;
@Override
public void prepare(Map<String, Object> conf) {
this.conf = conf;
- schedulingPrioritystrategy = (ISchedulingPriorityStrategy) ReflectionUtils.newInstance(
+ schedulingPriorityStrategy = ReflectionUtils.newInstance(
(String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY));
- evictionStrategy = (IEvictionStrategy) ReflectionUtils.newInstance(
- (String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY));
configLoader = ConfigLoaderFactoryService.createConfigLoader(conf);
+ maxSchedulingAttempts = ObjectReader.getInt(
+ conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_MAX_TOPOLOGY_SCHEDULING_ATTEMPTS), 5);
}
@Override
@@ -62,123 +70,144 @@ public class ResourceAwareScheduler implements IScheduler {
@Override
public void schedule(Topologies topologies, Cluster cluster) {
- //initialize data structures
- for (TopologyDetails td : cluster.getTopologies()) {
+ Map<String, User> userMap = getUsers(cluster);
+ List<TopologyDetails> orderedTopologies = new ArrayList<>(schedulingPriorityStrategy.getOrderedTopologies(cluster, userMap));
+ LOG.info("Ordered list of topologies is: {}", orderedTopologies.stream().map((t) -> t.getId()).collect(Collectors.toList()));
+ for (TopologyDetails td : orderedTopologies) {
if (!cluster.needsSchedulingRas(td)) {
//cluster forgets about its previous status, so if it is scheduled just leave it.
cluster.setStatusIfAbsent(td.getId(), "Fully Scheduled");
- }
- }
- Map<String, User> userMap = getUsers(cluster);
-
- while (true) {
- TopologyDetails td;
- try {
- //Call scheduling priority strategy
- td = schedulingPrioritystrategy.getNextTopologyToSchedule(cluster, userMap);
- } catch (Exception ex) {
- LOG.error("Exception thrown when running priority strategy {}. No topologies will be scheduled!",
- schedulingPrioritystrategy.getClass().getName(), ex);
- break;
- }
- if (td == null) {
- break;
- }
- User submitter = userMap.get(td.getTopologySubmitter());
- if (cluster.needsSchedulingRas(td)) {
- scheduleTopology(td, cluster, submitter, userMap);
} else {
- LOG.warn("Topology {} is already fully scheduled!", td.getName());
- cluster.setStatusIfAbsent(td.getId(), "Fully Scheduled");
+ User submitter = userMap.get(td.getTopologySubmitter());
+ scheduleTopology(td, cluster, submitter, orderedTopologies);
}
}
}
+ private static void markFailedTopology(User u, Cluster c, TopologyDetails td, String message) {
+ markFailedTopology(u, c, td, message, null);
+ }
- public void scheduleTopology(TopologyDetails td, Cluster cluster, final User topologySubmitter,
- Map<String, User> userMap) {
+ private static void markFailedTopology(User u, Cluster c, TopologyDetails td, String message, Throwable t) {
+ c.setStatus(td, message);
+ String realMessage = td.getId() + " " + message;
+ if (t != null) {
+ LOG.error(realMessage, t);
+ } else {
+ LOG.error(realMessage);
+ }
+ u.markTopoUnsuccess(td);
+ }
+
+ private void scheduleTopology(TopologyDetails td, Cluster cluster, final User topologySubmitter,
+ List<TopologyDetails> orderedTopologies) {
//A copy of cluster that we can modify, but does not get committed back to cluster unless scheduling succeeds
Cluster workingState = new Cluster(cluster);
+ RAS_Nodes nodes = new RAS_Nodes(workingState);
IStrategy rasStrategy = null;
String strategyConf = (String) td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY);
try {
- rasStrategy = (IStrategy) ReflectionUtils.newSchedulerStrategyInstance((String) td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY), conf);
+ rasStrategy = ReflectionUtils.newSchedulerStrategyInstance((String) td.getConf().get(Config.TOPOLOGY_SCHEDULER_STRATEGY), conf);
rasStrategy.prepare(conf);
} catch (DisallowedStrategyException e) {
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling - " + e.getAttemptedClass()
- + " is not an allowed strategy. Please make sure your " + Config.TOPOLOGY_SCHEDULER_STRATEGY
- + " config is one of the allowed strategies: " + e.getAllowedStrategies().toString());
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Unsuccessful in scheduling - " + e.getAttemptedClass()
+ + " is not an allowed strategy. Please make sure your "
+ + Config.TOPOLOGY_SCHEDULER_STRATEGY
+ + " config is one of the allowed strategies: "
+ + e.getAllowedStrategies(), e);
return;
} catch (RuntimeException e) {
- LOG.error("failed to create instance of IStrategy: {} Topology {} will not be scheduled.",
- strategyConf, td.getName(), e);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling - failed to create instance of topology strategy "
- + strategyConf + ". Please check logs for details");
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Unsuccessful in scheduling - failed to create instance of topology strategy "
+ + strategyConf
+ + ". Please check logs for details", e);
return;
}
- while (true) {
- // A copy of the cluster that restricts the strategy to only modify a single topology
+ for (int i = 0; i < maxSchedulingAttempts; i++) {
SingleTopologyCluster toSchedule = new SingleTopologyCluster(workingState, td.getId());
- SchedulingResult result = null;
try {
- result = rasStrategy.schedule(toSchedule, td);
- } catch (Exception ex) {
- LOG.error("Exception thrown when running strategy {} to schedule topology {}."
- + " Topology will not be scheduled!", rasStrategy.getClass().getName(), td.getName(), ex);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling - Exception thrown when running strategy {}"
- + rasStrategy.getClass().getName() + ". Please check logs for details");
- }
- LOG.debug("scheduling result: {}", result);
- if (result != null) {
- if (result.isSuccess()) {
- try {
+ SchedulingResult result = rasStrategy.schedule(toSchedule, td);
+ LOG.debug("scheduling result: {}", result);
+ if (result != null) {
+ if (result.isSuccess()) {
cluster.updateFrom(toSchedule);
cluster.setStatus(td.getId(), "Running - " + result.getMessage());
- } catch (Exception ex) {
- LOG.error("Unsuccessful attempting to assign executors to nodes.", ex);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Unsuccessful in scheduling - "
- + "IllegalStateException thrown when attempting to assign executors to nodes. Please check"
- + " log for details.");
- }
- return;
- } else {
- if (result.getStatus() == SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES) {
- boolean madeSpace = false;
+ //DONE
+ return;
+ } else if (result.getStatus() == SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES) {
+ LOG.info("Not enough resources to schedule {}", td.getName());
+ List<TopologyDetails> reversedList = ImmutableList.copyOf(orderedTopologies).reverse();
try {
- //need to re prepare since scheduling state might have been restored
- madeSpace = evictionStrategy.makeSpaceForTopo(td, workingState, userMap);
+ boolean evictedSomething = false;
+ LOG.debug("attempting to make space for topo {} from user {}", td.getName(), td.getTopologySubmitter());
+ int tdIndex = reversedList.indexOf(td);
+ double cpuNeeded = td.getTotalRequestedCpu();
+ double memoryNeeded = td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap();
+ SchedulerAssignment assignment = cluster.getAssignmentById(td.getId());
+ if (assignment != null) {
+ cpuNeeded -= getCpuUsed(assignment);
+ memoryNeeded -= getMemoryUsed(assignment);
+ }
+ cluster.getTopologyResourcesMap();
+ for (int index = 0; index < tdIndex; index++) {
+ TopologyDetails topologyEvict = reversedList.get(index);
+ SchedulerAssignment evictAssignemnt = workingState.getAssignmentById(topologyEvict.getId());
+ if (evictAssignemnt != null && !evictAssignemnt.getSlots().isEmpty()) {
+ Collection<WorkerSlot> workersToEvict = workingState.getUsedSlotsByTopologyId(topologyEvict.getId());
+
+ LOG.debug("Evicting Topology {} with workers: {} from user {}", topologyEvict.getName(), workersToEvict,
+ topologyEvict.getTopologySubmitter());
+ cpuNeeded -= getCpuUsed(evictAssignemnt);
+ memoryNeeded -= getMemoryUsed(evictAssignemnt);
+ evictedSomething = true;
+ nodes.freeSlots(workersToEvict);
+ if (cpuNeeded <= 0 && memoryNeeded <= 0) {
+ //We evicted enough topologies to have a hope of scheduling, so try it now, and don't evict more
+ // than is needed
+ break;
+ }
+ }
+ }
+
+ if (!evictedSomething) {
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Not enough resources to schedule - " + result.getErrorMessage());
+ return;
+ }
} catch (Exception ex) {
- LOG.error("Exception thrown when running eviction strategy {} to schedule topology {}."
- + " No evictions will be done!", evictionStrategy.getClass().getName(),
- td.getName(), ex);
- topologySubmitter.markTopoUnsuccess(td);
- return;
- }
- if (!madeSpace) {
- LOG.debug("Could not make space for topo {} will move to attempted", td);
- topologySubmitter.markTopoUnsuccess(td);
- cluster.setStatus(td.getId(), "Not enough resources to schedule - "
- + result.getErrorMessage());
- return;
+ LOG.error("Exception thrown when running eviction to schedule topology {}."
+ + " No evictions will be done! Error: {}",
+ td.getName(), ex.getClass().getName(), ex);
}
+ //Only place we fall though to do the loop over again...
continue;
} else {
+ //The assumption is that the strategy set the status...
topologySubmitter.markTopoUnsuccess(td, cluster);
return;
}
+ } else {
+ markFailedTopology(topologySubmitter, cluster, td, "Internal scheduler error");
+ return;
}
- } else {
- LOG.warn("Scheduling results returned from topology {} is not vaild! Topology with be ignored.",
- td.getName());
- topologySubmitter.markTopoUnsuccess(td, cluster);
+ } catch (Exception ex) {
+ markFailedTopology(topologySubmitter, cluster, td,
+ "Internal Error - Exception thrown when scheduling. Please check logs for details", ex);
return;
}
}
+ markFailedTopology(topologySubmitter, cluster, td, "Failed to schedule within " + maxSchedulingAttempts + " attempts");
+ }
+
+ private static double getCpuUsed(SchedulerAssignment assignment) {
+ return assignment.getScheduledResources().values().stream().mapToDouble((wr) -> wr.get_cpu()).sum();
+ }
+
+ private static double getMemoryUsed(SchedulerAssignment assignment) {
+ return assignment.getScheduledResources().values().stream()
+ .mapToDouble((wr) -> wr.get_mem_on_heap() + wr.get_mem_off_heap()).sum();
}
/**
@@ -206,12 +235,12 @@ public class ResourceAwareScheduler implements IScheduler {
}
private Map<String, Map<String, Double>> convertToDouble(Map<String, Map<String, Number>> raw) {
- Map<String, Map<String, Double>> ret = new HashMap<String, Map<String, Double>>();
+ Map<String, Map<String, Double>> ret = new HashMap<>();
if (raw != null) {
for (Map.Entry<String, Map<String, Number>> userPoolEntry : raw.entrySet()) {
String user = userPoolEntry.getKey();
- ret.put(user, new HashMap<String, Double>());
+ ret.put(user, new HashMap<>());
for (Map.Entry<String, Number> resourceEntry : userPoolEntry.getValue().entrySet()) {
ret.get(user).put(resourceEntry.getKey(), resourceEntry.getValue().doubleValue());
}
@@ -226,7 +255,7 @@ public class ResourceAwareScheduler implements IScheduler {
* Load from configLoaders first; if no config available, read from user-resource-pools.yaml;
* if no config available from user-resource-pools.yaml, get configs from conf. Only one will be used.
* @return a map that contains resource guarantees of every user of the following format
- * {userid->{resourceType->amountGuaranteed}}
+ * {userid->{resourceType->amountGuaranteed}}
*/
private Map<String, Map<String, Double>> getUserResourcePools() {
Map<String, Map<String, Number>> raw;
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
index ab8eb14..44403ef 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
@@ -43,8 +43,7 @@ public class ResourceUtils {
if (topology.get_bolts() != null) {
for (Map.Entry<String, Bolt> bolt : topology.get_bolts().entrySet()) {
Map<String, Double> topologyResources = parseResources(bolt.getValue().get_common().get_json_conf());
- checkIntialization(topologyResources, bolt.getValue().toString(), topologyConf);
- LOG.warn("Turned {} into {}", bolt.getValue().get_common().get_json_conf(), topologyResources);
+ checkInitialization(topologyResources, bolt.getKey(), topologyConf);
boltResources.put(bolt.getKey(), topologyResources);
}
}
@@ -57,7 +56,7 @@ public class ResourceUtils {
if (topology.get_spouts() != null) {
for (Map.Entry<String, SpoutSpec> spout : topology.get_spouts().entrySet()) {
Map<String, Double> topologyResources = parseResources(spout.getValue().get_common().get_json_conf());
- checkIntialization(topologyResources, spout.getValue().toString(), topologyConf);
+ checkInitialization(topologyResources, spout.getKey(), topologyConf);
spoutResources.put(spout.getKey(), topologyResources);
}
}
@@ -129,20 +128,23 @@ public class ResourceUtils {
}
}
- public static void checkIntialization(Map<String, Double> topologyResources, String com,
- Map<String, Object> topologyConf) {
+ public static void checkInitialization(Map<String, Double> topologyResources, String com,
+ Map<String, Object> topologyConf) {
checkInitMem(topologyResources, com, topologyConf);
checkInitCpu(topologyResources, com, topologyConf);
}
private static void checkInitMem(Map<String, Double> topologyResources, String com,
- Map<String, Object> topologyConf) {
+ Map<String, Object> topologyConf) {
if (!topologyResources.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB)) {
Double onHeap = ObjectReader.getDouble(
topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null);
if (onHeap != null) {
topologyResources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, onHeap);
- debugMessage("ONHEAP", com, topologyConf);
+ LOG.debug(
+ "Unable to extract resource requirement for Component {}\n"
+ + " Resource : Memory Type : On Heap set to default {}",
+ com, onHeap);
}
}
if (!topologyResources.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB)) {
@@ -150,7 +152,10 @@ public class ResourceUtils {
topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null);
if (offHeap != null) {
topologyResources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, offHeap);
- debugMessage("OFFHEAP", com, topologyConf);
+ LOG.debug(
+ "Unable to extract resource requirement for Component {}\n"
+ + " Resource : Memory Type : Off Heap set to default {}",
+ com, offHeap);
}
}
}
@@ -161,7 +166,10 @@ public class ResourceUtils {
Double cpu = ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null);
if (cpu != null) {
topologyResources.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, cpu);
- debugMessage("CPU", com, topologyConf);
+ LOG.debug(
+ "Unable to extract resource requirement for Component {}\n"
+ + " Resource : CPU Pcore Percent set to default {}",
+ com, cpu);
}
}
}
@@ -198,25 +206,6 @@ public class ResourceUtils {
return topologyResources;
}
- private static void debugMessage(String memoryType, String com, Map<String, Object> topologyConf) {
- if (memoryType.equals("ONHEAP")) {
- LOG.debug(
- "Unable to extract resource requirement for Component {}\n"
- + " Resource : Memory Type : On Heap set to default {}",
- com, topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB));
- } else if (memoryType.equals("OFFHEAP")) {
- LOG.debug(
- "Unable to extract resource requirement for Component {}\n"
- + " Resource : Memory Type : Off Heap set to default {}",
- com, topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB));
- } else {
- LOG.debug(
- "Unable to extract resource requirement for Component {}\n"
- + " Resource : CPU Pcore Percent set to default {}",
- com, topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
- }
- }
-
/**
* Calculate the sum of a collection of doubles.
* @param list collection of doubles
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/User.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/User.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/User.java
index 4850ebb..9083510 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/User.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/User.java
@@ -34,7 +34,7 @@ import org.apache.storm.scheduler.TopologyDetails;
public class User {
private String userId;
- //Topologies that was deemed to be invalid
+ //Topologies that were deemed to be invalid
private final Set<TopologyDetails> unsuccess = new HashSet<>();
private final double cpuGuarantee;
@@ -63,7 +63,7 @@ public class User {
public TreeSet<TopologyDetails> getRunningTopologies(ISchedulingState cluster) {
TreeSet<TopologyDetails> ret =
- new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
+ new TreeSet<>(new PQsortByPriorityAndSubmittionTime());
for (TopologyDetails td : cluster.getTopologies().getTopologiesOwnedBy(userId)) {
if (!cluster.needsSchedulingRas(td)) {
ret.add(td);
@@ -74,7 +74,7 @@ public class User {
public TreeSet<TopologyDetails> getPendingTopologies(ISchedulingState cluster) {
TreeSet<TopologyDetails> ret =
- new TreeSet<TopologyDetails>(new PQsortByPriorityAndSubmittionTime());
+ new TreeSet<>(new PQsortByPriorityAndSubmittionTime());
for (TopologyDetails td : cluster.getTopologies().getTopologiesOwnedBy(userId)) {
if (cluster.needsSchedulingRas(td) && !unsuccess.contains(td)) {
ret.add(td);
@@ -118,6 +118,24 @@ public class User {
return getMemoryResourceUsedByUser(cluster) / memoryGuarantee;
}
+ public double getMemoryResourceRequest(ISchedulingState cluster) {
+ double sum = 0.0;
+ Set<TopologyDetails> topologyDetailsSet = new HashSet<>(cluster.getTopologies().getTopologiesOwnedBy(userId));
+ for (TopologyDetails topo : topologyDetailsSet) {
+ sum += topo.getTotalRequestedMemOnHeap() + topo.getTotalRequestedMemOffHeap();
+ }
+ return sum;
+ }
+
+ public double getCpuResourceRequest(ISchedulingState cluster) {
+ double sum = 0.0;
+ Set<TopologyDetails> topologyDetailsSet = new HashSet<>(cluster.getTopologies().getTopologiesOwnedBy(userId));
+ for (TopologyDetails topo : topologyDetailsSet) {
+ sum += topo.getTotalRequestedCpu();
+ }
+ return sum;
+ }
+
public double getCpuResourceUsedByUser(ISchedulingState cluster) {
double sum = 0.0;
for (TopologyDetails td : cluster.getTopologies().getTopologiesOwnedBy(userId)) {
@@ -187,22 +205,6 @@ public class User {
return this.userId;
}
- public String getDetailedInfo() {
- String ret = "\nUser: " + userId;
- ret += "\n - " + " Resource Pool: " + cpuGuarantee + "% " + memoryGuarantee + "MB ";
- ret += "\n - " + " Unsuccess Queue: " + unsuccess + " size: " + unsuccess.size();
- return ret;
- }
-
- public static String getResourcePoolAverageUtilizationForUsers(
- Collection<User> users, Cluster cluster) {
- String ret = "";
- for (User user : users) {
- ret += user.getId() + " - " + user.getResourcePoolAverageUtilization(cluster) + " ";
- }
- return ret;
- }
-
/**
* Comparator that sorts topologies by priority and then by submission time First sort by Topology
* Priority, if there is a tie for topology priority, topology uptime is used to sort.
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/DefaultEvictionStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/DefaultEvictionStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/DefaultEvictionStrategy.java
deleted file mode 100644
index 3abc65e..0000000
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/DefaultEvictionStrategy.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/**
- * 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 java.util.Collection;
-import java.util.Map;
-
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.TopologyDetails;
-import org.apache.storm.scheduler.WorkerSlot;
-import org.apache.storm.scheduler.resource.RAS_Nodes;
-import org.apache.storm.scheduler.resource.User;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DefaultEvictionStrategy implements IEvictionStrategy {
- private static final Logger LOG = LoggerFactory
- .getLogger(DefaultEvictionStrategy.class);
-
- private Cluster cluster;
- private Map<String, User> userMap;
- private RAS_Nodes nodes;
-
- @Override
- public boolean makeSpaceForTopo(TopologyDetails td, Cluster schedulingState, Map<String, User> userMap) {
- this.cluster = schedulingState;
- this.userMap = userMap;
- this.nodes = new RAS_Nodes(schedulingState);
- LOG.debug("attempting to make space for topo {} from user {}", td.getName(), td.getTopologySubmitter());
- User submitter = this.userMap.get(td.getTopologySubmitter());
- if (submitter.getCpuResourceGuaranteed() == 0.0 || submitter.getMemoryResourceGuaranteed() == 0.0) {
- return false;
- }
-
- double cpuNeeded = td.getTotalRequestedCpu() / submitter.getCpuResourceGuaranteed();
- double memoryNeeded = (td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap())
- / submitter.getMemoryResourceGuaranteed();
-
- User evictUser = this.findUserWithHighestAverageResourceUtilAboveGuarantee();
- //check if user has enough resource under his or her resource guarantee to schedule topology
- if ((1.0 - submitter.getCpuResourcePoolUtilization(schedulingState)) >= cpuNeeded
- && (1.0 - submitter.getMemoryResourcePoolUtilization(schedulingState)) >= memoryNeeded) {
- if (evictUser != null) {
- TopologyDetails topologyEvict = evictUser.getRunningTopologyWithLowestPriority(schedulingState);
- evictTopology(topologyEvict);
- return true;
- }
- } else {
- if (evictUser != null) {
- if ((evictUser.getResourcePoolAverageUtilization(schedulingState) - 1.0)
- > (((cpuNeeded + memoryNeeded) / 2)
- + (submitter.getResourcePoolAverageUtilization(schedulingState) - 1.0))) {
- TopologyDetails topologyEvict = evictUser.getRunningTopologyWithLowestPriority(schedulingState);
- evictTopology(topologyEvict);
- return true;
- }
- }
- }
- //See if there is a lower priority topology that can be evicted from the current user
- //topologies should already be sorted in order of increasing priority.
- //Thus, topology at the front of the queue has the lowest priority
- for (TopologyDetails topo : submitter.getRunningTopologies(schedulingState)) {
- //check to if there is a topology with a lower priority we can evict
- if (topo.getTopologyPriority() > td.getTopologyPriority()) {
- LOG.debug("POTENTIALLY Evicting Topology {} from user {} (itself) since topology {} has a lower "
- + "priority than topology {}", topo, submitter, topo, td);
- evictTopology(topo);
- return true;
- }
- }
- return false;
- }
-
- private void evictTopology(TopologyDetails topologyEvict) {
- Collection<WorkerSlot> workersToEvict = this.cluster.getUsedSlotsByTopologyId(topologyEvict.getId());
-
- LOG.info("Evicting Topology {} with workers: {} from user {}", topologyEvict.getName(), workersToEvict,
- topologyEvict.getTopologySubmitter());
- this.nodes.freeSlots(workersToEvict);
- }
-
- private User findUserWithHighestAverageResourceUtilAboveGuarantee() {
- double most = 0.0;
- User mostOverUser = null;
- for (User user : this.userMap.values()) {
- double over = user.getResourcePoolAverageUtilization(cluster) - 1.0;
- if ((over > most) && (!user.getRunningTopologies(cluster).isEmpty())) {
- most = over;
- mostOverUser = user;
- }
- }
- return mostOverUser;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/IEvictionStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/IEvictionStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/IEvictionStrategy.java
deleted file mode 100644
index 2947f7a..0000000
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/eviction/IEvictionStrategy.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/**
- * 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 java.util.Map;
-
-import org.apache.storm.scheduler.Cluster;
-import org.apache.storm.scheduler.TopologyDetails;
-import org.apache.storm.scheduler.resource.User;
-
-public interface IEvictionStrategy {
- /**
- * This method when invoked should attempt to make space on the cluster so that the topology specified can be scheduled
- * @param td the topology to make space for
- * @return return true to indicate that space has been made for topology and try schedule topology td again.
- * Return false to indicate that no space could be made for the topology on the cluster and the scheduler should give up
- * trying to schedule the topology for this round of scheduling. This method will be invoked until the topology indicated
- * could be scheduled or the method returns false
- */
- public boolean makeSpaceForTopo(TopologyDetails td, Cluster schedulingState, Map<String, User> userMap);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/DefaultSchedulingPriorityStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/DefaultSchedulingPriorityStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/DefaultSchedulingPriorityStrategy.java
index b6d8ada..8b401aa 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/DefaultSchedulingPriorityStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/DefaultSchedulingPriorityStrategy.java
@@ -18,8 +18,12 @@
package org.apache.storm.scheduler.resource.strategies.priority;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
-
import org.apache.storm.scheduler.ISchedulingState;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.resource.User;
@@ -27,45 +31,138 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DefaultSchedulingPriorityStrategy implements ISchedulingPriorityStrategy {
- private static final Logger LOG = LoggerFactory
- .getLogger(DefaultSchedulingPriorityStrategy.class);
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultSchedulingPriorityStrategy.class);
+
+ protected static class SimulatedUser {
+ protected final LinkedList<TopologyDetails> tds = new LinkedList<>();
+ private double assignedCpu = 0.0;
+ private double assignedMemory = 0.0;
+ public final double guaranteedCpu;
+ public final double guaranteedMemory;
+
+ public SimulatedUser(User other, ISchedulingState cluster) {
+ tds.addAll(cluster.getTopologies().getTopologiesOwnedBy(other.getId()));
+ Collections.sort(tds, new TopologyByPriorityAndSubmissionTimeComparator());
+ Double guaranteedCpu = other.getCpuResourceGuaranteed();
+ if (guaranteedCpu == null) {
+ guaranteedCpu = 0.0;
+ }
+ this.guaranteedCpu = guaranteedCpu;
+ Double guaranteedMemory = other.getMemoryResourceGuaranteed();
+ if (guaranteedMemory == null) {
+ guaranteedMemory = 0.0;
+ }
+ this.guaranteedMemory = guaranteedMemory;
+ }
+
+ public TopologyDetails getNextHighest() {
+ return tds.peekFirst();
+ }
+
+ public TopologyDetails simScheduleNextHighest() {
+ TopologyDetails td = tds.pop();
+ assignedCpu += td.getTotalRequestedCpu();
+ assignedMemory += td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap();
+ return td;
+ }
+
+ /**
+ * Get a score for the simulated user. This is used to sort the users, by their highest priority topology.
+ * The only requirement is that if the user is over their guarantees, or there are no available resources the
+ * returned score will be > 0. If they are under their guarantee it must be negative.
+ * @param availableCpu available CPU on the cluster.
+ * @param availableMemory available memory on the cluster.
+ * @param td the topology we are looking at.
+ * @return the score.
+ */
+ protected double getScore(double availableCpu, double availableMemory, TopologyDetails td) {
+ //(Requested + Assigned - Guaranteed)/Available
+ if (td == null || availableCpu <= 0 || availableMemory <= 0) {
+ return Double.MAX_VALUE;
+ }
+ double wouldBeCpu = assignedCpu + td.getTotalRequestedCpu();
+ double wouldBeMem = assignedMemory + td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap();
+ double cpuScore = (wouldBeCpu - guaranteedCpu)/availableCpu;
+ double memScore = (wouldBeMem - guaranteedMemory)/availableMemory;
+ return Math.max(cpuScore, memScore);
+ }
+
+ public double getScore(double availableCpu, double availableMemory) {
+ TopologyDetails td = getNextHighest();
+ return getScore(availableCpu, availableMemory, td);
+ }
+
+ }
+
+ protected SimulatedUser getSimulatedUserFor(User u, ISchedulingState cluster) {
+ return new SimulatedUser(u, cluster);
+ }
@Override
- public TopologyDetails getNextTopologyToSchedule(ISchedulingState schedulingState, Map<String, User> userMap) {
- User nextUser = getNextUser(schedulingState, userMap);
- if (nextUser == null) {
- return null;
+ public List<TopologyDetails> getOrderedTopologies(ISchedulingState cluster, Map<String, User> userMap) {
+ double cpuAvail = cluster.getClusterTotalCpuResource();
+ double memAvail = cluster.getClusterTotalMemoryResource();
+
+ List<TopologyDetails> allUserTopologies = new ArrayList<>();
+ List<SimulatedUser> users = new ArrayList<>();
+ for (User u : userMap.values()) {
+ users.add(getSimulatedUserFor(u, cluster));
+ }
+ while (!users.isEmpty()) {
+ Collections.sort(users, new SimulatedUserComparator(cpuAvail, memAvail));
+ SimulatedUser u = users.get(0);
+ TopologyDetails td = u.getNextHighest();
+ if (td == null) {
+ users.remove(0);
+ } else {
+ double score = u.getScore(cpuAvail, memAvail);
+ td = u.simScheduleNextHighest();
+ LOG.info("SIM Scheduling {} with score of {}", td.getId(), score);
+ cpuAvail -= td.getTotalRequestedCpu();
+ memAvail -= (td.getTotalRequestedMemOffHeap() + td.getTotalRequestedMemOnHeap());
+ allUserTopologies.add(td);
+ }
}
- return nextUser.getNextTopologyToSchedule(schedulingState);
+ return allUserTopologies;
}
- public User getNextUser(ISchedulingState cluster, Map<String, User> userMap) {
- double least = Double.POSITIVE_INFINITY;
- User ret = null;
- final double totalCpu = cluster.getClusterTotalCpuResource();
- final double totalMem = cluster.getClusterTotalMemoryResource();
- for (User user : userMap.values()) {
- if (user.hasTopologyNeedSchedule(cluster)) {
- double userResourcePoolAverageUtilization = user.getResourcePoolAverageUtilization(cluster);
- if (least > userResourcePoolAverageUtilization) {
- ret = user;
- least = userResourcePoolAverageUtilization;
- } else if (Math.abs(least - userResourcePoolAverageUtilization) < 0.0001) {
- // if ResourcePoolAverageUtilization is equal to the user that is being compared
- double currentCpuPercentage = ret.getCpuResourceGuaranteed() / totalCpu;
- double currentMemoryPercentage = ret.getMemoryResourceGuaranteed() / totalMem;
- double currentAvgPercentage = (currentCpuPercentage + currentMemoryPercentage) / 2.0;
-
- double userCpuPercentage = user.getCpuResourceGuaranteed() / totalCpu;
- double userMemoryPercentage = user.getMemoryResourceGuaranteed() / totalMem;
- double userAvgPercentage = (userCpuPercentage + userMemoryPercentage) / 2.0;
- if (userAvgPercentage > currentAvgPercentage) {
- ret = user;
- least = userResourcePoolAverageUtilization;
- }
+ private static class SimulatedUserComparator implements Comparator<SimulatedUser> {
+
+ private final double cpuAvail;
+ private final double memAvail;
+
+ private SimulatedUserComparator(double cpuAvail, double memAvail) {
+ this.cpuAvail = cpuAvail;
+ this.memAvail = memAvail;
+ }
+
+ @Override
+ public int compare(SimulatedUser o1, SimulatedUser o2) {
+ return Double.compare(o1.getScore(cpuAvail, memAvail), o2.getScore(cpuAvail, memAvail));
+ }
+ }
+
+ /**
+ * Comparator that sorts topologies by priority and then by submission time
+ * First sort by Topology Priority, if there is a tie for topology priority, topology uptime is used to sort
+ */
+ private static class TopologyByPriorityAndSubmissionTimeComparator implements Comparator<TopologyDetails> {
+
+ @Override
+ public int compare(TopologyDetails topo1, TopologyDetails topo2) {
+ if (topo1.getTopologyPriority() > topo2.getTopologyPriority()) {
+ return 1;
+ } else if (topo1.getTopologyPriority() < topo2.getTopologyPriority()) {
+ return -1;
+ } else {
+ if (topo1.getUpTime() > topo2.getUpTime()) {
+ return -1;
+ } else if (topo1.getUpTime() < topo2.getUpTime()) {
+ return 1;
+ } else {
+ return topo1.getId().compareTo(topo2.getId());
}
}
}
- return ret;
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/FIFOSchedulingPriorityStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/FIFOSchedulingPriorityStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/FIFOSchedulingPriorityStrategy.java
new file mode 100644
index 0000000..1c76b8d
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/FIFOSchedulingPriorityStrategy.java
@@ -0,0 +1,79 @@
+/**
+ * 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.priority;
+
+import java.util.Collections;
+import java.util.Comparator;
+import org.apache.storm.scheduler.ISchedulingState;
+import org.apache.storm.scheduler.TopologyDetails;
+import org.apache.storm.scheduler.resource.User;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FIFOSchedulingPriorityStrategy extends DefaultSchedulingPriorityStrategy {
+ private static final Logger LOG = LoggerFactory.getLogger(FIFOSchedulingPriorityStrategy.class);
+
+ protected static class FIFOSimulatedUser extends SimulatedUser {
+
+ public FIFOSimulatedUser(User other, ISchedulingState cluster) {
+ super(other, cluster);
+ }
+
+ @Override
+ public double getScore(double availableCpu, double availableMemory) {
+ TopologyDetails td = getNextHighest();
+ double origScore = getScore(availableCpu, availableMemory, td);
+ if (origScore < 0) {
+ return origScore;
+ }
+ //Not enough guaranteed use the age of the topology instead.
+ //TODO need a good way to only do this once...
+ Collections.sort(tds, new TopologyBySubmissionTimeComparator());
+ td = getNextHighest();
+ if (td != null) {
+ LOG.info("SCORE FOR {} is {}", td.getId(), td.getUpTime());
+ return td.getUpTime();
+ }
+ return Double.MAX_VALUE;
+ }
+ }
+
+ @Override
+ protected SimulatedUser getSimulatedUserFor(User u, ISchedulingState cluster) {
+ return new FIFOSimulatedUser(u, cluster);
+ }
+
+ /**
+ * Comparator that sorts topologies by priority and then by submission time
+ * First sort by Topology Priority, if there is a tie for topology priority, topology uptime is used to sort
+ */
+ private static class TopologyBySubmissionTimeComparator implements Comparator<TopologyDetails> {
+
+ @Override
+ public int compare(TopologyDetails topo1, TopologyDetails topo2) {
+ if (topo1.getUpTime() > topo2.getUpTime()) {
+ return 1;
+ } else if (topo1.getUpTime() < topo2.getUpTime()) {
+ return -1;
+ } else {
+ return topo1.getId().compareTo(topo2.getId());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/ISchedulingPriorityStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/ISchedulingPriorityStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/ISchedulingPriorityStrategy.java
index 8654bb3..77ce0d9 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/ISchedulingPriorityStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/priority/ISchedulingPriorityStrategy.java
@@ -18,6 +18,7 @@
package org.apache.storm.scheduler.resource.strategies.priority;
+import java.util.List;
import java.util.Map;
import org.apache.storm.scheduler.ISchedulingState;
@@ -25,9 +26,10 @@ import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.resource.User;
public interface ISchedulingPriorityStrategy {
+
/**
- * Gets the next topology to schedule
- * @return return the next topology to schedule. If there is no topologies left to schedule, return null
+ * Prioritize the list of all topologies in the cluster.
+ * @return ordered list of topologies to schedule.
*/
- public TopologyDetails getNextTopologyToSchedule(ISchedulingState schedulingState, Map<String, User> userMap);
+ List<TopologyDetails> getOrderedTopologies(ISchedulingState schedulingState, Map<String, User> userMap);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
index 9f644c9..cb324a3 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/DefaultResourceAwareStrategy.java
@@ -196,12 +196,12 @@ public class DefaultResourceAwareStrategy implements IStrategy {
targetSlot,
nodeToRack(targetNode));
} else {
- LOG.error("Not Enough Resources to schedule Task {}", exec);
+ LOG.error("Not Enough Resources to schedule Task {} - {}", td.getName(), exec);
}
}
/**
- * Find a worker to schedule executor exec on
+ * 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
@@ -473,7 +473,7 @@ public class DefaultResourceAwareStrategy implements IStrategy {
}
TreeSet<ObjectResources> sortedObjectResources =
- new TreeSet<>(( o1, o2) -> {
+ new TreeSet<>((o1, o2) -> {
int execsScheduled1 = existingScheduleFunc.getNumExistingSchedule(o1.id);
int execsScheduled2 = existingScheduleFunc.getNumExistingSchedule(o2.id);
if (execsScheduled1 > execsScheduled2) {
@@ -611,7 +611,8 @@ public class DefaultResourceAwareStrategy implements IStrategy {
* Order executors based on how many in and out connections it will potentially need to make, in descending order.
* 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
+ * 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
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/test/java/org/apache/storm/TestRebalance.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/TestRebalance.java b/storm-server/src/test/java/org/apache/storm/TestRebalance.java
index bd72b3a..1f2e8b1 100644
--- a/storm-server/src/test/java/org/apache/storm/TestRebalance.java
+++ b/storm-server/src/test/java/org/apache/storm/TestRebalance.java
@@ -24,7 +24,6 @@ import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.TopologySummary;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
-import org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy;
import org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.topology.BoltDeclarer;
@@ -57,7 +56,6 @@ public class TestRebalance {
Config conf = new Config();
conf.put(DaemonConfig.STORM_SCHEDULER, ResourceAwareScheduler.class.getName());
- conf.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, DefaultEvictionStrategy.class.getName());
conf.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, DefaultSchedulingPriorityStrategy.class.getName());
conf.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, DefaultResourceAwareStrategy.class.getName());
conf.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10.0);
http://git-wip-us.apache.org/repos/asf/storm/blob/5f0abb21/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
index 0458ca4..4274b45 100644
--- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
+++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
@@ -20,6 +20,8 @@ package org.apache.storm.daemon.nimbus;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.generated.StormTopology;
+import org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy;
+import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.junit.Test;
@@ -36,10 +38,9 @@ public class NimbusTest {
StormTopology stormTopology1 = builder1.createTopology();
Config config1 = new Config();
config1.put(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN, "org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping");
- config1.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY, org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy.class.getName());
- config1.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy.class.getName());
+ config1.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, DefaultSchedulingPriorityStrategy.class.getName());
- config1.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy.class.getName());
+ config1.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, DefaultResourceAwareStrategy.class.getName());
config1.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10.0);
config1.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 0.0);
config1.put(Config.TOPOLOGY_PRIORITY, 0);