Launch task through the mesos scheduler

Author: Luc Bourlier

core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala

@@ -66,6 +66,8 @@ object MesosClusterDispatcher {
   def main(args: Array[String]) {
     val conf = new SparkConf
     val dispatcherArgs = new ClusterDispatcherArguments(args, conf)
+    conf.setMaster(dispatcherArgs.masterUrl)
+    conf.setAppName("Mesos Cluster Dispatcher")
     val scheduler = new MesosClusterScheduler(conf)
     scheduler.start()
     new MesosClusterDispatcher(
@@ -74,6 +76,10 @@ object MesosClusterDispatcher {
       dispatcherArgs.webUiPort,
       conf,
       scheduler).start()
+    this.synchronized {
+      // block indefinitely
+      this.wait() // TODO: bad
+    }
   }
 
   class ClusterDispatcherArguments(args: Array[String], conf: SparkConf) {
@@ -103,7 +109,7 @@ object MesosClusterDispatcher {
           System.err.println("Cluster dispatcher only supports mesos (uri begins with mesos://)")
           System.exit(1)
         }
-        masterUrl = value
+        masterUrl = value.stripPrefix("mesos://")
         parse(tail)
 
       case ("--help") :: tail =>

core/src/main/scala/org/apache/spark/deploy/mesos/ui/DriverOutputPage.scala

@@ -55,15 +55,15 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
     <tr>
       <td>{submission.submissionId}</td>
       <td>{submission.submitDate}</td>
-      <td>{submission.desc.command.mainClass}</td>
+      <td>{submission.req.desc.command.mainClass}</td>
     </tr>
   }
 
   def driverRow(state: ClusterTaskState): Seq[Node] = {
     <tr>
       <td>{state.submission.submissionId}</td>
       <td>{state.submission.submitDate}</td>
-      <td>{state.submission.desc.command.mainClass}</td>
+      <td>{state.submission.req.desc.command.mainClass}</td>
       <td>{state.startDate}</td>
       <td>{state.slaveId}</td>
       <td>{state.taskId}</td>

core/src/main/scala/org/apache/spark/deploy/rest/MesosRestServer.scala

@@ -23,10 +23,10 @@ import javax.servlet.http.HttpServletResponse
 import org.apache.spark.deploy.DriverDescription
 import org.apache.spark.deploy.ClientArguments._
 import org.apache.spark.deploy.Command
-
 import org.apache.spark.{SparkConf, SPARK_VERSION => sparkVersion}
 import org.apache.spark.util.Utils
 import org.apache.spark.scheduler.cluster.mesos.ClusterScheduler
+import org.apache.spark.scheduler.cluster.mesos.DriverRequest
 
 /**
  * A server that responds to requests submitted by the [[RestClient]].
@@ -74,7 +74,7 @@ class MesosSubmitRequestServlet(
    * This does not currently consider fields used by python applications since python
    * is not supported in mesos cluster mode yet.
    */
-  private def buildDriverDescription(request: CreateSubmissionRequest): DriverDescription = {
+  private def buildDriverRequest(request: CreateSubmissionRequest): DriverRequest = {
     // Required fields, including the main class because python is not yet supported
     val appResource = Option(request.appResource).getOrElse {
       throw new SubmitRestMissingFieldException("Application jar is missing.")
@@ -109,8 +109,9 @@ class MesosSubmitRequestServlet(
     val actualDriverMemory = driverMemory.map(Utils.memoryStringToMb).getOrElse(DEFAULT_MEMORY)
     val actualDriverCores = driverCores.map(_.toInt).getOrElse(DEFAULT_CORES)
 
-    new DriverDescription(
-      appResource, actualDriverMemory, actualDriverCores, actualSuperviseDriver, command)
+    DriverRequest(new DriverDescription(
+      appResource, actualDriverMemory, actualDriverCores, actualSuperviseDriver, command),
+      conf)
   }
 
   protected override def handleSubmit(
@@ -119,7 +120,7 @@ class MesosSubmitRequestServlet(
       responseServlet: HttpServletResponse): SubmitRestProtocolResponse = {
     requestMessage match {
       case submitRequest: CreateSubmissionRequest =>
-        val driverDescription = buildDriverDescription(submitRequest)
+        val driverDescription = buildDriverRequest(submitRequest)
         val response = scheduler.submitDriver(driverDescription)
         val submitResponse = new CreateSubmissionResponse
         submitResponse.serverSparkVersion = sparkVersion

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala

@@ -18,30 +18,28 @@
 package org.apache.spark.scheduler.cluster.mesos
 
 import java.text.SimpleDateFormat
-
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.{LinkedBlockingQueue, TimeUnit}
 import java.util.{List => JList}
 import java.util.{Collections, Date}
-
 import org.apache.mesos.{SchedulerDriver, Scheduler}
 import org.apache.mesos.Protos._
-
 import org.apache.spark.deploy.DriverDescription
 import org.apache.spark.deploy.master.DriverState
 import org.apache.spark.deploy.master.DriverState.DriverState
-
 import org.apache.spark.SparkConf
 import org.apache.spark.util.Utils
-
 import scala.collection.mutable
 import scala.collection.JavaConversions._
 import org.apache.mesos.Protos.Environment.Variable
+import org.apache.spark.SparkException
+import java.io.File
 
+case class DriverRequest(desc: DriverDescription, conf: SparkConf)
 
 private[spark] class DriverSubmission(
     val submissionId: String,
-    val desc: DriverDescription,
+    val req: DriverRequest,
     val submitDate: Date) {
 
   def canEqual(other: Any): Boolean = other.isInstanceOf[DriverSubmission]
@@ -76,7 +74,7 @@ private[spark] case class ClusterSchedulerState(
     finishedDrivers: Iterable[ClusterTaskState])
 
 private[spark] trait ClusterScheduler {
-  def submitDriver(desc: DriverDescription): SubmitResponse
+  def submitDriver(desc: DriverRequest): SubmitResponse
   def killDriver(submissionId: String): KillResponse
   def getStatus(submissionId: String): StatusResponse
   def getState(): ClusterSchedulerState
@@ -88,7 +86,6 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
   val master = conf.get("spark.master")
   val appName = conf.get("spark.app.name")
   val capacity = conf.getInt("spark.mesos.driver.capacity", 200)
-  val executorUri = conf.getOption("spark.executor.uri")
   val stateLock = new Object
   val launchedDrivers = new mutable.HashMap[String, ClusterTaskState]()
 
@@ -105,10 +102,10 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
         createDateFormat.format(submitDate), nextDriverNumber.incrementAndGet())
   }
 
-  def submitDriver(desc: DriverDescription): SubmitResponse = {
+  def submitDriver(req: DriverRequest): SubmitResponse = {
     val submitDate: Date = new Date()
     val submissionId: String = newDriverId(submitDate)
-    val submission = new DriverSubmission(submissionId, desc, submitDate)
+    val submission = new DriverSubmission(submissionId, req, submitDate)
     if (queue.offer(submission)) {
       SubmitResponse(submissionId, true, None)
     } else {
@@ -147,9 +144,16 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
     markRegistered()
   }
 
-  private def buildCommand(desc: DriverDescription): CommandInfo = {
+  private def buildCommand(req: DriverRequest): CommandInfo = {
+
+    val desc = req.desc
+
+    val cleanedJarUrl = desc.jarUrl.stripPrefix("file:")
+
+    logInfo(s"jarUrl: $cleanedJarUrl")
+
     val builder = CommandInfo.newBuilder()
-      .addUris(CommandInfo.URI.newBuilder().setValue(desc.jarUrl).build())
+      .addUris(CommandInfo.URI.newBuilder().setValue(cleanedJarUrl).build())
 
     val entries =
       (conf.getOption("spark.executor.extraLibraryPath").toList ++ desc.command.libraryPathEntries)
@@ -160,11 +164,7 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
       ""
     }
 
-    val stringBuilder = new StringBuilder
-    stringBuilder
-      .append(desc.command.mainClass)
-      .append(" ")
-      .append(desc.command.arguments.mkString("\"", "\"", "\""))
+    // TODO: add support for more spark-submit parameters
 
     val envBuilder = Environment.newBuilder()
     desc.command.environment.foreach {
@@ -175,14 +175,35 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
 
     builder.setEnvironment(envBuilder.build())
 
+    val executorUri = req.conf.getOption("spark.executor.uri")
     if (executorUri.isDefined) {
       builder.addUris(CommandInfo.URI.newBuilder().setValue(executorUri.get).build())
+
       val basename = executorUri.get.split('/').last.split('.').head
+      val cmd =
+        Seq("bin/spark-submit",
+            "--class", desc.command.mainClass,
+            "--master", s"mesos://${conf.get("spark.master")}",
+            s"../${desc.jarUrl.split("/").last}")
+            .mkString(" ")
+
       builder.setValue(
-        s"cd $basename*; $prefixEnv ${stringBuilder.toString()}")
+        s"cd $basename*; $prefixEnv $cmd")
     } else {
+      val executorSparkHome = req.conf.getOption("spark.mesos.executor.home")
+        .getOrElse {
+          throw new SparkException("Executor Spark home `spark.mesos.executor.home` is not set!")
+        }
+
+      val cmd =
+        Seq(new File(executorSparkHome, "./bin/spark-submit"),
+            "--class", desc.command.mainClass,
+            "--master", s"mesos://${conf.get("spark.master")}",
+            desc.jarUrl.split("/").last)
+            .mkString(" ")
+
       builder.setValue(
-        s"$prefixEnv ${stringBuilder.toString()}")
+        s"$prefixEnv $cmd")
     }
 
     builder.build
@@ -203,8 +224,8 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
 
     var remainingOffers = offers
 
-    val driverCpu = submission.desc.cores
-    val driverMem = submission.desc.mem
+    val driverCpu = submission.req.desc.cores
+    val driverMem = submission.req.desc.mem
 
     // Should use the passed in driver cpu and memory.
     val offerOption = offers.find { o =>
@@ -216,15 +237,18 @@ private[spark] class MesosClusterScheduler(conf: SparkConf)
       val taskId = TaskID.newBuilder().setValue(submission.submissionId).build()
 
       val cpuResource = Resource.newBuilder()
-        .setName("cpus").setScalar(Value.Scalar.newBuilder().setValue(driverCpu)).build()
+        .setName("cpus").setType(Value.Type.SCALAR)
+        .setScalar(Value.Scalar.newBuilder().setValue(driverCpu)).build()
 
       val memResource = Resource.newBuilder()
-        .setName("mem").setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()
+        .setName("mem").setType(Value.Type.SCALAR)
+        .setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()
 
-      val commandInfo = buildCommand(submission.desc)
+      val commandInfo = buildCommand(submission.req)
 
       val taskInfo = TaskInfo.newBuilder()
         .setTaskId(taskId)
+        .setName(s"driver for ${submission.req.desc.command.mainClass}")
         .setSlaveId(offer.getSlaveId)
         .setCommand(commandInfo)
         .addResources(cpuResource)

core/src/test/scala/org/apache/spark/scheduler/mesos/MesosClusterSchedulerSuite.scala

@@ -21,6 +21,7 @@ import org.scalatest.FunSuite
 import org.apache.spark.{SparkConf, LocalSparkContext}
 import org.scalatest.mock.MockitoSugar
 import org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler
+import org.apache.spark.scheduler.cluster.mesos.DriverRequest
 import org.apache.spark.deploy.{Command, DriverDescription}
 
 class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with MockitoSugar {
@@ -35,11 +36,11 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     conf.setAppName("spark mesos")
     val scheduler = new MesosClusterScheduler(conf)
     val response =
-      scheduler.submitDriver(new DriverDescription("jar", 1000, 1, true, createCommand))
+      scheduler.submitDriver(DriverRequest(new DriverDescription("jar", 1000, 1, true, createCommand), new SparkConf))
     assert(response.success)
 
     val response2 =
-      scheduler.submitDriver(new DriverDescription("jar", 1000, 1, true, createCommand))
+      scheduler.submitDriver(DriverRequest(new DriverDescription("jar", 1000, 1, true, createCommand), new SparkConf))
     assert(response2.success)
 
     val state = scheduler.getState()
@@ -53,7 +54,7 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     conf.setAppName("spark mesos")
     val scheduler = new MesosClusterScheduler(conf)
     val response =
-      scheduler.submitDriver(new DriverDescription("jar", 1000, 1, true, createCommand))
+      scheduler.submitDriver(DriverRequest(new DriverDescription("jar", 1000, 1, true, createCommand), new SparkConf))
     assert(response.success)
 
     val killResponse = scheduler.killDriver(response.id)