WIP: Making a cluster mode a mesos framework.

Author: tnachen

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

@@ -17,37 +17,13 @@
 
 package org.apache.spark.deploy.mesos
 
-import akka.actor.{Props, ActorSystem, Actor}
-
-import scala.collection.mutable.{ArrayBuffer, HashMap}
-import scala.Some
-
-import org.apache.spark.deploy.rest.MesosRestServer
 import org.apache.spark.{Logging, SecurityManager, SparkConf}
-import org.apache.spark.util.{ActorLogReceive, AkkaUtils, IntParam, Utils}
-
-import org.apache.spark.deploy.DeployMessages._
+import org.apache.spark.util.{IntParam, Utils}
 
-import org.apache.spark.deploy.DriverDescription
-import org.apache.spark.deploy.master.DriverInfo
-import org.apache.spark.deploy.master.DriverState.DriverState
-import org.apache.spark.deploy.master.DriverState
-import org.apache.spark.deploy.worker.DriverRunner
-
-import java.io.{IOException, File}
-import java.util.Date
-import java.text.SimpleDateFormat
+import java.io.File
 import org.apache.spark.deploy.mesos.ui.MesosClusterUI
-import org.apache.spark.deploy.mesos.Messages.{DispatcherStateResponse, RequestDispatcherState}
-
-private [deploy] object Messages {
-  case object RequestDispatcherState
-
-  case class DispatcherStateResponse(
-      activeDrivers: Iterable[DriverInfo],
-      completedDrivers: Iterable[DriverInfo]) {
-  }
-}
+import org.apache.spark.deploy.rest.MesosRestServer
+import org.apache.spark.scheduler.cluster.mesos.{ClusterScheduler, MesosClusterScheduler}
 
 /*
  * A dispatcher actor that is responsible for managing drivers, that is intended to
@@ -58,221 +34,46 @@ private [deploy] object Messages {
 private [spark] class MesosClusterDispatcher(
     host: String,
     serverPort: Int,
-    actorPort: Int,
     webUiPort: Int,
-    systemName: String,
-    actorName: String,
     conf: SparkConf,
-    masterUrl: String,
-    workDirPath: Option[String] = None) extends Actor with ActorLogReceive with Logging {
-  val server = new MesosRestServer(host, serverPort, self, conf, masterUrl)
+    scheduler: ClusterScheduler) extends Logging {
+
+  val server = new MesosRestServer(host, serverPort, conf, scheduler)
 
   val dispatcherPublicAddress = {
     val envVar = System.getenv("SPARK_PUBLIC_DNS")
     if (envVar != null) envVar else host
   }
 
-  lazy val workDir: File = {
-    val dir = workDirPath.map(new File(_)).getOrElse(new File(sparkHome, "work"))
-
-    // Attempt to remove the work directory if it exists on startup.
-    // This is to avoid unbounded growing the work directory as drivers
-    // are only deleted when it is over the retained count while it's running.
-    // We don't fail startup if we are not able to remove, as this is
-    // a short-term solution.
-    try {
-      if (dir.exists()) {
-        dir.delete()
-      }
-    } catch {
-      case e: IOException =>
-        logError("Unable to remove work directory " + workDir, e)
-    }
-
-    try {
-      // This sporadically fails - not sure why ... !workDir.exists() && !workDir.mkdirs()
-      // So attempting to create and then check if directory was created or not.
-      dir.mkdirs()
-      if (!dir.exists() || !dir.isDirectory) {
-        logError("Failed to create work directory " + dir)
-        System.exit(1)
-      }
-      assert (dir.isDirectory)
-    } catch {
-      case e: Exception =>
-        logError("Failed to create work directory " + dir, e)
-        System.exit(1)
-    }
-    dir
-  }
-
   val webUi = new MesosClusterUI(
-    self, new SecurityManager(conf), webUiPort, conf, workDir, dispatcherPublicAddress)
-
-  val runners = new HashMap[String, DriverRunner]
-  val drivers = new HashMap[String, DriverInfo]
-  val completedDrivers = new ArrayBuffer[DriverInfo]
-  val RETAINED_DRIVERS = conf.getInt("spark.deploy.retainedDrivers", 200)
-  var nextDriverNumber = 0
-  val securityManager = new SecurityManager(conf)
-
-  def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")
+    new SecurityManager(conf), webUiPort, conf, dispatcherPublicAddress, scheduler)
 
   val sparkHome =
     new File(sys.env.get("SPARK_HOME").getOrElse("."))
 
-  val akkaUrl = AkkaUtils.address(
-    AkkaUtils.protocol(context.system),
-    systemName,
-    host,
-    actorPort,
-    actorName)
-
-  def newDriverId(submitDate: Date): String = {
-    val appId = "driver-%s-%04d".format(createDateFormat.format(submitDate), nextDriverNumber)
-    nextDriverNumber += 1
-    appId
-  }
-
-  def createDriver(desc: DriverDescription): DriverInfo = {
-    val now = System.currentTimeMillis()
-    val date = new Date(now)
-    new DriverInfo(now, newDriverId(date), desc, date)
-  }
-
-  override def preStart() {
+  def start() {
     server.start()
     webUi.bind()
   }
 
-  override def postStop() {
+  def stop() {
     webUi.stop()
     server.stop()
-    runners.values.foreach(_.kill())
-  }
-
-  override def receiveWithLogging = {
-    case RequestSubmitDriver(driverDescription) => {
-      val driverInfo = createDriver(driverDescription)
-      val runner = new DriverRunner(conf, driverInfo.id, workDir,
-        sparkHome, driverDescription, self, akkaUrl, securityManager)
-      runners(driverInfo.id) = runner
-      drivers(driverInfo.id) = driverInfo
-      runner.start()
-      sender ! SubmitDriverResponse(true, Option(driverInfo.id), "")
-    }
-
-    case RequestKillDriver(driverId) => {
-      if (!drivers.contains(driverId)) {
-        if (completedDrivers.exists(_.id == driverId)) {
-          sender ! KillDriverResponse(driverId, false, "Driver already completed")
-        } else {
-          sender ! KillDriverResponse(driverId, false, "Unknown driver")
-        }
-      } else {
-        runners(driverId).kill()
-        sender ! KillDriverResponse(driverId, true, "")
-      }
-    }
-
-    case RequestDriverStatus(driverId) => {
-      drivers.get(driverId).orElse(completedDrivers.find(_.id == driverId)) match {
-        case Some(driver) =>
-          sender ! DriverStatusResponse(found = true, Some(driver.state),
-            None, None, driver.exception)
-        case None =>
-          sender ! DriverStatusResponse(found = false, None, None, None, None)
-      }
-    }
-
-    case DriverStateChanged(driverId, state, exception) => {
-      logDriverChange(driverId, state, exception)
-      state match {
-        case DriverState.ERROR | DriverState.FINISHED | DriverState.KILLED | DriverState.FAILED =>
-          removeDriver(driverId, state, exception)
-        case _ =>
-          throw new Exception(s"Received unexpected state update for driver $driverId: $state")
-      }
-    }
-
-    case RequestDispatcherState => {
-      sender ! DispatcherStateResponse(drivers.values, completedDrivers)
-    }
-  }
-
-  def logDriverChange(driverId: String, state: DriverState, exception: Option[Exception]) {
-    state match {
-      case DriverState.ERROR =>
-        logWarning(s"Driver $driverId failed with unrecoverable exception: ${exception.get}")
-      case DriverState.FAILED =>
-        logWarning(s"Driver $driverId exited with failure")
-      case DriverState.FINISHED =>
-        logInfo(s"Driver $driverId exited successfully")
-      case DriverState.KILLED =>
-        logInfo(s"Driver $driverId was killed by user")
-      case _ =>
-        logDebug(s"Driver $driverId changed state to $state")
-    }
-  }
-
-  def removeDriver(driverId: String, state: DriverState, exception: Option[Exception]) {
-    if (completedDrivers.size >= RETAINED_DRIVERS) {
-      val toRemove = math.max(RETAINED_DRIVERS / 10, 1)
-      for (i <- 0 to (toRemove - 1)) {
-        val driverId = completedDrivers(i).id
-        try {
-          new File(workDir, driverId).delete()
-        } catch {
-          case e: Exception =>
-            logWarning("Unable to remove work dir for completed driver " + driverId, e)
-        }
-      }
-      completedDrivers.trimStart(toRemove)
-    }
-    val driverInfo = drivers.remove(driverId).get
-    driverInfo.exception = exception
-    driverInfo.state = state
-    completedDrivers += driverInfo
   }
 }
 
 object MesosClusterDispatcher {
   def main(args: Array[String]) {
     val conf = new SparkConf
-    val clusterArgs = new ClusterDispatcherArguments(args, conf)
-    val actorSystem = startSystemAndActor(clusterArgs)
-    Runtime.getRuntime().addShutdownHook(new Thread("MesosClusterDispatcherShutdownHook") {
-      override def run() = {
-        // Makes sure we shut down the actor, which will kill all the drivers.
-        actorSystem.shutdown()
-        actorSystem.awaitTermination()
-      }
-    })
-    actorSystem.awaitTermination()
-  }
-
-  def startSystemAndActor(
-      args: ClusterDispatcherArguments): ActorSystem = {
-    // The LocalSparkCluster runs multiple local sparkWorkerX actor systems
-    val conf = new SparkConf
-    val systemName = "spark-mesos-cluster"
-    val actorName = "MesosClusterDispatcher"
-    val securityMgr = new SecurityManager(conf)
-    val (actorSystem, boundPort) = AkkaUtils.createActorSystem(
-      systemName, args.host, 0, conf, securityMgr)
-    actorSystem.actorOf(
-      Props(classOf[MesosClusterDispatcher],
-        args.host,
-        args.port,
-        boundPort,
-        args.webUiPort,
-        systemName,
-        actorName,
-        conf,
-        args.masterUrl,
-        None),
-      name = actorName)
-    actorSystem
+    val dispatcherArgs = new ClusterDispatcherArguments(args, conf)
+    val scheduler = new MesosClusterScheduler(conf)
+    scheduler.start()
+    new MesosClusterDispatcher(
+      dispatcherArgs.host,
+      dispatcherArgs.port,
+      dispatcherArgs.webUiPort,
+      conf,
+      scheduler).start()
   }
 
   class ClusterDispatcherArguments(args: Array[String], conf: SparkConf) {

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

@@ -20,62 +20,58 @@ 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.deploy.mesos.Messages.{DispatcherStateResponse, RequestDispatcherState}
-import scala.concurrent.Await
 import org.apache.spark.deploy.master.DriverInfo
+import org.apache.spark.scheduler.cluster.mesos.{ClusterTaskState, DriverSubmission, ClusterScheduler}
 
 class DriverOutputPage(parent: MesosClusterUI) extends WebUIPage("") {
-  private val dispatcher = parent.dispatcherActorRef
-  private val timeout = parent.timeout
 
   def render(request: HttpServletRequest): Seq[Node] = {
-    val stateFuture = (dispatcher ? RequestDispatcherState)(timeout).mapTo[DispatcherStateResponse]
-    val state = Await.result(stateFuture, timeout)
+    val state = parent.scheduler.getState()
 
-    val driverHeaders = Seq("DriverID", "Submit Date", "Start Date", "Logs")
-    val completedDriverHeaders = driverHeaders ++ Seq("State", "Exception")
-    val driverTable = UIUtils.listingTable(driverHeaders, driverRow, state.activeDrivers)
-    val completedDriverTable =
-      UIUtils.listingTable(completedDriverHeaders, completedDriverRow, state.completedDrivers)
+    val queuedHeaders = Seq("DriverID", "Submit Date", "Description")
+    val driverHeaders = queuedHeaders ++
+      Seq("Start Date", "Mesos Slave ID", "Mesos Task ID", "State", "Logs")
+
+    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 =
       <div class="row-fluid">
         <div class="span12">
-          <h4>Running Drivers</h4>
-          {driverTable}
-          <h4>Finished Drivers</h4>
-          {completedDriverTable}
+          <h4>Queued Drivers:</h4>
+          {queuedTable}
+          <h4>Launched Drivers:</h4>
+          {launchedTable}
+          <h4>Finished Drivers:</h4>
+          {finishedTable}
         </div>
       </div>;
     UIUtils.basicSparkPage(content, "Spark Drivers for Mesos cluster")
+
+    null
   }
 
-  def driverRow(info: DriverInfo): Seq[Node] = {
+  def queuedRow(submission: DriverSubmission): Seq[Node] = {
     <tr>
-      <td>{info.id}</td>
-      <td>{info.submitDate}</td>
-      <td>{info.startTime}</td>
-      <td>
-        <a href={"logPage?driverId=%s&logType=stdout"
-          .format(info.id)}>stdout</a>,
-        <a href={"logPage?driverId=%s&logType=stderr"
-          .format(info.id)}>stderr</a>
-      </td>
+      <td>{submission.submissionId}</td>
+      <td>{submission.submitDate}</td>
+      <td>{submission.desc.command.mainClass}</td>
     </tr>
   }
 
-  def completedDriverRow(info: DriverInfo): Seq[Node] = {
+  def driverRow(state: ClusterTaskState): Seq[Node] = {
     <tr>
-      <td>{info.id}</td>
-      <td>{info.submitDate}</td>
-      <td>{info.startTime}</td>
+      <td>{state.submission.submissionId}</td>
+      <td>{state.submission.submitDate}</td>
+      <td>{state.submission.desc.command.mainClass}</td>
+      <td>{state.startDate}</td>
+      <td>{state.slaveId}</td>
+      <td>{state.taskId}</td>
+      <td>{state.taskState}</td>
       <td>
-        <a href={"logPage?driverId=%s&logType=stdout"
-          .format(info.id)}>stdout</a>,
-        <a href={"logPage?driverId=%s&logType=stderr"
-          .format(info.id)}>stderr</a>
+        <a href={""}>stdout</a>,
+        <a href={""}>stderr</a>
       </td>
-      <td>{info.state}</td>
-      <td>{info.exception}</td>
     </tr>
   }
 }