Handle cluster mode recovery and state persistence.

Author: tnachen

core/src/main/scala/org/apache/spark/deploy/master/SparkCuratorUtil.scala renamed to core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.spark.deploy.master
+package org.apache.spark.deploy
 
 import scala.collection.JavaConversions._
 
@@ -25,7 +25,7 @@ import org.apache.zookeeper.KeeperException
 
 import org.apache.spark.{Logging, SparkConf}
 
-private[deploy] object SparkCuratorUtil extends Logging {
+private[spark] object SparkCuratorUtil extends Logging {
 
   private val ZK_CONNECTION_TIMEOUT_MILLIS = 15000
   private val ZK_SESSION_TIMEOUT_MILLIS = 60000

core/src/main/scala/org/apache/spark/deploy/master/DriverState.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy.master
 
-private[deploy] object DriverState extends Enumeration {
+private[spark] object DriverState extends Enumeration {
 
   type DriverState = Value
 

core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala

@@ -23,6 +23,7 @@ import org.apache.spark.{Logging, SparkConf}
 import org.apache.spark.deploy.master.MasterMessages._
 import org.apache.curator.framework.CuratorFramework
 import org.apache.curator.framework.recipes.leader.{LeaderLatchListener, LeaderLatch}
+import org.apache.spark.deploy.SparkCuratorUtil
 
 private[master] class ZooKeeperLeaderElectionAgent(val masterActor: LeaderElectable,
     conf: SparkConf) extends LeaderLatchListener with LeaderElectionAgent with Logging  {

core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala

@@ -26,6 +26,7 @@ import org.apache.curator.framework.CuratorFramework
 import org.apache.zookeeper.CreateMode
 
 import org.apache.spark.{Logging, SparkConf}
+import org.apache.spark.deploy.SparkCuratorUtil
 
 
 private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serialization: Serialization)

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

@@ -18,15 +18,16 @@
 package org.apache.spark.deploy.mesos
 
 import org.apache.spark.{Logging, SecurityManager, SparkConf}
-import org.apache.spark.util.{IntParam, Utils}
+import org.apache.spark.util.{SignalLogger, IntParam, Utils}
 
 import java.io.File
-import java.util.concurrent.CountDownLatch
+import java.util.concurrent.{TimeUnit, CountDownLatch}
 
 import org.apache.spark.deploy.mesos.ui.MesosClusterUI
 import org.apache.spark.deploy.rest.MesosRestServer
-import org.apache.spark.scheduler.cluster.mesos.{ClusterScheduler, MesosClusterScheduler}
-
+import org.apache.spark.scheduler.cluster.mesos._
+import org.apache.spark.deploy.mesos.MesosClusterDispatcher.ClusterDispatcherArguments
+import org.apache.mesos.state.{ZooKeeperState, InMemoryState}
 
 /*
  * A dispatcher actor that is responsible for managing drivers, that is intended to
@@ -35,76 +36,101 @@ import org.apache.spark.scheduler.cluster.mesos.{ClusterScheduler, MesosClusterS
  * a daemon to launch drivers as Mesos frameworks upon request.
  */
 private [spark] class MesosClusterDispatcher(
-    host: String,
-    serverPort: Int,
-    conf: SparkConf,
-    webUi: MesosClusterUI,
-    scheduler: ClusterScheduler) extends Logging {
+    args: ClusterDispatcherArguments,
+    conf: SparkConf) extends Logging {
+
+  def dispatcherPublicAddress(conf: SparkConf, host: String): String = {
+    val envVar = conf.getenv("SPARK_PUBLIC_DNS")
+    if (envVar != null) envVar else host
+  }
+
+  val recoveryMode = conf.get("spark.deploy.recoveryMode", "NONE").toUpperCase()
+  logInfo("Recovery mode in Mesos dispatcher set to: " + recoveryMode)
+
+  val engineFactory = recoveryMode match {
+    case "NONE" => new BlackHolePersistenceEngineFactory
+    case "ZOOKEEPER" => {
+      new ZookeeperClusterPersistenceEngineFactory(conf)
+    }
+  }
 
-  val server = new MesosRestServer(host, serverPort, conf, scheduler)
+  val scheduler = new MesosClusterScheduler(
+    engineFactory,
+    conf)
+
+  val server = new MesosRestServer(args.host, args.port, conf, scheduler)
+
+  val webUi = new MesosClusterUI(
+    new SecurityManager(conf),
+    args.webUiPort,
+    conf,
+    dispatcherPublicAddress(conf, args.host),
+    scheduler)
+
+  val shutdownLatch = new CountDownLatch(1)
 
   val sparkHome =
     new File(sys.env.get("SPARK_HOME").getOrElse("."))
 
   def start() {
+    webUi.bind()
+    scheduler.frameworkUrl = webUi.activeWebUiUrl
+    scheduler.start()
     server.start()
-    // We assume web ui is already started as the scheduler needs the bound port.
+  }
+
+  def awaitShutdown() {
+    shutdownLatch.await()
   }
 
   def stop() {
     webUi.stop()
     server.stop()
+    scheduler.stop()
+    shutdownLatch.countDown()
   }
 }
 
-object MesosClusterDispatcher {
-  def dispatcherPublicAddress(conf: SparkConf, host: String): String = {
-    val envVar = conf.getenv("SPARK_PUBLIC_DNS")
-    if (envVar != null) envVar else host
-  }
-
-  val shutdownLatch = new CountDownLatch(1)
-
+private[mesos] object MesosClusterDispatcher extends Logging {
   def main(args: Array[String]) {
+    SignalLogger.register(log)
+
     val conf = new SparkConf
     val dispatcherArgs = new ClusterDispatcherArguments(args, conf)
+
     conf.setMaster(dispatcherArgs.masterUrl)
     conf.setAppName("Mesos Cluster Dispatcher")
-    val scheduler = new MesosClusterScheduler(conf)
-
-    // We have to create the webui and bind it early as we need to
-    // pass the framework web ui url to Mesos which is before the
-    // scheduler starts.
-    val webUi =
-      new MesosClusterUI(
-        new SecurityManager(conf),
-        dispatcherArgs.webUiPort,
-        conf,
-        dispatcherPublicAddress(conf, dispatcherArgs.host),
-        scheduler)
 
-    webUi.bind()
+    val dispatcher = new MesosClusterDispatcher(
+      dispatcherArgs,
+      conf)
 
-    scheduler.frameworkUrl = webUi.activeWebUiUrl
-    scheduler.start()
-    new MesosClusterDispatcher(
-      dispatcherArgs.host,
-      dispatcherArgs.port,
-      conf,
-      webUi,
-      scheduler).start()
+    dispatcher.start()
 
-    shutdownLatch.await()
+    val shutdownHook = new Thread() {
+      override def run() {
+        logInfo("Shutdown hook is shutting down dispatcher")
+        dispatcher.stop()
+        dispatcher.awaitShutdown()
+      }
+    }
+
+    Runtime.getRuntime.addShutdownHook(shutdownHook)
+
+    dispatcher.awaitShutdown()
   }
 
   class ClusterDispatcherArguments(args: Array[String], conf: SparkConf) {
     var host = Utils.localHostName()
     var port = 7077
     var webUiPort = 8081
-    var masterUrl: String = null
+    var masterUrl: String = _
+    var propertiesFile: String = _
 
     parse(args.toList)
 
+    propertiesFile = Utils.loadDefaultSparkProperties(conf, propertiesFile)
+
     def parse(args: List[String]): Unit = args match {
       case ("--host" | "-h") :: value :: tail =>
         Utils.checkHost(value, "Please use hostname " + value)
@@ -127,6 +153,10 @@ object MesosClusterDispatcher {
         masterUrl = value.stripPrefix("mesos://")
         parse(tail)
 
+      case ("--properties-file") :: value :: tail =>
+        propertiesFile = value
+        parse(tail)
+
       case ("--help") :: tail =>
         printUsageAndExit(0)
 
@@ -152,7 +182,9 @@ object MesosClusterDispatcher {
           "  -h HOST, --host HOST   Hostname to listen on\n" +
           "  -p PORT, --port PORT   Port to listen on (default: 7077)\n" +
           "  --webui-port WEBUI_PORT   WebUI Port to listen on (default: 8081)\n" +
-          "  -m --master MASTER      URI for connecting to Mesos master\n")
+          "  -m --master MASTER      URI for connecting to Mesos master\n" +
+          "  --properties-file FILE Path to a custom Spark properties file.\n" +
+          "                         Default is conf/spark-defaults.conf.")
       System.exit(exitCode)
     }
   }

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

@@ -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
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos
+
+import org.apache.spark.deploy.DriverDescription
+import scala.collection.mutable
+
+private[spark] class MesosDriverDescription(
+    val desc: DriverDescription,
+    val schedulerProperties: mutable.HashMap[String, String]) extends Serializable {
+}

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

@@ -30,12 +30,13 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
 
     val queuedHeaders = Seq("DriverID", "Submit Date", "Description")
     val driverHeaders = queuedHeaders ++
-      Seq("Start Date", "Mesos Slave ID", "Mesos Task ID", "State")
+      Seq("Start Date", "Mesos Slave ID", "State")
 
     val queuedTable = UIUtils.listingTable(queuedHeaders, queuedRow, state.queuedDrivers)
     val launchedTable = UIUtils.listingTable(driverHeaders, driverRow, state.launchedDrivers)
     val finishedTable = UIUtils.listingTable(driverHeaders, driverRow, state.finishedDrivers)
     val content =
+      <p>Mesos Framework ID: {state.appId}</p>
       <div class="row-fluid">
         <div class="span12">
           <h4>Queued Drivers:</h4>
@@ -53,18 +54,17 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
     <tr>
       <td>{submission.submissionId}</td>
       <td>{submission.submitDate}</td>
-      <td>{submission.req.desc.command.mainClass}</td>
+      <td>{submission.desc.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.req.desc.command.mainClass}</td>
+      <td>{state.submission.desc.desc.command.mainClass}</td>
       <td>{state.startDate}</td>
       <td>{state.slaveId.getValue}</td>
-      <td>{state.taskId.getValue}</td>
       <td>{stateString(state.taskState)}</td>
     </tr>
   }

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

@@ -21,7 +21,6 @@ import org.apache.spark.ui.{SparkUI, WebUI}
 import org.apache.spark.SparkConf
 import org.apache.spark.SecurityManager
 import org.apache.spark.ui.JettyUtils._
-import org.apache.spark.deploy.worker.ui.ActiveWebUiUrlAccessor
 import org.apache.spark.scheduler.cluster.mesos.ClusterScheduler
 
 /**
@@ -33,7 +32,7 @@ private [spark] class MesosClusterUI(
     conf: SparkConf,
     dispatcherPublicAddress: String,
     val scheduler: ClusterScheduler)
-  extends WebUI(securityManager, port, conf) with ActiveWebUiUrlAccessor {
+  extends WebUI(securityManager, port, conf) {
 
   initialize()
 

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

@@ -26,7 +26,8 @@ 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
+import scala.collection.mutable
+import org.apache.spark.deploy.mesos.MesosDriverDescription
 
 /**
  * A server that responds to requests submitted by the [[RestClient]].
@@ -74,7 +75,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 buildDriverRequest(request: CreateSubmissionRequest): DriverRequest = {
+  private def buildDriverDescription(request: CreateSubmissionRequest): MesosDriverDescription = {
     // 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.")
@@ -93,6 +94,25 @@ class MesosSubmitRequestServlet(
     val driverCores = sparkProperties.get("spark.driver.cores")
     val appArgs = request.appArgs
     val environmentVariables = request.environmentVariables
+    val schedulerProperties = new mutable.HashMap[String, String]
+    // Store Spark submit specific arguments here to pass to the scheduler.
+    schedulerProperties("spark.app.name") = sparkProperties.getOrElse("spark.app.name", mainClass)
+
+    sparkProperties.get("spark.executor.memory").foreach {
+      v => schedulerProperties("spark.executor.memory") = v
+    }
+
+    sparkProperties.get("spark.cores.max").foreach {
+      v => schedulerProperties("spark.cores.max") = v
+    }
+
+    sparkProperties.get("spark.executor.uri").foreach {
+      v => schedulerProperties("spark.executor.uri") = v
+    }
+
+    sparkProperties.get("spark.mesos.executor.home").foreach {
+      v => schedulerProperties("spark.mesos.executor.home") = v
+    }
 
     // Construct driver description
     val conf = new SparkConf(false)
@@ -109,9 +129,10 @@ class MesosSubmitRequestServlet(
     val actualDriverMemory = driverMemory.map(Utils.memoryStringToMb).getOrElse(DEFAULT_MEMORY)
     val actualDriverCores = driverCores.map(_.toInt).getOrElse(DEFAULT_CORES)
 
-    DriverRequest(new DriverDescription(
-      appResource, actualDriverMemory, actualDriverCores, actualSuperviseDriver, command),
-      conf)
+    val desc = new DriverDescription(
+      appResource, actualDriverMemory, actualDriverCores, actualSuperviseDriver, command)
+
+    new MesosDriverDescription(desc, schedulerProperties)
   }
 
   protected override def handleSubmit(
@@ -120,7 +141,7 @@ class MesosSubmitRequestServlet(
       responseServlet: HttpServletResponse): SubmitRestProtocolResponse = {
     requestMessage match {
       case submitRequest: CreateSubmissionRequest =>
-        val driverDescription = buildDriverRequest(submitRequest)
+        val driverDescription = buildDriverDescription(submitRequest)
         val response = scheduler.submitDriver(driverDescription)
         val submitResponse = new CreateSubmissionResponse
         submitResponse.serverSparkVersion = sparkVersion

core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala

@@ -44,8 +44,7 @@ private[deploy] class DriverRunner(
     val sparkHome: File,
     val driverDesc: DriverDescription,
     val worker: ActorRef,
-    val workerUrl: String,
-    val securityManager: SecurityManager)
+    val workerUrl: String)
   extends Logging {
 
   @volatile private var process: Option[Process] = None
@@ -137,9 +136,12 @@ private[deploy] class DriverRunner(
    * Will throw an exception if there are errors downloading the jar.
    */
   private def downloadUserJar(driverDir: File): String = {
+
     val jarPath = new Path(driverDesc.jarUrl)
 
     val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
+    val jarFileSystem = jarPath.getFileSystem(hadoopConf)
+
     val destPath = new File(driverDir.getAbsolutePath, jarPath.getName)
     val jarFileName = jarPath.getName
     val localJarFile = new File(driverDir, jarFileName)