Add supervise support and persist retries.

Author: tnachen

core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala

@@ -32,7 +32,7 @@ import org.json4s._
 import org.json4s.jackson.JsonMethods
 
 import org.apache.spark.{Logging, SparkConf, SparkContext}
-import org.apache.spark.deploy.master.{RecoveryState, SparkCuratorUtil}
+import org.apache.spark.deploy.master.{RecoveryState}
 import org.apache.spark.util.Utils
 
 /**

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -381,11 +381,13 @@ object SparkSubmit {
 
       // Standalone cluster only
       // Do not set CL arguments here because there are multiple possibilities for the main class
-      OptionAssigner(args.jars, STANDALONE, CLUSTER, sysProp = "spark.jars"),
+      OptionAssigner(args.jars, STANDALONE | MESOS, CLUSTER, sysProp = "spark.jars"),
       OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy"),
-      OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, sysProp = "spark.driver.memory"),
-      OptionAssigner(args.driverCores, STANDALONE, CLUSTER, sysProp = "spark.driver.cores"),
-      OptionAssigner(args.supervise.toString, STANDALONE, CLUSTER,
+      OptionAssigner(args.driverMemory, STANDALONE | MESOS, CLUSTER,
+        sysProp = "spark.driver.memory"),
+      OptionAssigner(args.driverCores, STANDALONE | MESOS, CLUSTER,
+        sysProp = "spark.driver.cores"),
+      OptionAssigner(args.supervise.toString, STANDALONE | MESOS, CLUSTER,
         sysProp = "spark.driver.supervise"),
 
       // Yarn client only

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

@@ -20,7 +20,7 @@ package org.apache.spark.deploy.mesos.ui
 import org.apache.spark.ui.{UIUtils, WebUIPage}
 import javax.servlet.http.HttpServletRequest
 import scala.xml.Node
-import org.apache.spark.scheduler.cluster.mesos.{ClusterTaskState, DriverSubmission}
+import org.apache.spark.scheduler.cluster.mesos.{RetryState, ClusterTaskState, DriverSubmission}
 import org.apache.mesos.Protos.TaskStatus
 
 class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
@@ -31,10 +31,13 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
     val queuedHeaders = Seq("DriverID", "Submit Date", "Description")
     val driverHeaders = queuedHeaders ++
       Seq("Start Date", "Mesos Slave ID", "State")
+    val retryHeaders = Seq("DriverID", "Submit Date", "Description") ++
+      Seq("Last Failed Status", "Next Retry Time", "Attempt Count")
 
     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 retryTable = UIUtils.listingTable(retryHeaders, retryRow, state.retryList)
     val content =
       <p>Mesos Framework ID: {state.appId}</p>
       <div class="row-fluid">
@@ -45,6 +48,8 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
           {launchedTable}
           <h4>Finished Drivers:</h4>
           {finishedTable}
+          <h4>Supervise drivers waiting for retry:</h4>
+          {retryTable}
         </div>
       </div>;
     UIUtils.basicSparkPage(content, "Spark Drivers for Mesos cluster")
@@ -69,6 +74,17 @@ class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
     </tr>
   }
 
+  def retryRow(state: RetryState): Seq[Node] = {
+    <tr>
+      <td>{state.submission.submissionId}</td>
+      <td>{state.submission.submitDate}</td>
+      <td>{state.submission.desc.desc.command.mainClass}</td>
+      <td>{state.lastFailureStatus}</td>
+      <td>{state.nextRetry}</td>
+      <td>{state.retries}</td>
+    </tr>
+  }
+
   def stateString(status: Option[TaskStatus]): String = {
     if (status.isEmpty) {
       return ""

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

@@ -167,7 +167,7 @@ class MesosKillRequestServlet(scheduler: ClusterScheduler, conf: SparkConf)
     val response = scheduler.killDriver(submissionId)
     val k = new KillSubmissionResponse
     k.serverSparkVersion = sparkVersion
-    k.message = response.message.orNull
+    k.message = response.message
     k.submissionId = submissionId
     k.success = response.success
     k

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

@@ -18,7 +18,6 @@
 package org.apache.spark.scheduler.cluster.mesos
 
 import scala.collection.mutable
-import javax.annotation.concurrent.NotThreadSafe
 
 /**
  * A request queue for launching drivers in Mesos cluster mode.
@@ -27,8 +26,7 @@ import javax.annotation.concurrent.NotThreadSafe
  * This queue is also bounded and rejects offers when it's full.
  * @param state Mesos state abstraction to fetch persistent state.
  */
-@NotThreadSafe
-class DriverQueue(state: ClusterPersistenceEngine, capacity: Int) {
+private[mesos] class DriverQueue(state: ClusterPersistenceEngine, capacity: Int) {
   var queue: mutable.Queue[DriverSubmission] = new mutable.Queue[DriverSubmission]()
   private var count = 0
 

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

@@ -18,9 +18,13 @@
 package org.apache.spark.scheduler.cluster.mesos
 
 import scala.collection.mutable
-import org.apache.mesos.Protos.{TaskID, SlaveID}
+import org.apache.mesos.Protos.SlaveID
 
-class LaunchedDrivers(state: ClusterPersistenceEngine) {
+/**
+ * Tracks all the launched or running drivers in the Mesos cluster scheduler.
+ * @param state Persistence engine to store state.
+ */
+private[mesos] class LaunchedDrivers(state: ClusterPersistenceEngine) {
   private val drivers = new mutable.HashMap[String, ClusterTaskState]
 
   // Holds the list of tasks that needs to reconciliation from the master.