myriad-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From smare...@apache.org
Subject [13/20] incubator-myriad git commit: com.ebay => org.apache
Date Wed, 28 Oct 2015 16:07:45 GMT
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/DownloadNMExecutorCLGenImpl.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/DownloadNMExecutorCLGenImpl.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/DownloadNMExecutorCLGenImpl.java
new file mode 100644
index 0000000..23e9798
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/DownloadNMExecutorCLGenImpl.java
@@ -0,0 +1,97 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+/**
+ * Implementation assumes NM binaries will be downloaded
+ */
+public class DownloadNMExecutorCLGenImpl extends NMExecutorCLGenImpl {
+
+  private static final Logger LOGGER = LoggerFactory.
+      getLogger(DownloadNMExecutorCLGenImpl.class);
+
+  private final String nodeManagerUri;
+
+  public DownloadNMExecutorCLGenImpl(MyriadConfiguration cfg, String nodeManagerUri) {
+    super(cfg);
+    this.nodeManagerUri = nodeManagerUri;
+  }
+
+  @Override
+  public String generateCommandLine(ServiceResourceProfile profile, Ports ports) {
+    StringBuilder cmdLine = new StringBuilder();
+    LOGGER.info("Using remote distribution");
+
+    generateEnvironment(profile, (NMPorts) ports);
+    appendDistroExtractionCommands(cmdLine);
+    appendCgroupsCmds(cmdLine);
+    appendYarnHomeExport(cmdLine);
+    appendUser(cmdLine);
+    appendEnvForNM(cmdLine);
+    cmdLine.append(YARN_NM_CMD);
+    return cmdLine.toString();
+  }
+
+  protected void appendDistroExtractionCommands(StringBuilder cmdLine) {
+    /*
+    TODO(darinj): Overall this is messier than I'd like. We can't let mesos untar the distribution, since
+    it will change the permissions.  Instead we simply download the tarball and execute tar -xvpf. We also
+    pull the config from the resource manager and put them in the conf dir.  This is also why we need
+    frameworkSuperUser. This will be refactored after Mesos-1790 is resolved.
+   */
+
+    //TODO(DarinJ) support other compression, as this is a temp fix for Mesos 1760 may not get to it.
+    //Extract tarball keeping permissions, necessary to keep HADOOP_HOME/bin/container-executor suidbit set.
+    cmdLine.append("sudo tar -zxpf ").append(getFileName(nodeManagerUri));
+
+    //We need the current directory to be writable by frameworkUser for capsuleExecutor to create directories.
+    //Best to simply give owenership to the user running the executor but we don't want to use -R as this
+    //will silently remove the suid bit on container executor.
+    cmdLine.append(" && sudo chown ").append(cfg.getFrameworkUser().get()).append(" .");
+
+    //Place the hadoop config where in the HADOOP_CONF_DIR where it will be read by the NodeManager
+    //The url for the resource manager config is: http(s)://hostname:port/conf so fetcher.cpp downloads the
+    //config file to conf, It's an xml file with the parameters of yarn-site.xml, core-site.xml and hdfs.xml.
+    cmdLine.append(" && cp conf ").append(cfg.getYarnEnvironment().get("YARN_HOME")).append("/etc/hadoop/yarn-site.xml;");
+  }
+
+  protected void appendUser(StringBuilder cmdLine) {
+    cmdLine.append(" sudo -E -u ").append(cfg.getFrameworkUser().get()).append(" -H");
+  }
+
+  private static String getFileName(String uri) {
+    int lastSlash = uri.lastIndexOf('/');
+    if (lastSlash == -1) {
+      return uri;
+    } else {
+      String fileName = uri.substring(lastSlash + 1);
+      Preconditions.checkArgument(!Strings.isNullOrEmpty(fileName), "URI should not have a slash at the end");
+      return fileName;
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExecutorCommandLineGenerator.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExecutorCommandLineGenerator.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExecutorCommandLineGenerator.java
new file mode 100644
index 0000000..82782f2
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExecutorCommandLineGenerator.java
@@ -0,0 +1,29 @@
+/**
+ * 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.myriad.scheduler;
+
+/**
+ * Interface to plugin multiple implementations for executor command generation
+ */
+public interface ExecutorCommandLineGenerator {
+  String generateCommandLine(ServiceResourceProfile profile, Ports ports);
+
+  String getConfigurationUrl();
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExtendedResourceProfile.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExtendedResourceProfile.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExtendedResourceProfile.java
new file mode 100644
index 0000000..8119360
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ExtendedResourceProfile.java
@@ -0,0 +1,80 @@
+/**
+ * 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.myriad.scheduler;
+
+import com.google.gson.Gson;
+
+/**
+ * Extended ServiceResourceProfile for services that need to pass set of resources downstream
+ * currently the only such service is NodeManager
+ */
+public class ExtendedResourceProfile extends ServiceResourceProfile {
+
+  private NMProfile childProfile;
+
+  /**
+   * @param childProfile - should be null
+   * @param cpu
+   * @param mem          will throw NullPoiterException if childProfile is null
+   */
+  public ExtendedResourceProfile(NMProfile childProfile, Double cpu, Double mem) {
+    super(childProfile.getName(), cpu, mem);
+    this.childProfile = childProfile;
+    this.className = ExtendedResourceProfile.class.getName();
+  }
+
+  public NMProfile getChildProfile() {
+    return childProfile;
+  }
+
+  public void setChildProfile(NMProfile nmProfile) {
+    this.childProfile = nmProfile;
+  }
+
+  @Override
+  public String getName() {
+    return childProfile.getName();
+  }
+
+  @Override
+  public Double getCpus() {
+    return childProfile.getCpus().doubleValue();
+  }
+
+  @Override
+  public Double getMemory() {
+    return childProfile.getMemory().doubleValue();
+  }
+
+  @Override
+  public Double getAggregateMemory() {
+    return memory + childProfile.getMemory();
+  }
+
+  @Override
+  public Double getAggregateCpu() {
+    return cpus + childProfile.getCpus();
+  }
+
+  @Override
+  public String toString() {
+    Gson gson = new Gson();
+    return gson.toJson(this);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriver.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriver.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriver.java
new file mode 100644
index 0000000..741c6de
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriver.java
@@ -0,0 +1,66 @@
+/**
+ * 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.myriad.scheduler;
+
+import javax.inject.Inject;
+
+import org.apache.mesos.Protos.Status;
+import org.apache.mesos.Protos.TaskID;
+import org.apache.mesos.SchedulerDriver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Driver for Myriad scheduler.
+ */
+public class MyriadDriver {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MyriadDriver.class);
+
+  private final SchedulerDriver driver;
+
+  @Inject
+  public MyriadDriver(SchedulerDriver driver) {
+    this.driver = driver;
+  }
+
+  public Status start() {
+    LOGGER.info("Starting driver");
+    Status status = driver.start();
+    LOGGER.info("Driver started with status: {}", status);
+    return status;
+  }
+
+  public Status kill(final TaskID taskId) {
+    LOGGER.info("Killing task {}", taskId);
+    Status status = driver.killTask(taskId);
+    LOGGER.info("Task {} killed with status: {}", taskId, status);
+    return status;
+  }
+
+  public Status abort() {
+    LOGGER.info("Aborting driver");
+    Status status = driver.abort();
+    LOGGER.info("Driver aborted with status: {}", status);
+    return status;
+  }
+
+  public SchedulerDriver getDriver() {
+    return driver;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriverManager.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriverManager.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriverManager.java
new file mode 100644
index 0000000..44d7e06
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadDriverManager.java
@@ -0,0 +1,103 @@
+/**
+ * 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.myriad.scheduler;
+
+import com.google.common.base.Preconditions;
+import org.apache.mesos.Protos;
+import org.apache.mesos.Protos.Status;
+import org.apache.mesos.Protos.TaskID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Manager for the myriad scheduler driver
+ */
+public class MyriadDriverManager {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MyriadDriverManager.class);
+  private final Lock driverLock;
+  private MyriadDriver driver;
+  private Status driverStatus;
+
+  @Inject
+  public MyriadDriverManager(MyriadDriver driver) {
+    this.driver = driver;
+    this.driverLock = new ReentrantLock();
+    this.driverStatus = Protos.Status.DRIVER_NOT_STARTED;
+  }
+
+  public Status startDriver() {
+    this.driverLock.lock();
+    try {
+      Preconditions.checkState(this.isStartable());
+      LOGGER.info("Starting driver...");
+      this.driverStatus = driver.start();
+      LOGGER.info("Driver started with status: {}", this.driverStatus);
+    } finally {
+      this.driverLock.unlock();
+    }
+    return this.driverStatus;
+  }
+
+  public Status stopDriver() {
+    this.driverLock.lock();
+    try {
+      if (isRunning()) {
+        LOGGER.info("Aborting driver...");
+        this.driverStatus = this.driver.abort();
+        LOGGER.info("Aborted driver with status: {}", this.driverStatus);
+      }
+    } finally {
+      this.driverLock.unlock();
+    }
+    return driverStatus;
+  }
+
+  public Status kill(final TaskID taskId) {
+    LOGGER.info("Killing task {}", taskId);
+    this.driverLock.lock();
+    try {
+      if (isRunning()) {
+        this.driverStatus = driver.kill(taskId);
+        LOGGER.info("Task {} killed with status: {}", taskId, this.driverStatus);
+      } else {
+        LOGGER.warn("Cannot kill task, driver is not running");
+      }
+    } finally {
+      this.driverLock.unlock();
+    }
+
+    return driverStatus;
+  }
+
+  public Status getDriverStatus() {
+    return this.driverStatus;
+  }
+
+  private boolean isStartable() {
+    return this.driver != null && this.driverStatus == Status.DRIVER_NOT_STARTED;
+  }
+
+  private boolean isRunning() {
+    return this.driver != null && this.driverStatus == Status.DRIVER_RUNNING;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadOperations.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadOperations.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadOperations.java
new file mode 100644
index 0000000..6b0de6d
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadOperations.java
@@ -0,0 +1,236 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.apache.myriad.configuration.MyriadBadConfigurationException;
+import org.apache.myriad.configuration.ServiceConfiguration;
+import org.apache.myriad.configuration.MyriadConfiguration;
+import org.apache.myriad.configuration.NodeManagerConfiguration;
+import org.apache.myriad.policy.NodeScaleDownPolicy;
+import org.apache.myriad.scheduler.constraints.Constraint;
+import org.apache.myriad.scheduler.constraints.LikeConstraint;
+import org.apache.myriad.state.NodeTask;
+import org.apache.myriad.state.SchedulerState;
+import com.google.common.collect.Lists;
+import com.google.inject.Inject;
+
+import org.apache.mesos.Protos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Myriad scheduler operations
+ */
+public class MyriadOperations {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MyriadOperations.class);
+  private final SchedulerState schedulerState;
+
+  private MyriadConfiguration cfg;
+  private NodeScaleDownPolicy nodeScaleDownPolicy;
+
+  @Inject
+  public MyriadOperations(MyriadConfiguration cfg, SchedulerState schedulerState, NodeScaleDownPolicy nodeScaleDownPolicy) {
+    this.cfg = cfg;
+    this.schedulerState = schedulerState;
+    this.nodeScaleDownPolicy = nodeScaleDownPolicy;
+  }
+
+  public void flexUpCluster(ServiceResourceProfile serviceResourceProfile, int instances, Constraint constraint) {
+    Collection<NodeTask> nodes = new HashSet<>();
+    for (int i = 0; i < instances; i++) {
+      NodeTask nodeTask = new NodeTask(serviceResourceProfile, constraint);
+      nodeTask.setTaskPrefix(NodeManagerConfiguration.NM_TASK_PREFIX);
+      nodes.add(nodeTask);
+    }
+
+    LOGGER.info("Adding {} NM instances to cluster", nodes.size());
+    this.schedulerState.addNodes(nodes);
+  }
+
+  public void flexDownCluster(ServiceResourceProfile serviceResourceProfile, Constraint constraint, int numInstancesToScaleDown) {
+    // Flex down Pending tasks, if any
+    int numPendingTasksScaledDown = flexDownPendingTasks(serviceResourceProfile, constraint, numInstancesToScaleDown);
+
+    // Flex down Staging tasks, if any
+    int numStagingTasksScaledDown = flexDownStagingTasks(serviceResourceProfile, constraint, numInstancesToScaleDown - numPendingTasksScaledDown);
+
+    // Flex down Active tasks, if any
+    int numActiveTasksScaledDown = flexDownActiveTasks(serviceResourceProfile, constraint, numInstancesToScaleDown - numPendingTasksScaledDown - numStagingTasksScaledDown);
+
+    if (numActiveTasksScaledDown + numStagingTasksScaledDown + numPendingTasksScaledDown == 0) {
+      LOGGER.info("No Node Managers with profile '{}' and constraint '{}' found for scaling down.", serviceResourceProfile.getName(), constraint == null ? "null" : constraint.toString());
+    } else {
+      LOGGER.info("Flexed down {} active, {} staging  and {} pending Node Managers with " + "'{}' profile and constraint '{}'.", numActiveTasksScaledDown, numStagingTasksScaledDown, numPendingTasksScaledDown, serviceResourceProfile.getName(),
+          constraint == null ? "null" : constraint.toString());
+    }
+  }
+
+  /**
+   * Flexup a service
+   *
+   * @param instances
+   * @param serviceName
+   */
+  public void flexUpAService(int instances, String serviceName) throws MyriadBadConfigurationException {
+    final ServiceConfiguration auxTaskConf = cfg.getServiceConfiguration(serviceName);
+
+    int totalflexInstances = instances + getFlexibleInstances(serviceName);
+    Integer maxInstances = auxTaskConf.getMaxInstances().orNull();
+    if (maxInstances != null && maxInstances > 0) {
+      // check number of instances
+      // sum of active, staging, pending should be < maxInstances
+      if (totalflexInstances > maxInstances) {
+        LOGGER.error("Current number of active, staging, pending and requested instances: {}" + ", while it is greater then max instances allowed: {}", totalflexInstances, maxInstances);
+        throw new MyriadBadConfigurationException("Current number of active, staging, pending instances and requested: " + totalflexInstances + ", while it is greater then max instances allowed: " + maxInstances);
+      }
+    }
+
+    final Double cpu = auxTaskConf.getCpus().or(ServiceConfiguration.DEFAULT_CPU);
+    final Double mem = auxTaskConf.getJvmMaxMemoryMB().or(ServiceConfiguration.DEFAULT_MEMORY);
+
+    Collection<NodeTask> nodes = new HashSet<>();
+    for (int i = 0; i < instances; i++) {
+      NodeTask nodeTask = new NodeTask(new ServiceResourceProfile(serviceName, cpu, mem), null);
+      nodeTask.setTaskPrefix(serviceName);
+      nodes.add(nodeTask);
+    }
+
+    LOGGER.info("Adding {} {} instances to cluster", nodes.size(), serviceName);
+    this.schedulerState.addNodes(nodes);
+  }
+
+  /**
+   * Flexing down any service defined in the configuration
+   *
+   * @param numInstancesToScaleDown
+   * @param serviceName             - name of the service
+   */
+  public void flexDownAService(int numInstancesToScaleDown, String serviceName) {
+    LOGGER.info("About to flex down {} instances of {}", numInstancesToScaleDown, serviceName);
+
+    int numScaledDown = 0;
+
+    // Flex down Pending tasks, if any
+    if (numScaledDown < numInstancesToScaleDown) {
+      Collection<Protos.TaskID> pendingTasks = this.schedulerState.getPendingTaskIds(serviceName);
+
+      for (Protos.TaskID taskId : pendingTasks) {
+        this.schedulerState.makeTaskKillable(taskId);
+        numScaledDown++;
+        if (numScaledDown >= numInstancesToScaleDown) {
+          break;
+        }
+      }
+    }
+    int numPendingTasksScaledDown = numScaledDown;
+
+    // Flex down Staging tasks, if any
+    if (numScaledDown < numInstancesToScaleDown) {
+      Collection<Protos.TaskID> stagingTasks = this.schedulerState.getStagingTaskIds(serviceName);
+
+      for (Protos.TaskID taskId : stagingTasks) {
+        this.schedulerState.makeTaskKillable(taskId);
+        numScaledDown++;
+        if (numScaledDown >= numInstancesToScaleDown) {
+          break;
+        }
+      }
+    }
+    int numStagingTasksScaledDown = numScaledDown - numPendingTasksScaledDown;
+
+    Set<NodeTask> activeTasks = this.schedulerState.getActiveTasksByType(serviceName);
+    if (numScaledDown < numInstancesToScaleDown) {
+      for (NodeTask nodeTask : activeTasks) {
+        this.schedulerState.makeTaskKillable(nodeTask.getTaskStatus().getTaskId());
+        numScaledDown++;
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("Marked NodeTask {} on host {} for kill.", nodeTask.getTaskStatus().getTaskId(), nodeTask.getHostname());
+        }
+        if (numScaledDown >= numInstancesToScaleDown) {
+          break;
+        }
+      }
+    }
+
+    LOGGER.info("Flexed down {} of {} instances including {} staging instances, and {} pending instances of {}", numScaledDown, numInstancesToScaleDown, numStagingTasksScaledDown, numPendingTasksScaledDown, serviceName);
+  }
+
+  private int flexDownPendingTasks(ServiceResourceProfile profile, Constraint constraint, int numInstancesToScaleDown) {
+    return numInstancesToScaleDown > 0 ? flexDownTasks(schedulerState.getPendingTaskIDsForProfile(profile), profile, constraint, numInstancesToScaleDown) : 0;
+  }
+
+  private int flexDownStagingTasks(ServiceResourceProfile profile, Constraint constraint, int numInstancesToScaleDown) {
+    return numInstancesToScaleDown > 0 ? flexDownTasks(schedulerState.getStagingTaskIDsForProfile(profile), profile, constraint, numInstancesToScaleDown) : 0;
+  }
+
+  private int flexDownActiveTasks(ServiceResourceProfile profile, Constraint constraint, int numInstancesToScaleDown) {
+    if (numInstancesToScaleDown > 0) {
+      List<Protos.TaskID> activeTasksForProfile = Lists.newArrayList(schedulerState.getActiveTaskIDsForProfile(profile));
+      nodeScaleDownPolicy.apply(activeTasksForProfile);
+      return flexDownTasks(activeTasksForProfile, profile, constraint, numInstancesToScaleDown);
+    }
+    return 0;
+  }
+
+  private int flexDownTasks(Collection<Protos.TaskID> taskIDs, ServiceResourceProfile profile, Constraint constraint, int numInstancesToScaleDown) {
+    int numInstancesScaledDown = 0;
+    for (Protos.TaskID taskID : taskIDs) {
+      NodeTask nodeTask = schedulerState.getTask(taskID);
+      if (nodeTask.getProfile().getName().equals(profile.getName()) && meetsConstraint(nodeTask, constraint)) {
+        this.schedulerState.makeTaskKillable(taskID);
+        numInstancesScaledDown++;
+        if (numInstancesScaledDown == numInstancesToScaleDown) {
+          break;
+        }
+      }
+    }
+    return numInstancesScaledDown;
+  }
+
+  private boolean meetsConstraint(NodeTask nodeTask, Constraint constraint) {
+    if (constraint != null) {
+      if (constraint.equals(nodeTask.getConstraint())) {
+        return true;
+      }
+      switch (constraint.getType()) {
+        case LIKE:
+          LikeConstraint likeConstraint = (LikeConstraint) constraint;
+          if (likeConstraint.isConstraintOnHostName()) {
+            return likeConstraint.matchesHostName(nodeTask.getHostname());
+          } else {
+            return likeConstraint.matchesSlaveAttributes(nodeTask.getSlaveAttributes());
+          }
+
+        default:
+          return false;
+      }
+    }
+    return true;
+  }
+
+  public Integer getFlexibleInstances(String taskPrefix) {
+    return this.schedulerState.getActiveTaskIds(taskPrefix).size() + this.schedulerState.getStagingTaskIds(taskPrefix).size() + this.schedulerState.getPendingTaskIds(taskPrefix).size();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadScheduler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadScheduler.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadScheduler.java
new file mode 100644
index 0000000..235d01b
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/MyriadScheduler.java
@@ -0,0 +1,161 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+import org.apache.myriad.scheduler.event.ErrorEvent;
+import org.apache.myriad.scheduler.event.FrameworkMessageEvent;
+import org.apache.myriad.scheduler.event.OfferRescindedEvent;
+import org.apache.myriad.scheduler.event.ReRegisteredEvent;
+import org.apache.myriad.scheduler.event.ResourceOffersEvent;
+import org.apache.myriad.scheduler.event.SlaveLostEvent;
+import org.apache.myriad.scheduler.event.StatusUpdateEvent;
+import com.lmax.disruptor.EventTranslator;
+import org.apache.mesos.Protos;
+import org.apache.mesos.Scheduler;
+import org.apache.mesos.SchedulerDriver;
+
+import javax.inject.Inject;
+import java.util.List;
+
+/**
+ * Myriad Scheduler
+ */
+public class MyriadScheduler implements Scheduler {
+  private org.apache.myriad.DisruptorManager disruptorManager;
+
+  @Inject
+  public MyriadScheduler(final MyriadConfiguration cfg, final org.apache.myriad.DisruptorManager disruptorManager) {
+    this.disruptorManager = disruptorManager;
+  }
+
+  @Override
+  public void registered(final SchedulerDriver driver, final Protos.FrameworkID frameworkId, final Protos.MasterInfo masterInfo) {
+    disruptorManager.getRegisteredEventDisruptor().publishEvent(new EventTranslator<org.apache.myriad.scheduler.event.RegisteredEvent>() {
+      @Override
+      public void translateTo(org.apache.myriad.scheduler.event.RegisteredEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setFrameworkId(frameworkId);
+        event.setMasterInfo(masterInfo);
+      }
+    });
+  }
+
+  @Override
+  public void reregistered(final SchedulerDriver driver, final Protos.MasterInfo masterInfo) {
+    disruptorManager.getReRegisteredEventDisruptor().publishEvent(new EventTranslator<ReRegisteredEvent>() {
+      @Override
+      public void translateTo(ReRegisteredEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setMasterInfo(masterInfo);
+      }
+    });
+  }
+
+  @Override
+  public void resourceOffers(final SchedulerDriver driver, final List<Protos.Offer> offers) {
+    disruptorManager.getResourceOffersEventDisruptor().publishEvent(new EventTranslator<ResourceOffersEvent>() {
+      @Override
+      public void translateTo(ResourceOffersEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setOffers(offers);
+      }
+    });
+  }
+
+  @Override
+  public void offerRescinded(final SchedulerDriver driver, final Protos.OfferID offerId) {
+    disruptorManager.getOfferRescindedEventDisruptor().publishEvent(new EventTranslator<OfferRescindedEvent>() {
+      @Override
+      public void translateTo(OfferRescindedEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setOfferId(offerId);
+      }
+    });
+  }
+
+  @Override
+  public void statusUpdate(final SchedulerDriver driver, final Protos.TaskStatus status) {
+    disruptorManager.getStatusUpdateEventDisruptor().publishEvent(new EventTranslator<StatusUpdateEvent>() {
+      @Override
+      public void translateTo(StatusUpdateEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setStatus(status);
+      }
+    });
+  }
+
+  @Override
+  public void frameworkMessage(final SchedulerDriver driver, final Protos.ExecutorID executorId, final Protos.SlaveID slaveId, final byte[] bytes) {
+    disruptorManager.getFrameworkMessageEventDisruptor().publishEvent(new EventTranslator<FrameworkMessageEvent>() {
+      @Override
+      public void translateTo(FrameworkMessageEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setBytes(bytes);
+        event.setExecutorId(executorId);
+        event.setSlaveId(slaveId);
+      }
+    });
+  }
+
+  @Override
+  public void disconnected(final SchedulerDriver driver) {
+    disruptorManager.getDisconnectedEventDisruptor().publishEvent(new EventTranslator<org.apache.myriad.scheduler.event.DisconnectedEvent>() {
+      @Override
+      public void translateTo(org.apache.myriad.scheduler.event.DisconnectedEvent event, long sequence) {
+        event.setDriver(driver);
+      }
+    });
+  }
+
+  @Override
+  public void slaveLost(final SchedulerDriver driver, final Protos.SlaveID slaveId) {
+    disruptorManager.getSlaveLostEventDisruptor().publishEvent(new EventTranslator<SlaveLostEvent>() {
+      @Override
+      public void translateTo(SlaveLostEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setSlaveId(slaveId);
+      }
+    });
+  }
+
+  @Override
+  public void executorLost(final SchedulerDriver driver, final Protos.ExecutorID executorId, final Protos.SlaveID slaveId, final int exitStatus) {
+    disruptorManager.getExecutorLostEventDisruptor().publishEvent(new EventTranslator<org.apache.myriad.scheduler.event.ExecutorLostEvent>() {
+      @Override
+      public void translateTo(org.apache.myriad.scheduler.event.ExecutorLostEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setExecutorId(executorId);
+        event.setSlaveId(slaveId);
+        event.setExitStatus(exitStatus);
+      }
+    });
+  }
+
+  @Override
+  public void error(final SchedulerDriver driver, final String message) {
+    disruptorManager.getErrorEventDisruptor().publishEvent(new EventTranslator<ErrorEvent>() {
+      @Override
+      public void translateTo(ErrorEvent event, long sequence) {
+        event.setDriver(driver);
+        event.setMessage(message);
+      }
+    });
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMExecutorCLGenImpl.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMExecutorCLGenImpl.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMExecutorCLGenImpl.java
new file mode 100644
index 0000000..411518f
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMExecutorCLGenImpl.java
@@ -0,0 +1,180 @@
+/**
+ * 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.myriad.scheduler;
+
+import java.util.Map;
+import java.util.HashMap;
+
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+
+/**
+ * Implementation assumes NM binaries already deployed
+ */
+public class NMExecutorCLGenImpl implements ExecutorCommandLineGenerator {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(NMExecutorCLGenImpl.class);
+
+  public static final String ENV_YARN_NODEMANAGER_OPTS = "YARN_NODEMANAGER_OPTS";
+  public static final String KEY_YARN_NM_CGROUPS_PATH = "yarn.nodemanager.cgroups.path";
+  public static final String KEY_YARN_RM_HOSTNAME = "yarn.resourcemanager.hostname";
+
+  /**
+   * YARN container executor class.
+   */
+  public static final String KEY_YARN_NM_CONTAINER_EXECUTOR_CLASS = "yarn.nodemanager.container-executor.class";
+  // TODO (mohit): Should it be configurable ?
+  public static final String VAL_YARN_NM_CONTAINER_EXECUTOR_CLASS = "org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor";
+  public static final String DEFAULT_YARN_NM_CONTAINER_EXECUTOR_CLASS = "org.apache.hadoop.yarn.server.nodemanager.DefaultContainerExecutor";
+
+  /**
+   * YARN class to help handle LCE resources
+   */
+  public static final String KEY_YARN_NM_LCE_RH_CLASS = "yarn.nodemanager.linux-container-executor.resources-handler.class";
+
+  // TODO (mohit): Should it be configurable ?
+  public static final String VAL_YARN_NM_LCE_RH_CLASS = "org.apache.hadoop.yarn.server.nodemanager.util.CgroupsLCEResourcesHandler";
+  public static final String KEY_YARN_NM_LCE_CGROUPS_HIERARCHY = "yarn.nodemanager.linux-container-executor.cgroups.hierarchy";
+  public static final String VAL_YARN_NM_LCE_CGROUPS_HIERARCHY = "mesos/$TASK_DIR";
+  public static final String KEY_YARN_NM_LCE_CGROUPS_MOUNT = "yarn.nodemanager.linux-container-executor.cgroups.mount";
+  public static final String KEY_YARN_NM_LCE_CGROUPS_MOUNT_PATH = "yarn.nodemanager.linux-container-executor.cgroups.mount-path";
+  public static final String VAL_YARN_NM_LCE_CGROUPS_MOUNT_PATH = "/sys/fs/cgroup";
+  public static final String KEY_YARN_NM_LCE_GROUP = "yarn.nodemanager.linux-container-executor.group";
+  public static final String KEY_YARN_NM_LCE_PATH = "yarn.nodemanager.linux-container-executor.path";
+  public static final String KEY_YARN_HOME = "yarn.home";
+  public static final String KEY_NM_RESOURCE_CPU_VCORES = "nodemanager.resource.cpu-vcores";
+  public static final String KEY_NM_RESOURCE_MEM_MB = "nodemanager.resource.memory-mb";
+  public static final String YARN_NM_CMD = " $YARN_HOME/bin/yarn nodemanager";
+  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";
+
+  private static final String ALL_LOCAL_IPV4ADDR = "0.0.0.0:";
+  private static final String PROPERTY_FORMAT = "-D%s=%s";
+
+  private Map<String, String> environment = new HashMap<>();
+  protected MyriadConfiguration cfg;
+
+  public NMExecutorCLGenImpl(MyriadConfiguration cfg) {
+    this.cfg = cfg;
+  }
+
+  @Override
+  public String generateCommandLine(ServiceResourceProfile profile, Ports ports) {
+    StringBuilder cmdLine = new StringBuilder();
+
+    generateEnvironment(profile, (NMPorts) ports);
+    appendCgroupsCmds(cmdLine);
+    appendYarnHomeExport(cmdLine);
+    appendEnvForNM(cmdLine);
+    cmdLine.append(YARN_NM_CMD);
+    return cmdLine.toString();
+  }
+
+  protected void generateEnvironment(ServiceResourceProfile profile, NMPorts ports) {
+    //yarnEnvironemnt configuration from yaml file
+    Map<String, String> yarnEnvironmentMap = cfg.getYarnEnvironment();
+    if (yarnEnvironmentMap != null) {
+      environment.putAll(yarnEnvironmentMap);
+    }
+
+    String rmHostName = System.getProperty(KEY_YARN_RM_HOSTNAME);
+    if (rmHostName != null && !rmHostName.isEmpty()) {
+      addYarnNodemanagerOpt(KEY_YARN_RM_HOSTNAME, rmHostName);
+    }
+
+    if (cfg.getNodeManagerConfiguration().getCgroups().or(Boolean.FALSE)) {
+      addYarnNodemanagerOpt(KEY_YARN_NM_CONTAINER_EXECUTOR_CLASS, VAL_YARN_NM_CONTAINER_EXECUTOR_CLASS);
+      addYarnNodemanagerOpt(KEY_YARN_NM_LCE_RH_CLASS, VAL_YARN_NM_LCE_RH_CLASS);
+
+      // TODO: Configure hierarchy
+      addYarnNodemanagerOpt(KEY_YARN_NM_LCE_CGROUPS_HIERARCHY, VAL_YARN_NM_LCE_CGROUPS_HIERARCHY);
+      addYarnNodemanagerOpt(KEY_YARN_NM_LCE_CGROUPS_MOUNT, "true");
+      // TODO: Make it configurable
+      addYarnNodemanagerOpt(KEY_YARN_NM_LCE_CGROUPS_MOUNT_PATH, VAL_YARN_NM_LCE_CGROUPS_MOUNT_PATH);
+      addYarnNodemanagerOpt(KEY_YARN_NM_LCE_GROUP, "root");
+      if (environment.containsKey("YARN_HOME")) {
+        addYarnNodemanagerOpt(KEY_YARN_HOME, environment.get("YARN_HOME"));
+      }
+    } else {
+      // Otherwise configure to use Default
+      addYarnNodemanagerOpt(KEY_YARN_NM_CONTAINER_EXECUTOR_CLASS, DEFAULT_YARN_NM_CONTAINER_EXECUTOR_CLASS);
+    }
+    addYarnNodemanagerOpt(KEY_NM_RESOURCE_CPU_VCORES, Integer.toString(profile.getCpus().intValue()));
+    addYarnNodemanagerOpt(KEY_NM_RESOURCE_MEM_MB, Integer.toString(profile.getMemory().intValue()));
+    addYarnNodemanagerOpt(KEY_NM_ADDRESS, ALL_LOCAL_IPV4ADDR + Long.valueOf(ports.getRpcPort()).toString());
+    addYarnNodemanagerOpt(KEY_NM_LOCALIZER_ADDRESS, ALL_LOCAL_IPV4ADDR + Long.valueOf(ports.getLocalizerPort()).toString());
+    addYarnNodemanagerOpt(KEY_NM_WEBAPP_ADDRESS, ALL_LOCAL_IPV4ADDR + Long.valueOf(ports.getWebAppHttpPort()).toString());
+    addYarnNodemanagerOpt(KEY_NM_SHUFFLE_PORT, Long.valueOf(ports.getShufflePort()).toString());
+  }
+
+  protected void appendEnvForNM(StringBuilder cmdLine) {
+    cmdLine.append(" env ");
+    for (Map.Entry<String, String> env : environment.entrySet()) {
+      cmdLine.append(env.getKey()).append("=").append("\"").append(env.getValue()).append("\" ");
+    }
+  }
+
+  protected void appendCgroupsCmds(StringBuilder cmdLine) {
+    if (cfg.getNodeManagerConfiguration().getCgroups().or(Boolean.FALSE)) {
+      cmdLine.append(" export TASK_DIR=`basename $PWD`;");
+      cmdLine.append(" chmod +x /sys/fs/cgroup/cpu/mesos/$TASK_DIR;");
+    }
+  }
+
+  protected void appendYarnHomeExport(StringBuilder cmdLine) {
+    if (environment.containsKey("YARN_HOME")) {
+      cmdLine.append(" export YARN_HOME=" + environment.get("YARN_HOME") + ";");
+    }
+  }
+
+  protected void addYarnNodemanagerOpt(String propertyName, String propertyValue) {
+    String envOpt = String.format(PROPERTY_FORMAT, propertyName, propertyValue);
+    if (environment.containsKey(ENV_YARN_NODEMANAGER_OPTS)) {
+      String existingOpts = environment.get(ENV_YARN_NODEMANAGER_OPTS);
+      environment.put(ENV_YARN_NODEMANAGER_OPTS, existingOpts + " " + envOpt);
+    } else {
+      environment.put(ENV_YARN_NODEMANAGER_OPTS, envOpt);
+    }
+  }
+
+  @Override
+  public String getConfigurationUrl() {
+    YarnConfiguration conf = new YarnConfiguration();
+    String httpPolicy = conf.get(TaskFactory.YARN_HTTP_POLICY);
+    if (httpPolicy != null && httpPolicy.equals(TaskFactory.YARN_HTTP_POLICY_HTTPS_ONLY)) {
+      String address = conf.get(TaskFactory.YARN_RESOURCEMANAGER_WEBAPP_HTTPS_ADDRESS);
+      if (address == null || address.isEmpty()) {
+        address = conf.get(TaskFactory.YARN_RESOURCEMANAGER_HOSTNAME) + ":8090";
+      }
+      return "https://" + address + "/conf";
+    } else {
+      String address = conf.get(TaskFactory.YARN_RESOURCEMANAGER_WEBAPP_ADDRESS);
+      if (address == null || address.isEmpty()) {
+        address = conf.get(TaskFactory.YARN_RESOURCEMANAGER_HOSTNAME) + ":8088";
+      }
+      return "http://" + address + "/conf";
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMPorts.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMPorts.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMPorts.java
new file mode 100644
index 0000000..0944a2a
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMPorts.java
@@ -0,0 +1,78 @@
+/**
+ * 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.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 implements Ports {
+  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/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfile.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfile.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfile.java
new file mode 100644
index 0000000..3de82a5
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfile.java
@@ -0,0 +1,63 @@
+/**
+ * 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.myriad.scheduler;
+
+import com.google.gson.Gson;
+
+/**
+ * Node Manager Profile
+ */
+public class NMProfile {
+  private String name;
+
+  /**
+   * Number of CPU advertised to YARN Resource Manager.
+   */
+  private Long cpus;
+
+  /**
+   * Memory in MB advertised to YARN Resource Manager.
+   */
+  private Long memory;
+
+  public NMProfile(String name, Long cpus, Long memory) {
+    this.name = name;
+    this.cpus = cpus;
+    this.memory = memory;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public Long getCpus() {
+    return cpus;
+  }
+
+  public Long getMemory() {
+    return memory;
+  }
+
+  @Override
+  public String toString() {
+    Gson gson = new Gson();
+    return gson.toJson(this);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfileManager.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfileManager.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfileManager.java
new file mode 100644
index 0000000..38fad1d
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMProfileManager.java
@@ -0,0 +1,54 @@
+/**
+ * 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.myriad.scheduler;
+
+import com.google.gson.Gson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Node Manager Profile Manager
+ */
+public class NMProfileManager {
+  private static final Logger LOGGER = LoggerFactory.getLogger(NMProfileManager.class);
+
+  private Map<String, NMProfile> profiles = new ConcurrentHashMap<>();
+
+  public NMProfile get(String name) {
+    return profiles.get(name);
+  }
+
+  public void add(NMProfile profile) {
+    LOGGER.info("Adding profile {} with CPU: {} and Memory: {}", profile.getName(), profile.getCpus(), profile.getMemory());
+
+    profiles.put(profile.getName(), profile);
+  }
+
+  public boolean exists(String name) {
+    return this.profiles.containsKey(name);
+  }
+
+  public String toString() {
+    Gson gson = new Gson();
+    return gson.toJson(this);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMTaskFactoryAnnotation.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMTaskFactoryAnnotation.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMTaskFactoryAnnotation.java
new file mode 100644
index 0000000..62d9f52
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/NMTaskFactoryAnnotation.java
@@ -0,0 +1,36 @@
+/**
+ * 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.myriad.scheduler;
+
+import com.google.inject.BindingAnnotation;
+import java.lang.annotation.Target;
+import java.lang.annotation.Retention;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+
+/**
+ * NMTaskFactory annotation that allows to bind TaskFactory to NM specific implementation
+ */
+@BindingAnnotation
+@Target({FIELD, PARAMETER, METHOD})
+@Retention(RUNTIME)
+public @interface NMTaskFactoryAnnotation {
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Ports.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Ports.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Ports.java
new file mode 100644
index 0000000..03150fb
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Ports.java
@@ -0,0 +1,26 @@
+/**
+ * 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.myriad.scheduler;
+
+/**
+ * Generic interface to represent ports
+ */
+public interface Ports {
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Rebalancer.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Rebalancer.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Rebalancer.java
new file mode 100644
index 0000000..955bc77
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/Rebalancer.java
@@ -0,0 +1,101 @@
+/**
+ * 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.myriad.scheduler;
+
+import javax.inject.Inject;
+import java.util.Set;
+
+import org.apache.mesos.Protos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link Rebalancer} is responsible for scaling registered YARN clusters as per
+ * configured rules and policies.
+ */
+public class Rebalancer implements Runnable {
+  private static final Logger LOGGER = LoggerFactory.getLogger(Rebalancer.class);
+
+  private final org.apache.myriad.state.SchedulerState schedulerState;
+  private final MyriadOperations myriadOperations;
+  private final ServiceProfileManager profileManager;
+
+  @Inject
+  public Rebalancer(org.apache.myriad.state.SchedulerState schedulerState, MyriadOperations myriadOperations, ServiceProfileManager profileManager) {
+    this.schedulerState = schedulerState;
+    this.myriadOperations = myriadOperations;
+    this.profileManager = profileManager;
+  }
+
+  @Override
+  public void run() {
+    final Set<Protos.TaskID> activeIds = schedulerState.getActiveTaskIds(org.apache.myriad.configuration.NodeManagerConfiguration.NM_TASK_PREFIX);
+    final Set<Protos.TaskID> pendingIds = schedulerState.getPendingTaskIds(org.apache.myriad.configuration.NodeManagerConfiguration.NM_TASK_PREFIX);
+    LOGGER.info("Active {}, Pending {}", activeIds.size(), pendingIds.size());
+    if (activeIds.size() < 1 && pendingIds.size() < 1) {
+      myriadOperations.flexUpCluster(profileManager.get("small"), 1, null);
+    }
+    //            RestAdapter restAdapter = new RestAdapter.Builder()
+    //                    .setEndpoint("http://" + host + ":" + port)
+    //                    .setLogLevel(LogLevel.FULL).build();
+    //            YARNResourceManagerService service = restAdapter
+    //                    .create(YARNResourceManagerService.class);
+    //
+    //            ClusterMetrics metrics = service.metrics().getClusterMetrics();
+    //            AppsResponse appsResponse = service.apps("ACCEPTED");
+    //
+    //            int acceptedApps = 0;
+    //
+    //            if (appsResponse == null || appsResponse.getApps() == null
+    //                    || appsResponse.getApps().getApps() == null) {
+    //                acceptedApps = 0;
+    //            } else {
+    //                acceptedApps = appsResponse.getApps().getApps().size();
+    //            }
+    //            LOGGER.info("Metrics: {}", metrics);
+    //            LOGGER.info("Apps: {}", appsResponse);
+    //
+    //            long availableMB = metrics.getAvailableMB();
+    //            long allocatedMB = metrics.getAllocatedMB();
+    //            long reservedMB = metrics.getReservedMB();
+    //            int activeNodes = metrics.getActiveNodes();
+    //            int unhealthyNodes = metrics.getUnhealthyNodes();
+    //            int appsPending = metrics.getAppsPending();
+    //            int appsRunning = metrics.getAppsRunning();
+
+    //            if (activeNodes == 0 && appsPending > 0) {
+    //                LOGGER.info(
+    //                        "Flexing up for condition: activeNodes ({}) == 0 && appsPending ({}) > 0",
+    //                        activeNodes, appsPending);
+    //                this.myriadOperations.flexUpCluster(clusterId, 1, "small");
+    //            } else if (appsPending == 0 && appsRunning == 0 && activeNodes > 0) {
+    //                LOGGER.info(
+    //                        "Flexing down for condition: appsPending ({}) == 0 && appsRunning ({}) == 0 && activeNodes ({}) > 0",
+    //                        appsPending, appsRunning, activeNodes);
+    //                this.myriadOperations.flexDownCluster(cluster, 1);
+    //            } else if (acceptedApps > 0) {
+    //                LOGGER.info("Flexing up for condition: acceptedApps ({}) > 0",
+    //                        acceptedApps);
+    //                this.myriadOperations.flexUpCluster(clusterId, 1, "small");
+    //            } else {
+    //                LOGGER.info("Nothing to rebalance");
+    //                this.schedulerState.releaseLock(clusterId);
+    //            }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ReconcileService.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ReconcileService.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ReconcileService.java
new file mode 100644
index 0000000..d0c74de
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ReconcileService.java
@@ -0,0 +1,84 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+import org.apache.mesos.Protos;
+import org.apache.mesos.SchedulerDriver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+
+/**
+ * {@link ReconcileService} is responsible for reconciling tasks with the mesos master
+ */
+public class ReconcileService {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ReconcileService.class);
+
+  public static final long DEFAULT_RECONCILATION_DELAY_MS = 10000;
+  public static final long MAX_RECONCILE_ATTEMPTS = 10;
+
+  private org.apache.myriad.state.SchedulerState state;
+  private MyriadConfiguration cfg;
+  private Date lastReconcileTime;
+
+  @Inject
+  public ReconcileService(org.apache.myriad.state.SchedulerState state, MyriadConfiguration cfg) {
+    this.state = state;
+    this.cfg = cfg;
+  }
+
+  public void reconcile(SchedulerDriver driver) {
+    Collection<Protos.TaskStatus> taskStatuses = state.getTaskStatuses();
+
+    if (taskStatuses.size() == 0) {
+      return;
+    }
+    LOGGER.info("Reconciling {} tasks.", taskStatuses.size());
+
+    driver.reconcileTasks(taskStatuses);
+
+    lastReconcileTime = new Date();
+
+    int attempt = 1;
+
+    while (attempt <= MAX_RECONCILE_ATTEMPTS) {
+      try {
+        // TODO(mohit): Using exponential backoff here, maybe backoff strategy should be configurable.
+        Thread.sleep(DEFAULT_RECONCILATION_DELAY_MS * attempt);
+      } catch (InterruptedException e) {
+        LOGGER.error("Interrupted", e);
+      }
+      Collection<Protos.TaskStatus> notYetReconciled = new ArrayList<>();
+      for (Protos.TaskStatus status : state.getTaskStatuses()) {
+        if (status.getTimestamp() < lastReconcileTime.getTime()) {
+          notYetReconciled.add(status);
+        }
+      }
+      LOGGER.info("Reconcile attempt {} for {} tasks", attempt, notYetReconciled.size());
+      driver.reconcileTasks(notYetReconciled);
+      lastReconcileTime = new Date();
+      attempt++;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/SchedulerUtils.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/SchedulerUtils.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/SchedulerUtils.java
new file mode 100644
index 0000000..0cbca37
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/SchedulerUtils.java
@@ -0,0 +1,73 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.apache.myriad.configuration.NodeManagerConfiguration;
+import org.apache.myriad.state.NodeTask;
+import org.apache.myriad.state.SchedulerState;
+import com.google.common.base.Preconditions;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.mesos.Protos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+
+/**
+ * Provides utilities for scheduling with the mesos offers
+ */
+public class SchedulerUtils {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerUtils.class);
+
+  public static boolean isUniqueHostname(Protos.OfferOrBuilder offer, NodeTask taskToLaunch, Collection<NodeTask> tasks) {
+    Preconditions.checkArgument(offer != null);
+    String offerHostname = offer.getHostname();
+
+    if (!CollectionUtils.isEmpty(tasks)) {
+      for (NodeTask task : tasks) {
+        if (offerHostname.equalsIgnoreCase(task.getHostname())) {
+          LOGGER.debug("Offer's hostname {} is not unique", offerHostname);
+          return false;
+        }
+      }
+    }
+    LOGGER.debug("Offer's hostname {} is unique", offerHostname);
+    return true;
+  }
+
+  /**
+   * Determines if a given host has a nodemanager running with zero profile. Node Managers
+   * launched with zero profile (zero cpu & memory) are eligible for fine grained scaling.
+   * Node Managers launched with a non-zero profile size are not eligible for fine grained scaling.
+   *
+   * @param hostName
+   * @return
+   */
+  public static boolean isEligibleForFineGrainedScaling(String hostName, SchedulerState state) {
+    for (NodeTask activeNMTask : state.getActiveTasksByType(NodeManagerConfiguration.NM_TASK_PREFIX)) {
+      if (activeNMTask.getProfile().getCpus() == 0 &&
+          activeNMTask.getProfile().getMemory() == 0 &&
+          activeNMTask.getHostname().equals(hostName)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceCommandLineGenerator.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceCommandLineGenerator.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceCommandLineGenerator.java
new file mode 100644
index 0000000..6fd8872
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceCommandLineGenerator.java
@@ -0,0 +1,40 @@
+/**
+ * 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.myriad.scheduler;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+
+/**
+ * CommandLineGenerator for any aux service launched by Myriad as binary distro
+ */
+public class ServiceCommandLineGenerator extends DownloadNMExecutorCLGenImpl {
+
+
+  public ServiceCommandLineGenerator(MyriadConfiguration cfg, String nodeManagerUri) {
+    super(cfg, nodeManagerUri);
+  }
+
+  @Override
+  public String generateCommandLine(ServiceResourceProfile profile, Ports ports) {
+    StringBuilder strB = new StringBuilder();
+    appendDistroExtractionCommands(strB);
+    appendUser(strB);
+    return strB.toString();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceProfileManager.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceProfileManager.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceProfileManager.java
new file mode 100644
index 0000000..5319593
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceProfileManager.java
@@ -0,0 +1,55 @@
+/**
+ * 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.myriad.scheduler;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+
+/**
+ * Class to keep all the ServiceResourceProfiles together
+ */
+public class ServiceProfileManager {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceProfileManager.class);
+
+  private Map<String, ServiceResourceProfile> profiles = new ConcurrentHashMap<>();
+
+  public ServiceResourceProfile get(String name) {
+    return profiles.get(name);
+  }
+
+  public void add(ServiceResourceProfile profile) {
+    LOGGER.info("Adding profile {} with CPU: {} and Memory: {}", profile.getName(), profile.getCpus(), profile.getMemory());
+    profiles.put(profile.getName(), profile);
+  }
+
+  public boolean exists(String name) {
+    return this.profiles.containsKey(name);
+  }
+
+  public String toString() {
+    Gson gson = new Gson();
+    return gson.toJson(this);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceResourceProfile.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceResourceProfile.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceResourceProfile.java
new file mode 100644
index 0000000..021007b
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceResourceProfile.java
@@ -0,0 +1,131 @@
+/**
+ * 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.myriad.scheduler;
+
+import java.lang.reflect.Type;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+
+/**
+ * Resource Profile for any service
+ */
+public class ServiceResourceProfile {
+
+  protected final String name;
+
+  /**
+   * Number of CPU needed to run a service
+   */
+  protected final Double cpus;
+
+  /**
+   * Memory in MB needed to run a service
+   */
+  protected final Double memory;
+
+  protected Double executorCpu = 0.0;
+
+  protected Double executorMemory = 0.0;
+
+  protected String className;
+
+  public ServiceResourceProfile(String name, Double cpu, Double mem) {
+    this.name = name;
+    this.cpus = cpu;
+    this.memory = mem;
+    this.className = ServiceResourceProfile.class.getName();
+  }
+
+
+  public String getName() {
+    return name;
+  }
+
+  public Double getCpus() {
+    return cpus;
+  }
+
+  public Double getMemory() {
+    return memory;
+  }
+
+  public Double getAggregateMemory() {
+    return memory;
+  }
+
+  public Double getAggregateCpu() {
+    return cpus;
+  }
+
+  public Double getExecutorCpu() {
+    return executorCpu;
+  }
+
+  public void setExecutorCpu(Double executorCpu) {
+    this.executorCpu = executorCpu;
+  }
+
+  public Double getExecutorMemory() {
+    return executorMemory;
+  }
+
+  public void setExecutorMemory(Double executorMemory) {
+    this.executorMemory = executorMemory;
+  }
+
+
+  @Override
+  public String toString() {
+    Gson gson = new Gson();
+    return gson.toJson(this);
+  }
+
+  /**
+   * Custom serializer to help with deserialization of class hierarchy
+   * since at the point of deserialization we don't know class of the data
+   * that is being deserialized
+   */
+  public static class CustomDeserializer implements JsonDeserializer<ServiceResourceProfile> {
+    private static final Logger LOGGER = LoggerFactory.getLogger(CustomDeserializer.class);
+
+    @Override
+    public ServiceResourceProfile deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
+      String type = json.getAsJsonObject().get("className").getAsString();
+      try {
+        @SuppressWarnings("rawtypes") Class c = Class.forName(type);
+        if (ServiceResourceProfile.class.equals(c)) {
+          return new Gson().fromJson(json, typeOfT);
+        }
+        ServiceResourceProfile profile = context.deserialize(json, c);
+        return profile;
+      } catch (ClassNotFoundException e) {
+        LOGGER.error("Classname is not found", e);
+      }
+      return null;
+    }
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskConstraints.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskConstraints.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskConstraints.java
new file mode 100644
index 0000000..544c47f
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskConstraints.java
@@ -0,0 +1,53 @@
+/**
+ * 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.myriad.scheduler;
+
+import java.util.Map;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+import org.apache.myriad.configuration.ServiceConfiguration;
+
+/**
+ * ServiceTaskConstraints is an implementation of TaskConstraints for a service
+ * at this point constraints are on ports
+ * Later on there may be other types of constraints added
+ */
+public class ServiceTaskConstraints implements TaskConstraints {
+
+  private int portsCount;
+
+  public ServiceTaskConstraints(MyriadConfiguration cfg, String taskPrefix) {
+    this.portsCount = 0;
+    Map<String, ServiceConfiguration> auxConfigs = cfg.getServiceConfigurations();
+    if (auxConfigs == null) {
+      return;
+    }
+    ServiceConfiguration serviceConfig = auxConfigs.get(taskPrefix);
+    if (serviceConfig != null) {
+      if (serviceConfig.getPorts().isPresent()) {
+        this.portsCount = serviceConfig.getPorts().get().size();
+      }
+    }
+  }
+
+  @Override
+  public int portsCount() {
+    return portsCount;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/101bcad3/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskFactoryImpl.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskFactoryImpl.java b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskFactoryImpl.java
new file mode 100644
index 0000000..d7ca31d
--- /dev/null
+++ b/myriad-scheduler/src/main/java/org/apache/myriad/scheduler/ServiceTaskFactoryImpl.java
@@ -0,0 +1,239 @@
+/**
+ * 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.myriad.scheduler;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import javax.inject.Inject;
+
+import org.apache.mesos.Protos.CommandInfo;
+import org.apache.mesos.Protos.ExecutorInfo;
+import org.apache.mesos.Protos.FrameworkID;
+import org.apache.mesos.Protos.Offer;
+import org.apache.mesos.Protos.Resource;
+import org.apache.mesos.Protos.TaskID;
+import org.apache.mesos.Protos.TaskInfo;
+import org.apache.mesos.Protos.Value;
+import org.apache.mesos.Protos.CommandInfo.URI;
+import org.apache.mesos.Protos.Value.Scalar;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.myriad.configuration.MyriadConfiguration;
+import org.apache.myriad.configuration.MyriadExecutorConfiguration;
+import org.apache.myriad.configuration.ServiceConfiguration;
+import org.apache.myriad.state.NodeTask;
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Generic Service Class that allows to create a service solely base don the configuration
+ * Main properties of configuration are:
+ * 1. command to run
+ * 2. Additional env. variables to set (serviceOpts)
+ * 3. ports to use with names of the properties
+ * 4. TODO (yufeldman) executor info
+ */
+public class ServiceTaskFactoryImpl implements TaskFactory {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTaskFactoryImpl.class);
+
+  public static final long DEFAULT_PORT_NUMBER = 0;
+
+  private MyriadConfiguration cfg;
+  @SuppressWarnings("unused")
+  private TaskUtils taskUtils;
+  private ServiceCommandLineGenerator clGenerator;
+
+  @Inject
+  public ServiceTaskFactoryImpl(MyriadConfiguration cfg, TaskUtils taskUtils) {
+    this.cfg = cfg;
+    this.taskUtils = taskUtils;
+    this.clGenerator = new ServiceCommandLineGenerator(cfg, cfg.getMyriadExecutorConfiguration().getNodeManagerUri().orNull());
+  }
+
+  @Override
+  public TaskInfo createTask(Offer offer, FrameworkID frameworkId, TaskID taskId, NodeTask nodeTask) {
+    Objects.requireNonNull(offer, "Offer should be non-null");
+    Objects.requireNonNull(nodeTask, "NodeTask should be non-null");
+
+    ServiceConfiguration serviceConfig = cfg.getServiceConfiguration(nodeTask.getTaskPrefix());
+
+    Objects.requireNonNull(serviceConfig, "ServiceConfig should be non-null");
+    Objects.requireNonNull(serviceConfig.getCommand().orNull(), "command for ServiceConfig should be non-null");
+
+    final String serviceHostName = "0.0.0.0";
+    final String serviceEnv = serviceConfig.getEnvSettings();
+    final String rmHostName = System.getProperty(YARN_RESOURCEMANAGER_HOSTNAME);
+    List<Long> additionalPortsNumbers = null;
+
+    final StringBuilder strB = new StringBuilder("env ");
+    if (serviceConfig.getServiceOpts() != null) {
+      strB.append(serviceConfig.getServiceOpts()).append("=");
+
+      strB.append("\"");
+      if (rmHostName != null && !rmHostName.isEmpty()) {
+        strB.append("-D" + YARN_RESOURCEMANAGER_HOSTNAME + "=" + rmHostName + " ");
+      }
+
+      Map<String, Long> ports = serviceConfig.getPorts().orNull();
+      if (ports != null && !ports.isEmpty()) {
+        int neededPortsCount = 0;
+        for (Map.Entry<String, Long> portEntry : ports.entrySet()) {
+          Long port = portEntry.getValue();
+          if (port == DEFAULT_PORT_NUMBER) {
+            neededPortsCount++;
+          }
+        }
+        // use provided ports
+        additionalPortsNumbers = getAvailablePorts(offer, neededPortsCount);
+        LOGGER.info("No specified ports found or number of specified ports is not enough. Using ports from Mesos Offers: {}", additionalPortsNumbers);
+        int index = 0;
+        for (Map.Entry<String, Long> portEntry : ports.entrySet()) {
+          String portProperty = portEntry.getKey();
+          Long port = portEntry.getValue();
+          if (port == DEFAULT_PORT_NUMBER) {
+            port = additionalPortsNumbers.get(index++);
+          }
+          strB.append("-D" + portProperty + "=" + serviceHostName + ":" + port + " ");
+        }
+      }
+      strB.append(serviceEnv);
+      strB.append("\"");
+    }
+
+    strB.append(" ");
+    strB.append(serviceConfig.getCommand().get());
+
+    CommandInfo commandInfo = createCommandInfo(nodeTask.getProfile(), strB.toString());
+
+    LOGGER.info("Command line for service: {} is: {}", nodeTask.getTaskPrefix(), strB.toString());
+
+    Scalar taskMemory = Scalar.newBuilder().setValue(nodeTask.getProfile().getMemory()).build();
+    Scalar taskCpus = Scalar.newBuilder().setValue(nodeTask.getProfile().getCpus()).build();
+
+    TaskInfo.Builder taskBuilder = TaskInfo.newBuilder();
+
+    taskBuilder.setName(nodeTask.getTaskPrefix()).setTaskId(taskId).setSlaveId(offer.getSlaveId()).addResources(Resource.newBuilder().setName("cpus").setType(Value.Type.SCALAR).setScalar(taskCpus).build()).addResources(Resource.newBuilder()
+        .setName("mem").setType(Value.Type.SCALAR).setScalar(taskMemory).build());
+
+    if (additionalPortsNumbers != null && !additionalPortsNumbers.isEmpty()) {
+      // set ports
+      Value.Ranges.Builder valueRanger = Value.Ranges.newBuilder();
+      for (Long port : additionalPortsNumbers) {
+        valueRanger.addRange(Value.Range.newBuilder().setBegin(port).setEnd(port));
+      }
+
+      taskBuilder.addResources(Resource.newBuilder().setName("ports").setType(Value.Type.RANGES).setRanges(valueRanger.build()));
+    }
+    taskBuilder.setCommand(commandInfo);
+    return taskBuilder.build();
+  }
+
+  @VisibleForTesting
+  CommandInfo createCommandInfo(ServiceResourceProfile profile, String executorCmd) {
+    MyriadExecutorConfiguration myriadExecutorConfiguration = cfg.getMyriadExecutorConfiguration();
+    CommandInfo.Builder commandInfo = CommandInfo.newBuilder();
+    Map<String, String> envVars = cfg.getYarnEnvironment();
+    if (envVars != null && !envVars.isEmpty()) {
+      org.apache.mesos.Protos.Environment.Builder yarnHomeB = org.apache.mesos.Protos.Environment.newBuilder();
+      for (Map.Entry<String, String> envEntry : envVars.entrySet()) {
+        org.apache.mesos.Protos.Environment.Variable.Builder yarnEnvB = org.apache.mesos.Protos.Environment.Variable.newBuilder();
+        yarnEnvB.setName(envEntry.getKey()).setValue(envEntry.getValue());
+        yarnHomeB.addVariables(yarnEnvB.build());
+      }
+      commandInfo.mergeEnvironment(yarnHomeB.build());
+    }
+
+    if (myriadExecutorConfiguration.getNodeManagerUri().isPresent()) {
+      //Both FrameworkUser and FrameworkSuperuser to get all of the directory permissions correct.
+      if (!(cfg.getFrameworkUser().isPresent() && cfg.getFrameworkSuperUser().isPresent())) {
+        throw new RuntimeException("Trying to use remote distribution, but frameworkUser" + "and/or frameworkSuperUser not set!");
+      }
+
+      LOGGER.info("Using remote distribution");
+      String clGeneratedCommand = clGenerator.generateCommandLine(profile, null);
+
+      String nmURIString = myriadExecutorConfiguration.getNodeManagerUri().get();
+
+      //Concatenate all the subcommands
+      String cmd = clGeneratedCommand + " " + executorCmd;
+
+      //get the nodemanagerURI
+      //We're going to extract ourselves, so setExtract is false
+      LOGGER.info("Getting Hadoop distribution from:" + nmURIString);
+      URI nmUri = URI.newBuilder().setValue(nmURIString).setExtract(false).build();
+
+      //get configs directly from resource manager
+      String configUrlString = clGenerator.getConfigurationUrl();
+      LOGGER.info("Getting config from:" + configUrlString);
+      URI configUri = URI.newBuilder().setValue(configUrlString).build();
+
+      LOGGER.info("Slave will execute command:" + cmd);
+      commandInfo.addUris(nmUri).addUris(configUri).setValue("echo \"" + cmd + "\";" + cmd);
+      commandInfo.setUser(cfg.getFrameworkSuperUser().get());
+
+    } else {
+      commandInfo.setValue(executorCmd);
+    }
+    return commandInfo.build();
+  }
+
+  @Override
+  public ExecutorInfo getExecutorInfoForSlave(FrameworkID frameworkId, Offer offer, CommandInfo commandInfo) {
+    // TODO (yufeldman) if executor specified use it , otherwise return null
+    // nothing to implement here, since we are using default slave executor
+    return null;
+  }
+
+  /**
+   * Helper method to reserve ports
+   *
+   * @param offer
+   * @param requestedPorts
+   * @return
+   */
+  private List<Long> getAvailablePorts(Offer offer, int requestedPorts) {
+    if (requestedPorts == 0) {
+      return null;
+    }
+    final List<Long> returnedPorts = new ArrayList<>();
+    for (Resource resource : offer.getResourcesList()) {
+      if (resource.getName().equals("ports")) {
+        Iterator<Value.Range> itr = resource.getRanges().getRangeList().iterator();
+        while (itr.hasNext()) {
+          Value.Range range = itr.next();
+          if (range.getBegin() <= range.getEnd()) {
+            long i = range.getBegin();
+            while (i <= range.getEnd() && returnedPorts.size() < requestedPorts) {
+              returnedPorts.add(i);
+              i++;
+            }
+            if (returnedPorts.size() >= requestedPorts) {
+              return returnedPorts;
+            }
+          }
+        }
+      }
+    }
+    // this is actually an error condition - we did not have enough ports to use
+    return returnedPorts;
+  }
+}



Mime
View raw message