You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@myriad.apache.org by mo...@apache.org on 2015/09/26 00:45:55 UTC
[15/50] [abbrv] incubator-myriad git commit: Rebased and added
changes suggested by @pdread100.
Rebased and added changes suggested by @pdread100.
Tested again on 4 node cluster all appears well
Project: http://git-wip-us.apache.org/repos/asf/incubator-myriad/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-myriad/commit/e60bbaa7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-myriad/tree/e60bbaa7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-myriad/diff/e60bbaa7
Branch: refs/heads/master
Commit: e60bbaa7af4dc619e7b9bb8b6d99d95a677ba357
Parents: bfbefc3
Author: DarinJ <no...@thanks.com>
Authored: Fri Jul 10 01:00:12 2015 -0400
Committer: DarinJ <no...@thanks.com>
Committed: Fri Jul 10 01:19:05 2015 -0400
----------------------------------------------------------------------
.../handlers/ResourceOffersEventHandler.java | 113 ++++++++++++-------
1 file changed, 74 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/e60bbaa7/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
index dae3a03..c009ce8 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
@@ -80,6 +80,7 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
NodeTask taskToLaunch = schedulerState
.getTask(pendingTaskId);
NMProfile profile = taskToLaunch.getProfile();
+
if (matches(offer, profile)
&& SchedulerUtils.isUniqueHostname(offer,
schedulerState.getActiveTasks())) {
@@ -124,54 +125,34 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
}
private boolean matches(Offer offer, NMProfile profile) {
- double cpus = -1;
- double mem = -1;
- int ports = 0;
+ Map<String, Object> results = new HashMap<String, Object>(5);
+
for (Resource resource : offer.getResourcesList()) {
- if (resource.getName().equals("cpus")) {
- if (resource.getType().equals(Value.Type.SCALAR)) {
- cpus = resource.getScalar().getValue();
- } else {
- LOGGER.error("Cpus resource was not a scalar: {}", resource
- .getType().toString());
- }
- } else if (resource.getName().equals("mem")) {
- if (resource.getType().equals(Value.Type.SCALAR)) {
- mem = resource.getScalar().getValue();
- } else {
- LOGGER.error("Mem resource was not a scalar: {}", resource
- .getType().toString());
- }
- } else if (resource.getName().equals("disk")) {
- LOGGER.warn("Ignoring disk resources from offer");
- } else if (resource.getName().equals("ports")) {
- if (resource.getType().equals(Value.Type.RANGES)) {
- Value.Ranges ranges = resource.getRanges();
- for (Value.Range range : ranges.getRangeList()) {
- if (range.getBegin() < range.getEnd()) {
- ports += range.getEnd() - range.getBegin() + 1;
- }
- }
- } else {
- LOGGER.error("ports resource was not Ranges: {}", resource
- .getType().toString());
- }
+ if (resourceEvaluators.containsKey(resource.getName())) {
+ resourceEvaluators.get(resource.getName()).eval(resource, results);
} else {
LOGGER.warn("Ignoring unknown resource type: {}",
resource.getName());
}
}
+ double cpus = (Double) results.get("cpus");
+ double mem = (Double) results.get("mem");
+ int ports = (Integer) results.get("ports");
- if (cpus < 0) {
- LOGGER.error("No cpus resource present");
- }
- if (mem < 0) {
- LOGGER.error("No mem resource present");
- }
- if (ports < 0) {
- LOGGER.error("No port resources present");
+ checkResource(cpus < 0, "cpus");
+ checkResource(mem < 0, "mem");
+ checkResource(ports < 0, "port");
+
+ return checkAggregates(offer, profile, ports, cpus, mem);
+ }
+
+ private void checkResource(boolean fail, String resource) {
+ if (fail) {
+ LOGGER.info("No " + resource + " resources present");
}
+ }
+ private boolean checkAggregates(Offer offer, NMProfile profile, int ports, double cpus, double mem) {
Map<String, String> requestAttributes = new HashMap<>();
if (taskUtils.getAggregateCpus(profile) <= cpus
@@ -185,4 +166,58 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
return false;
}
}
+
+ private static Double scalarToDouble(Resource resource, String id) {
+ Double value = new Double(0.0);
+ if (resource.getType().equals(Value.Type.SCALAR)) {
+ value = new Double(resource.getScalar().getValue());
+ } else {
+ LOGGER.error(id + " resource was not a scalar: {}", resource
+ .getType().toString());
+ }
+ return value;
+ }
+
+ private interface EvalResources {
+ public void eval(Resource resource, Map<String, Object>results);
+ }
+
+ private static Map<String, EvalResources> resourceEvaluators;
+
+ static {
+ resourceEvaluators = new HashMap<String, EvalResources>(4);
+ resourceEvaluators.put("cpus", new EvalResources() {
+ public void eval(Resource resource, Map<String, Object> results) {
+ results.put("cpus", scalarToDouble(resource, "cpus"));
+ }
+ });
+ resourceEvaluators.put("mem", new EvalResources() {
+ public void eval(Resource resource, Map<String, Object> results) {
+ results.put("mem", scalarToDouble(resource, "mem"));
+ }
+ });
+ resourceEvaluators.put("disk", new EvalResources() {
+ public void eval(Resource resource, Map<String, Object> results) {
+ }
+ });
+ resourceEvaluators.put("ports", new EvalResources() {
+ public void eval(Resource resource, Map<String, Object> results) {
+ int ports = 0;
+ if (resource.getType().equals(Value.Type.RANGES)) {
+ Value.Ranges ranges = resource.getRanges();
+ for (Value.Range range : ranges.getRangeList()) {
+ if (range.getBegin() < range.getEnd()) {
+ ports += range.getEnd() - range.getBegin() + 1;
+ }
+ }
+
+ } else {
+ LOGGER.error("ports resource was not Ranges: {}", resource
+ .getType().toString());
+
+ }
+ results.put("ports", Integer.valueOf(ports));
+ }
+ });
+ }
}