Address comments and add logging.

Author: tnachen

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

@@ -130,7 +130,7 @@ private[master] class Master(
   private val restServer =
     if (restServerEnabled) {
       val port = conf.getInt("spark.master.rest.port", 6066)
-      Some(new StandaloneRestServer(host, port, self, masterUrl, conf))
+      Some(new StandaloneRestServer(host, port, conf, self, masterUrl))
     } else {
       None
     }

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

@@ -30,9 +30,9 @@ import org.apache.spark.{Logging, SecurityManager, SparkConf}
 
 /*
  * A dispatcher that is responsible for managing and launching drivers, and is intended to
- * be used for Mesos cluster mode. The dispatcher ls launched by the user in the cluster,
+ * be used for Mesos cluster mode. The dispatcher is launched by the user in the cluster,
  * which it launches a [[MesosRestServer]] for listening for driver requests, and launches a
- * [[MesoClusterScheduler]] to launch these drivers in the Mesos cluster.
+ * [[MesosClusterScheduler]] to launch these drivers in the Mesos cluster.
  *
  * A typical new driver lifecycle is the following:
  *
@@ -141,7 +141,7 @@ private[mesos] object MesosClusterDispatcher extends spark.Logging {
 
     propertiesFile = Utils.loadDefaultSparkProperties(conf, propertiesFile)
 
-    def parse(args: List[String]): Unit = args match {
+    private def parse(args: List[String]): Unit = args match {
       case ("--host" | "-h") :: value :: tail =>
         Utils.checkHost(value, "Please use hostname " + value)
         host = value

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

@@ -19,8 +19,6 @@ package org.apache.spark.deploy.mesos
 
 import java.util.Date
 
-import scala.collection.mutable
-
 import org.apache.spark.deploy.Command
 
 /**
@@ -34,13 +32,13 @@ import org.apache.spark.deploy.Command
  * @param command The command to launch the driver.
  * @param schedulerProperties Extra properties to pass the Mesos scheduler
  */
-private[spark] class MesosDriverDescription(
+private[spark] case class MesosDriverDescription(
     val jarUrl: String,
     val mem: Int,
-    val cores: Int,
+    val cores: Double,
     val supervise: Boolean,
     val command: Command,
-    val schedulerProperties: mutable.HashMap[String, String])
+    val schedulerProperties: Map[String, String])
   extends Serializable {
 
   var submissionId: Option[String] = None

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

@@ -30,7 +30,7 @@ import org.apache.spark.ui.{UIUtils, WebUIPage}
 private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage("") {
   def render(request: HttpServletRequest): Seq[Node] = {
     val state = parent.scheduler.getState()
-    val queuedHeaders = Seq("DriverID", "Submit Date", "Description")
+    val queuedHeaders = Seq("DriverID", "Submit Date", "Main Class", "Driver resources")
     val driverHeaders = queuedHeaders ++
       Seq("Start Date", "Mesos Slave ID", "State")
     val retryHeaders = Seq("DriverID", "Submit Date", "Description") ++
@@ -61,6 +61,7 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
       <td>{submission.submissionId}</td>
       <td>{submission.submissionDate}</td>
       <td>{submission.command.mainClass}</td>
+      <td>cpus: {submission.cores}, mem: {submission.mem}</td>
     </tr>
   }
 
@@ -91,21 +92,22 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
       return ""
     }
     val sb = new StringBuilder
-    sb.append(s"State: ${status.get.getState}")
+    val s = status.get
+    sb.append(s"State: ${s.getState}")
     if (status.get.hasMessage) {
-      sb.append(s", Message: ${status.get.getMessage}")
+      sb.append(s", Message: ${s.getMessage}")
     }
     if (status.get.hasHealthy) {
-      sb.append(s", Healthy: ${status.get.getHealthy}")
+      sb.append(s", Healthy: ${s.getHealthy}")
     }
     if (status.get.hasSource) {
-      sb.append(s", Source: ${status.get.getSource}")
+      sb.append(s", Source: ${s.getSource}")
     }
     if (status.get.hasReason) {
-      sb.append(s", Reason: ${status.get.getReason}")
+      sb.append(s", Reason: ${s.getReason}")
     }
     if (status.get.hasTimestamp) {
-      sb.append(s", Time: ${status.get.getTimestamp}")
+      sb.append(s", Time: ${s.getTimestamp}")
     }
     sb.toString()
   }

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

@@ -51,14 +51,14 @@ private[spark] abstract class RestServer extends Logging {
   val host: String
   val requestedPort: Int
   val masterConf: SparkConf
-  val submitRequestServlet: SubmitRequestServlet
-  val killRequestServlet: KillRequestServlet
-  val statusRequestServlet: StatusRequestServlet
+  def submitRequestServlet: SubmitRequestServlet
+  def killRequestServlet: KillRequestServlet
+  def statusRequestServlet: StatusRequestServlet
 
   private var _server: Option[Server] = None
 
   // A mapping from URL prefixes to servlets that serve them. Exposed for testing.
-  protected val baseContext = s"/$RestServer.PROTOCOL_VERSION/submissions"
+  protected val baseContext = s"/${RestServer.PROTOCOL_VERSION}/submissions"
   protected val contextToServlet = Map[String, RestServlet](
     s"$baseContext/create/*" -> submitRequestServlet,
     s"$baseContext/kill/*" -> killRequestServlet,

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

@@ -45,21 +45,23 @@ import org.apache.spark.deploy.ClientArguments._
  *
  * @param host the address this server should bind to
  * @param requestedPort the port this server will attempt to bind to
+ * @param masterConf the conf used by the Master
  * @param masterActor reference to the Master actor to which requests can be sent
  * @param masterUrl the URL of the Master new drivers will attempt to connect to
- * @param masterConf the conf used by the Master
  */
 private[deploy] class StandaloneRestServer(
     val host: String,
     val requestedPort: Int,
+    val masterConf: SparkConf,
     masterActor: ActorRef,
-    masterUrl: String,
-    val masterConf: SparkConf)
+    masterUrl: String)
   extends RestServer {
-
-  val submitRequestServlet = new StandaloneSubmitRequestServlet(masterActor, masterUrl, masterConf)
-  val killRequestServlet = new StandaloneKillRequestServlet(masterActor, masterConf)
-  val statusRequestServlet = new StandaloneStatusRequestServlet(masterActor, masterConf)
+  def submitRequestServlet: SubmitRequestServlet =
+    new StandaloneSubmitRequestServlet(masterActor, masterUrl, masterConf)
+  def killRequestServlet: KillRequestServlet =
+    new StandaloneKillRequestServlet(masterActor, masterConf)
+  def statusRequestServlet: StatusRequestServlet =
+    new StandaloneStatusRequestServlet(masterActor, masterConf)
 }
 
 /**

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

@@ -20,15 +20,15 @@ package org.apache.spark.deploy.rest.mesos
 import java.io.File
 import javax.servlet.http.HttpServletResponse
 
-import scala.collection.mutable
-
 import org.apache.spark.deploy.Command
 import org.apache.spark.deploy.mesos.MesosDriverDescription
 import org.apache.spark.deploy.rest._
 import org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler
 import org.apache.spark.util.Utils
 import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf}
 
+import scala.collection.mutable
+
 
 /**
  * A server that responds to requests submitted by the [[RestClient]].
@@ -44,9 +44,12 @@ private[spark] class MesosRestServer(
     val masterConf: SparkConf,
     scheduler: MesosClusterScheduler)
   extends RestServer {
-  val submitRequestServlet = new MesosSubmitRequestServlet(scheduler, masterConf)
-  val killRequestServlet = new MesosKillRequestServlet(scheduler, masterConf)
-  val statusRequestServlet = new MesosStatusRequestServlet(scheduler, masterConf)
+  def submitRequestServlet: SubmitRequestServlet =
+    new MesosSubmitRequestServlet(scheduler, masterConf)
+  def killRequestServlet: KillRequestServlet =
+    new MesosKillRequestServlet(scheduler, masterConf)
+  def statusRequestServlet: StatusRequestServlet =
+    new MesosStatusRequestServlet(scheduler, masterConf)
 }
 
 private[mesos] class MesosSubmitRequestServlet(
@@ -56,7 +59,7 @@ private[mesos] class MesosSubmitRequestServlet(
 
   private val DEFAULT_SUPERVISE = false
   private val DEFAULT_MEMORY = 512 // mb
-  private val DEFAULT_CORES = 1
+  private val DEFAULT_CORES = 1.0
 
   /**
    * Build a driver description from the fields specified in the submit request.
@@ -113,11 +116,11 @@ private[mesos] class MesosSubmitRequestServlet(
       mainClass, appArgs, environmentVariables, extraClassPath, extraLibraryPath, javaOpts)
     val actualSuperviseDriver = superviseDriver.map(_.toBoolean).getOrElse(DEFAULT_SUPERVISE)
     val actualDriverMemory = driverMemory.map(Utils.memoryStringToMb).getOrElse(DEFAULT_MEMORY)
-    val actualDriverCores = driverCores.map(_.toInt).getOrElse(DEFAULT_CORES)
+    val actualDriverCores = driverCores.map(_.toDouble).getOrElse(DEFAULT_CORES)
 
     new MesosDriverDescription(
       appResource, actualDriverMemory, actualDriverCores,
-      actualSuperviseDriver, command, schedulerProperties)
+      actualSuperviseDriver, command, schedulerProperties.toMap)
   }
 
   protected override def handleSubmit(

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

@@ -26,6 +26,7 @@ import org.apache.spark.deploy.mesos.MesosDriverDescription
  * This queue automatically stores the state after each pop/push
  * so it can be recovered later.
  * This queue is also bounded and rejects offers when it's full.
+ * This class is not thread-safe, and we expect the caller to handle synchronizing state.
  * @param state Mesos state abstraction to fetch persistent state.
  */
 private[mesos] class DriverQueue(state: MesosClusterPersistenceEngine, capacity: Int) {
@@ -42,7 +43,7 @@ private[mesos] class DriverQueue(state: MesosClusterPersistenceEngine, capacity:
     count = queue.size
   }
 
-  def isFull = count >= capacity
+  def isFull: Boolean = count >= capacity
 
   def size: Int = count
 

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

@@ -22,6 +22,7 @@ import org.apache.mesos.Protos.SlaveID
 
 /**
  * Tracks all the launched or running drivers in the Mesos cluster scheduler.
+ * This class is not thread-safe, and we expect the caller to handle synchronizing state.
  * @param state Persistence engine to store tasks state.
  */
 private[mesos] class LaunchedDrivers(state: MesosClusterPersistenceEngine) {
@@ -51,19 +52,14 @@ private[mesos] class LaunchedDrivers(state: MesosClusterPersistenceEngine) {
   def contains(submissionId: String): Boolean = drivers.contains(submissionId)
 
   def remove(submissionId: String): Option[MesosClusterTaskState] = {
-    if (pendingRecover.contains(submissionId)) {
-      pendingRecover.remove(submissionId)
-    }
-
+    pendingRecover.remove(submissionId)
     val removedState = drivers.remove(submissionId)
     state.expunge(submissionId)
     removedState
   }
 
   def set(submissionId: String, newState: MesosClusterTaskState): Unit = {
-    if (pendingRecover.contains(newState.taskId.getValue)) {
-      pendingRecover.remove(newState.taskId.getValue)
-    }
+    pendingRecover.remove(newState.taskId.getValue)
     drivers(submissionId) = newState
     state.persist(submissionId, newState)
   }

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

@@ -329,20 +329,34 @@ private[spark] class MesosClusterSchedulerDriver(
     options
   }
 
-  private class ResourceOffer(val offer: Offer, var cpu: Double, var mem: Double)
+  private class ResourceOffer(val offer: Offer, var cpu: Double, var mem: Double) {
+    override def toString(): String = {
+      s"Offer id: ${offer.getId.getValue}, cpu: $cpu, mem: $mem"
+    }
+  }
 
   override def resourceOffers(driver: SchedulerDriver, offers: JList[Offer]): Unit = {
+    def printOffers(offers: Iterable[ResourceOffer]): String = {
+      val builder = new StringBuilder()
+      offers.foreach { o =>
+        builder.append(o).append("\n")
+      }
+      builder.toString()
+    }
+
     val currentOffers = offers.map { o =>
       new ResourceOffer(
         o,
         getResource(o.getResourcesList, "cpus"),
         getResource(o.getResourcesList, "mem"))
     }
+    logTrace(s"Received offers from Mesos: ${printOffers(currentOffers)}")
     val tasks = new mutable.HashMap[OfferID, ArrayBuffer[TaskInfo]]()
     val currentTime = new Date()
+
     def scheduleTasks(
         taskFunc: () => (Option[MesosDriverDescription], Option[RetryState]),
-        scheduledCallback: (String) => Unit) {
+        scheduledCallback: (String) => Unit): Unit = {
       var nextItem = taskFunc()
       // TODO: We should not stop scheduling at the very first task
       // that cannot be scheduled. Instead we should exhaust the
@@ -352,11 +366,14 @@ private[spark] class MesosClusterSchedulerDriver(
         val (submission, retryState) = (nextItem._1.get, nextItem._2)
         val driverCpu = submission.cores
         val driverMem = submission.mem
+        logTrace(s"Finding offer to launch driver with cpu: $driverCpu, mem: $driverMem")
         val offerOption = currentOffers.find { o =>
           o.cpu >= driverCpu && o.mem >= driverMem
         }
 
         if (offerOption.isEmpty) {
+          logDebug(s"Unable to find offer to launch driver id: ${submission.submissionId.get}," +
+              s"cpu: $driverCpu, mem: $driverMem")
           return
         }
 
@@ -371,9 +388,10 @@ private[spark] class MesosClusterSchedulerDriver(
           .setName("mem").setType(Value.Type.SCALAR)
           .setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()
         val commandInfo = buildCommand(submission)
+        val appName = submission.schedulerProperties("spark.app.name")
         val taskInfo = TaskInfo.newBuilder()
           .setTaskId(taskId)
-          .setName(s"driver for ${submission.command.mainClass}")
+          .setName(s"Driver for $appName")
           .setSlaveId(offer.offer.getSlaveId)
           .setCommand(commandInfo)
           .addResources(cpuResource)
@@ -387,6 +405,9 @@ private[spark] class MesosClusterSchedulerDriver(
           tasks(offer.offer.getId)
         }
         queuedTasks += taskInfo
+        logTrace(s"Using offer ${offer.offer.getId.getValue} to launch driver " +
+          submission.submissionId.get)
+
         launchedDrivers.set(
           submission.submissionId.get,
           new MesosClusterTaskState(submission, taskId, offer.offer.getSlaveId,