Added feature: Mesos now dynamically configures ports of the Node Manager. This addresses Issue #40 completely and Issue #45 partially.
Done by first checking if four ports are available in the resource offer, if so they are added the taskInfo object and passed to the executor via NMTaskConfig. Added NMPorts.java, an immutable helper class used by TaskFactory, essentially a Tuple Object. Tested on four node dev cluster, with each slave having a different range of ports offered. Ran various hadoop-examples code. Behaves as expected. Netstat -nl shows all ports requested have services running and one can connect to them. Project: http://git-wip-us.apache.org/repos/asf/incubator-myriad/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-myriad/commit/bfbefc38 Tree: http://git-wip-us.apache.org/repos/asf/incubator-myriad/tree/bfbefc38 Diff: http://git-wip-us.apache.org/repos/asf/incubator-myriad/diff/bfbefc38 Branch: refs/heads/master Commit: bfbefc389af01b13ef73adf6ad621a06cd4918f1 Parents: 014b4a5 Author: dbjohnson1978 <dbjohnson1...@gmail.com> Authored: Tue May 12 18:15:08 2015 -0400 Committer: DarinJ <n...@thanks.com> Committed: Fri Jul 10 01:01:29 2015 -0400 ---------------------------------------------------------------------- docs/myriad-dev.md | 28 ++++++++ .../myriad-remote-distribution-configuration.md | 2 +- .../com/ebay/myriad/executor/NMTaskConfig.java | 37 ++++++++++ .../ebay/myriad/executor/MyriadExecutor.java | 18 ++++- .../java/com/ebay/myriad/scheduler/NMPorts.java | 65 +++++++++++++++++ .../com/ebay/myriad/scheduler/TaskFactory.java | 73 +++++++++++++++++--- .../handlers/ResourceOffersEventHandler.java | 26 +++++-- 7 files changed, 231 insertions(+), 18 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/docs/myriad-dev.md ---------------------------------------------------------------------- diff --git a/docs/myriad-dev.md b/docs/myriad-dev.md index 3ee31ac..16c95e8 100644 --- a/docs/myriad-dev.md +++ b/docs/myriad-dev.md @@ -100,6 +100,23 @@ export MESOS_NATIVE_JAVA_LIBRARY=/usr/local/lib/libmesos.so <name>yarn.nodemanager.resource.memory-mb</name> <value>${nodemanager.resource.memory-mb}</value> </property> +<!--These options enable dynamic port assignment by mesos --> +<property> + <name>yarn.nodemanager.address</name> + <value>${myriad.yarn.nodemanager.address}</value> +</property> +<property> + <name>yarn.nodemanager.webapp.address</name> + <value>${myriad.yarn.nodemanager.webapp.address}</value> +</property> +<property> + <name>yarn.nodemanager.webapp.https.address</name> + <value>${myriad.yarn.nodemanager.webapp.address}</value> +</property> +<property> + <name>yarn.nodemanager.localizer.address</name> + <value>${myriad.yarn.nodemanager.localizer.address}</value> +</property> <!-- Configure Myriad Scheduler here --> <property> @@ -107,6 +124,17 @@ export MESOS_NATIVE_JAVA_LIBRARY=/usr/local/lib/libmesos.so <value>com.ebay.myriad.scheduler.yarn.MyriadFairScheduler</value> <description>One can configure other scehdulers as well from following list: com.ebay.myriad.scheduler.yarn.MyriadCapacityScheduler, com.ebay.myriad.scheduler.yarn.MyriadFifoScheduler</description> </property> + +``` + +* Add following to ```$YARN_HOME/etc/hadoop/mapred-site.xml```: + +```xml +<!--This option enables dynamic port assignment by mesos --> +<property> + <name>mapreduce.shuffle.port</name> + <value>${myriad.mapreduce.shuffle.port}</value> +</property> ``` Optional: If you would like to enable cgroups, please add following to ```yarn-site.xml```: http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/docs/myriad-remote-distribution-configuration.md ---------------------------------------------------------------------- diff --git a/docs/myriad-remote-distribution-configuration.md b/docs/myriad-remote-distribution-configuration.md index 429eaaf..7b5d29d 100644 --- a/docs/myriad-remote-distribution-configuration.md +++ b/docs/myriad-remote-distribution-configuration.md @@ -52,7 +52,7 @@ Create the tarball and place it in hdfs: ``` cd ~ sudo cp -rp /opt/hadoop-2.5.0 . -sudo rm hadoop-2.5.0/etc/hadoop/*.xml +sudo rm hadoop-2.5.0/etc/hadoop/yarn-site.xml sudo tar -zcpf ~/hadoop-2.5.0.tar.gz hadoop-2.5.0 hadoop fs -put ~/hadoop-2.5.0.tar.gz /dist ``` http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/myriad-commons/src/main/java/com/ebay/myriad/executor/NMTaskConfig.java ---------------------------------------------------------------------- diff --git a/myriad-commons/src/main/java/com/ebay/myriad/executor/NMTaskConfig.java b/myriad-commons/src/main/java/com/ebay/myriad/executor/NMTaskConfig.java index 244b552..c733b20 100644 --- a/myriad-commons/src/main/java/com/ebay/myriad/executor/NMTaskConfig.java +++ b/myriad-commons/src/main/java/com/ebay/myriad/executor/NMTaskConfig.java @@ -11,6 +11,11 @@ public class NMTaskConfig { private Long advertisableMem; private String jvmOpts; private Boolean cgroups; + private Long rpcPort; + private Long localizerPort; + private Long webAppHttpPort; + private Long shufflePort; + private Map<String, String> yarnEnvironment; public String getYarnHome() { @@ -61,4 +66,36 @@ public class NMTaskConfig { this.yarnEnvironment = yarnEnvironment; } + public Long getRpcPort() { + return rpcPort; + } + + public void setRpcPort(long port) { + rpcPort = port; + } + + public Long gettWebAppHttpPort() { + return webAppHttpPort; + } + + public void setWebAppHttpPort(Long port) { + webAppHttpPort = port; + } + + public Long getLocalizerPort() { + return localizerPort; + } + + public void setLocalizerPort(Long localizerPort) { + this.localizerPort = localizerPort; + } + + public Long getShufflePort() { + return shufflePort; + } + + public void setShufflePort(Long shufflePort) { + this.shufflePort = shufflePort; + } + } http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/myriad-executor/src/main/java/com/ebay/myriad/executor/MyriadExecutor.java ---------------------------------------------------------------------- diff --git a/myriad-executor/src/main/java/com/ebay/myriad/executor/MyriadExecutor.java b/myriad-executor/src/main/java/com/ebay/myriad/executor/MyriadExecutor.java index 4b1a393..42535c5 100644 --- a/myriad-executor/src/main/java/com/ebay/myriad/executor/MyriadExecutor.java +++ b/myriad-executor/src/main/java/com/ebay/myriad/executor/MyriadExecutor.java @@ -12,7 +12,6 @@ import org.apache.mesos.Protos.TaskID; import org.apache.mesos.Protos.TaskInfo; import org.apache.mesos.Protos.TaskState; import org.apache.mesos.Protos.TaskStatus; -import org.apache.mesos.Protos.TaskStatus.Builder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -63,6 +62,16 @@ public class MyriadExecutor implements Executor { public static final String KEY_NM_RESOURCE_MEM_MB = "nodemanager.resource.memory-mb"; + public static final String KEY_NM_ADDRESS = "myriad.yarn.nodemanager.address"; + + public static final String KEY_NM_LOCALIZER_ADDRESS = "myriad.yarn.nodemanager.localizer.address"; + + public static final String KEY_NM_WEBAPP_ADDRESS = "myriad.yarn.nodemanager.webapp.address"; + + public static final String KEY_NM_SHUFFLE_PORT = "myriad.mapreduce.shuffle.port"; + + + /** * Allot 10% more memory to account for JVM overhead. */ @@ -115,7 +124,7 @@ public class MyriadExecutor implements Executor { public void launchTask(final ExecutorDriver driver, final TaskInfo task) { new Thread(new Runnable() { public void run() { - Builder statusBuilder = TaskStatus.newBuilder() + TaskStatus.Builder statusBuilder = TaskStatus.newBuilder() .setTaskId(task.getTaskId()); try { NMTaskConfig taskConfig = GSON.fromJson(task.getData().toStringUtf8(), NMTaskConfig.class); @@ -209,6 +218,11 @@ public class MyriadExecutor implements Executor { } envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_RESOURCE_CPU_VCORES, taskConfig.getAdvertisableCpus() + ""); envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_RESOURCE_MEM_MB, taskConfig.getAdvertisableMem() + ""); + envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_ADDRESS, "0.0.0.0:" + taskConfig.getRpcPort().toString() + ""); + envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_LOCALIZER_ADDRESS, "0.0.0.0:" + taskConfig.getLocalizerPort().toString() + ""); + envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_WEBAPP_ADDRESS, "0.0.0.0:" + taskConfig.gettWebAppHttpPort().toString() + ""); + envNMOptions += String.format(PROPERTY_FORMAT, KEY_NM_SHUFFLE_PORT, taskConfig.getShufflePort().toString() + ""); + return envNMOptions; } http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMPorts.java ---------------------------------------------------------------------- diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMPorts.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMPorts.java new file mode 100644 index 0000000..782fbd1 --- /dev/null +++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMPorts.java @@ -0,0 +1,65 @@ +package com.ebay.myriad.scheduler; + +import com.google.common.base.Preconditions; + +import java.util.HashMap; +import java.util.Map; + +/** + * Helper class for dynamically assigning ports to nodemanager + */ +public class NMPorts { + private static final String NM_RPC_PORT_KEY = "nm.rpc.port"; + private static final String NM_LOCALIZER_PORT_KEY = "nm.localizer.port"; + private static final String NM_WEBAPP_HTTP_PORT_KEY = "nm.webapp.http.port"; + private static final String NM_HTTP_SHUFFLE_PORT_KEY = "nm.http.shuffle.port"; + + private static final String [] NM_PORT_KEYS = { + NM_RPC_PORT_KEY, + NM_LOCALIZER_PORT_KEY, + NM_WEBAPP_HTTP_PORT_KEY, + NM_HTTP_SHUFFLE_PORT_KEY + }; + + private Map<String, Long> portsMap = new HashMap<>(NM_PORT_KEYS.length); + + public NMPorts(Long [] ports){ + Preconditions.checkState(ports.length == NM_PORT_KEYS.length, "NMPorts: array \"ports\" is of unexpected length"); + for (int i = 0; i < NM_PORT_KEYS.length; i++) { + portsMap.put(NM_PORT_KEYS[i], ports[i]); + } + } + + public long getRpcPort() { + return portsMap.get(NM_RPC_PORT_KEY); + } + + public long getLocalizerPort() { + return portsMap.get(NM_LOCALIZER_PORT_KEY); + } + + public long getWebAppHttpPort() { + return portsMap.get(NM_WEBAPP_HTTP_PORT_KEY); + } + + public long getShufflePort() { + return portsMap.get(NM_HTTP_SHUFFLE_PORT_KEY); + } + + public static int expectedNumPorts() { + return NM_PORT_KEYS.length; + } + + /** + * @return a string representation of NMPorts + */ + @Override + public String toString(){ + StringBuilder sb = new StringBuilder().append("{"); + for (String key: NM_PORT_KEYS) { + sb.append(key).append(": ").append(portsMap.get(key).toString()).append(", "); + } + sb.replace(sb.length() - 2, sb.length(), "}"); + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskFactory.java ---------------------------------------------------------------------- diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskFactory.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskFactory.java index 3bf9cfb..0e2ce16 100644 --- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskFactory.java +++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskFactory.java @@ -10,7 +10,6 @@ import com.google.common.base.Strings; import com.google.gson.Gson; import com.google.protobuf.ByteString; import org.apache.hadoop.yarn.conf.YarnConfiguration; -import org.apache.mesos.Protos; import org.apache.mesos.Protos.CommandInfo; import org.apache.mesos.Protos.CommandInfo.URI; import org.apache.mesos.Protos.ExecutorID; @@ -26,6 +25,8 @@ import org.slf4j.LoggerFactory; import javax.inject.Inject; import java.nio.charset.Charset; +import java.util.HashSet; +import java.util.Iterator; import java.util.Objects; /** @@ -57,6 +58,35 @@ public interface TaskFactory { this.taskUtils = taskUtils; } + //Utility function to get the first NMPorts.expectedNumPorts number of ports of an offer + private static NMPorts getPorts(Offer offer) { + HashSet<Long> ports = new HashSet<>(); + for (Resource resource : offer.getResourcesList()){ + if (resource.getName().equals("ports")){ + /* + ranges.getRangeList() returns a list of ranges, each range specifies a begin and end only. + so must loop though each range until we get all ports needed. We exit each loop as soon as all + ports are found so bounded by NMPorts.expectedNumPorts. + */ + Iterator<Value.Range> itr = resource.getRanges().getRangeList().iterator(); + while (itr.hasNext() && ports.size() < NMPorts.expectedNumPorts()) { + Value.Range range = itr.next(); + if (range.getBegin() <= range.getEnd()) { + long i = range.getBegin(); + while (i <= range.getEnd() && ports.size() < NMPorts.expectedNumPorts()) { + ports.add(i); + i++; + } + } + } + } + } + + Preconditions.checkState(ports.size() == NMPorts.expectedNumPorts(), "Not enough ports in offer"); + Long [] portArray = ports.toArray(new Long [ports.size()]); + return new NMPorts(portArray); + } + private static String getFileName(String uri) { int lastSlash = uri.lastIndexOf('/'); if (lastSlash == -1) { @@ -87,7 +117,7 @@ public interface TaskFactory { } } - private Protos.CommandInfo getCommandInfo() { + private CommandInfo getCommandInfo() { MyriadExecutorConfiguration myriadExecutorConfiguration = cfg.getMyriadExecutorConfiguration(); CommandInfo.Builder commandInfo = CommandInfo.newBuilder(); if (myriadExecutorConfiguration.getNodeManagerUri().isPresent()) { @@ -176,6 +206,9 @@ public interface TaskFactory { Objects.requireNonNull(offer, "Offer should be non-null"); Objects.requireNonNull(nodeTask, "NodeTask should be non-null"); + NMPorts ports = getPorts(offer); + LOGGER.debug(ports.toString()); + NMProfile profile = nodeTask.getProfile(); NMTaskConfig nmTaskConfig = new NMTaskConfig(); nmTaskConfig.setAdvertisableCpus(profile.getCpus()); @@ -183,6 +216,10 @@ public interface TaskFactory { NodeManagerConfiguration nodeManagerConfiguration = this.cfg.getNodeManagerConfiguration(); nmTaskConfig.setJvmOpts(nodeManagerConfiguration.getJvmOpts().orNull()); nmTaskConfig.setCgroups(nodeManagerConfiguration.getCgroups().or(Boolean.FALSE)); + nmTaskConfig.setRpcPort(ports.getRpcPort()); + nmTaskConfig.setLocalizerPort(ports.getLocalizerPort()); + nmTaskConfig.setWebAppHttpPort(ports.getWebAppHttpPort()); + nmTaskConfig.setShufflePort(ports.getShufflePort()); nmTaskConfig.setYarnEnvironment(cfg.getYarnEnvironment()); // if RM's hostname is passed in as a system property, pass it along @@ -206,23 +243,22 @@ public interface TaskFactory { String taskConfigJSON = new Gson().toJson(nmTaskConfig); - Scalar taskMemory = Value.Scalar.newBuilder() + Scalar taskMemory = Scalar.newBuilder() .setValue(taskUtils.getTaskMemory(profile)) .build(); - Scalar taskCpus = Value.Scalar.newBuilder() + Scalar taskCpus = Scalar.newBuilder() .setValue(taskUtils.getTaskCpus(profile)) .build(); - Scalar executorMemory = Value.Scalar.newBuilder() + Scalar executorMemory = Scalar.newBuilder() .setValue(taskUtils.getExecutorMemory()) .build(); - Scalar executorCpus = Value.Scalar.newBuilder() + Scalar executorCpus = Scalar.newBuilder() .setValue(taskUtils.getExecutorCpus()) .build(); - CommandInfo commandInfo = getCommandInfo(); - ExecutorID executorId = Protos.ExecutorID.newBuilder() + ExecutorID executorId = ExecutorID.newBuilder() .setValue(EXECUTOR_PREFIX + offer.getSlaveId().getValue()) .build(); ExecutorInfo executorInfo = ExecutorInfo @@ -246,7 +282,6 @@ public interface TaskFactory { .setTaskId(taskId) .setSlaveId(offer.getSlaveId()); - // TODO (mohit): Configure ports for multi-tenancy ByteString data = ByteString.copyFrom(taskConfigJSON.getBytes(Charset.defaultCharset())); return taskBuilder .addResources( @@ -259,6 +294,26 @@ public interface TaskFactory { .setType(Value.Type.SCALAR) .setScalar(taskMemory) .build()) + .addResources( + Resource.newBuilder().setName("ports") + .setType(Value.Type.RANGES) + .setRanges(Value.Ranges.newBuilder() + .addRange(Value.Range.newBuilder() + .setBegin(ports.getRpcPort()) + .setEnd(ports.getRpcPort()) + .build()) + .addRange(Value.Range.newBuilder() + .setBegin(ports.getLocalizerPort()) + .setEnd(ports.getLocalizerPort()) + .build()) + .addRange(Value.Range.newBuilder() + .setBegin(ports.getWebAppHttpPort()) + .setEnd(ports.getWebAppHttpPort()) + .build()) + .addRange(Value.Range.newBuilder() + .setBegin(ports.getShufflePort()) + .setEnd(ports.getShufflePort()) + .build()))) .setExecutor(executorInfo).setData(data).build(); } } http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/bfbefc38/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 165c9dd..dae3a03 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 @@ -15,6 +15,7 @@ */ package com.ebay.myriad.scheduler.event.handlers; +import com.ebay.myriad.scheduler.NMPorts; import com.ebay.myriad.scheduler.NMProfile; import com.ebay.myriad.scheduler.SchedulerUtils; import com.ebay.myriad.scheduler.TaskFactory; @@ -125,7 +126,7 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv private boolean matches(Offer offer, NMProfile profile) { double cpus = -1; double mem = -1; - + int ports = 0; for (Resource resource : offer.getResourcesList()) { if (resource.getName().equals("cpus")) { if (resource.getType().equals(Value.Type.SCALAR)) { @@ -144,7 +145,17 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv } else if (resource.getName().equals("disk")) { LOGGER.warn("Ignoring disk resources from offer"); } else if (resource.getName().equals("ports")) { - LOGGER.info("Ignoring ports resources from offer"); + 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()); + } } else { LOGGER.warn("Ignoring unknown resource type: {}", resource.getName()); @@ -157,18 +168,21 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv if (mem < 0) { LOGGER.error("No mem resource present"); } + if (ports < 0) { + LOGGER.error("No port resources present"); + } Map<String, String> requestAttributes = new HashMap<>(); if (taskUtils.getAggregateCpus(profile) <= cpus && taskUtils.getAggregateMemory(profile) <= mem - && SchedulerUtils.isMatchSlaveAttributes(offer, - requestAttributes)) { + && SchedulerUtils.isMatchSlaveAttributes(offer, requestAttributes) + && NMPorts.expectedNumPorts() <= ports) { return true; } else { - LOGGER.info("Offer not sufficient for task with, cpu: {}, memory: {}", taskUtils.getAggregateCpus(profile), taskUtils.getAggregateMemory(profile)); + LOGGER.info("Offer not sufficient for task with, cpu: {}, memory: {}, ports: {}", + taskUtils.getAggregateCpus(profile), taskUtils.getAggregateMemory(profile), ports); return false; } } - }
