Add end-to-end tests for standalone REST protocol

This is actually non-trivial because we must run standalone mode
instead of relying on the existing local-cluster mode. This means
we must manually start our own Master and Workers, and provide a
real jar when submitting the test application, which involves
manually packaging our own jar. Further, since the driver output
is difficult to obtain programmatically in cluster mode, we need
to write the results to a special file and verify them later.

Author: Andrew Or

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

@@ -45,7 +45,7 @@ class LocalSparkCluster(numWorkers: Int, coresPerWorker: Int, memoryPerWorker: I
 
     /* Start the Master */
     val conf = new SparkConf(false)
-    val (masterSystem, masterPort, _) = Master.startSystemAndActor(localHostname, 0, 0, conf)
+    val (masterSystem, masterPort, _, _) = Master.startSystemAndActor(localHostname, 0, 0, conf)
     masterActorSystems += masterSystem
     val masterUrl = "spark://" + localHostname + ":" + masterPort
     val masters = Array(masterUrl)

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

@@ -117,19 +117,10 @@ object SparkSubmit {
    * parameters directly to a REST client, which will submit the application using the
    * REST protocol instead.
    */
-  private def submit(args: SparkSubmitArguments): Unit = {
+  private[spark] def submit(args: SparkSubmitArguments): Unit = {
     val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)
-    val restKey = "spark.submit.rest.enabled"
-    val restEnabled = args.sparkProperties.get(restKey).getOrElse("false").toBoolean
-    if (args.isStandaloneCluster && restEnabled) {
+    if (args.isStandaloneCluster && args.isRestEnabled) {
       printStream.println("Running standalone cluster mode using the stable REST protocol.")
-      // NOTE: since we mutate the values of some configs in `prepareSubmitEnvironment`, we
-      // must update the corresponding fields in the original SparkSubmitArguments to reflect
-      // these changes.
-      args.sparkProperties.clear()
-      args.sparkProperties ++= sysProps
-      sysProps.get("spark.jars").foreach { args.jars = _ }
-      sysProps.get("spark.files").foreach { args.files = _ }
       new StandaloneRestClient().submitDriver(args)
     } else {
       runMain(childArgs, childClasspath, sysProps, childMainClass)
@@ -159,7 +150,7 @@ object SparkSubmit {
       case m if m.startsWith("spark") => STANDALONE
       case m if m.startsWith("mesos") => MESOS
       case m if m.startsWith("local") => LOCAL
-      case _ => printErrorAndExit("Master must start with yarn, spark, mesos or local"); -1
+      case _ => printErrorAndExit("Master must start with yarn, spark, mesos, or local"); -1
     }
 
     // Set the deploy mode; default is client mode
@@ -249,6 +240,8 @@ object SparkSubmit {
 
       // Standalone cluster only
       OptionAssigner(args.jars, STANDALONE, CLUSTER, sysProp = "spark.jars"),
+      OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, clOption = "--memory"),
+      OptionAssigner(args.driverCores, STANDALONE, CLUSTER, clOption = "--cores"),
 
       // Yarn client only
       OptionAssigner(args.queue, YARN, CLIENT, sysProp = "spark.yarn.queue"),
@@ -299,6 +292,20 @@ object SparkSubmit {
       }
     }
 
+    // In standalone-cluster mode, use Client as a wrapper around the user class
+    // Note that we won't actually launch this class if we're using the stable REST protocol
+    if (args.isStandaloneCluster && !args.isRestEnabled) {
+      childMainClass = "org.apache.spark.deploy.Client"
+      if (args.supervise) {
+        childArgs += "--supervise"
+      }
+      childArgs += "launch"
+      childArgs += (args.master, args.primaryResource, args.mainClass)
+      if (args.childArgs != null) {
+        childArgs ++= args.childArgs
+      }
+    }
+
     // Add the application jar automatically so the user doesn't have to call sc.addJar
     // For YARN cluster mode, the jar is already distributed on each node as "app.jar"
     // For python files, the primary resource is already distributed as a regular file
@@ -356,14 +363,24 @@ object SparkSubmit {
       sysProps("spark.submit.pyFiles") = formattedPyFiles
     }
 
+    // NOTE: If we are using the REST gateway, we will use the original arguments directly.
+    // Since we mutate the values of some configs in this method, we must update the
+    // corresponding fields in the original SparkSubmitArguments to reflect these changes.
+    if (args.isStandaloneCluster && args.isRestEnabled) {
+      args.sparkProperties.clear()
+      args.sparkProperties ++= sysProps
+      sysProps.get("spark.jars").foreach { args.jars = _ }
+      sysProps.get("spark.files").foreach { args.files = _ }
+    }
+
     (childArgs, childClasspath, sysProps, childMainClass)
   }
 
   /**
    * Run the main method of the child class using the provided launch environment.
    *
-   * Depending on the deploy mode, cluster manager, and the type of the application,
-   * this main class may not necessarily be the one provided by the user.
+   * Note that this main class will not be the one provided by the user if we're
+   * running cluster deploy mode or python applications.
    */
   private def runMain(
       childArgs: Seq[String],

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

@@ -256,6 +256,11 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
     master.startsWith("spark://") && deployMode == "cluster"
   }
 
+  /** Return whether the stable application submission REST gateway is enabled. */
+  def isRestEnabled: Boolean = {
+    sparkProperties.get("spark.submit.rest.enabled").getOrElse("false").toBoolean
+  }
+
   override def toString = {
     s"""Parsed arguments:
     |  master                  $master

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

@@ -53,12 +53,12 @@ private[spark] class Master(
     host: String,
     port: Int,
     webUiPort: Int,
-    val securityMgr: SecurityManager)
+    val securityMgr: SecurityManager,
+    val conf: SparkConf)
   extends Actor with ActorLogReceive with Logging with LeaderElectable {
 
   import context.dispatcher   // to use Akka's scheduler.schedule()
 
-  val conf = new SparkConf
   val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
 
   def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")  // For application IDs
@@ -444,8 +444,8 @@ private[spark] class Master(
       timeOutDeadWorkers()
     }
 
-    case RequestWebUIPort => {
-      sender ! WebUIPortResponse(webUi.boundPort)
+    case BoundPortsRequest => {
+      sender ! BoundPortsResponse(port, webUi.boundPort, restServerBoundPort)
     }
   }
 
@@ -866,7 +866,7 @@ private[spark] object Master extends Logging {
     SignalLogger.register(log)
     val conf = new SparkConf
     val args = new MasterArguments(argStrings, conf)
-    val (actorSystem, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)
+    val (actorSystem, _, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)
     actorSystem.awaitTermination()
   }
 
@@ -890,19 +890,26 @@ private[spark] object Master extends Logging {
     Address("akka.tcp", systemName, host, port)
   }
 
+  /**
+   * Start the Master and return a four tuple of:
+   *   (1) The Master actor system
+   *   (2) The bound port
+   *   (3) The web UI bound port
+   *   (4) The REST server bound port, if any
+   */
   def startSystemAndActor(
       host: String,
       port: Int,
       webUiPort: Int,
-      conf: SparkConf): (ActorSystem, Int, Int) = {
+      conf: SparkConf): (ActorSystem, Int, Int, Option[Int]) = {
     val securityMgr = new SecurityManager(conf)
     val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,
       securityManager = securityMgr)
-    val actor = actorSystem.actorOf(Props(classOf[Master], host, boundPort, webUiPort,
-      securityMgr), actorName)
+    val actor = actorSystem.actorOf(
+      Props(classOf[Master], host, boundPort, webUiPort, securityMgr, conf), actorName)
     val timeout = AkkaUtils.askTimeout(conf)
-    val respFuture = actor.ask(RequestWebUIPort)(timeout)
-    val resp = Await.result(respFuture, timeout).asInstanceOf[WebUIPortResponse]
-    (actorSystem, boundPort, resp.webUIBoundPort)
+    val portsRequest = actor.ask(BoundPortsRequest)(timeout)
+    val portsResponse = Await.result(portsRequest, timeout).asInstanceOf[BoundPortsResponse]
+    (actorSystem, boundPort, portsResponse.webUIPort, portsResponse.stablePort)
   }
 }

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

@@ -36,7 +36,7 @@ private[master] object MasterMessages {
 
   case object CompleteRecovery
 
-  case object RequestWebUIPort
+  case object BoundPortsRequest
 
-  case class WebUIPortResponse(webUIBoundPort: Int)
+  case class BoundPortsResponse(actorPort: Int, webUIPort: Int, stablePort: Option[Int])
 }

core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala

@@ -68,7 +68,8 @@ class JsonProtocolSuite extends FunSuite {
     val completedApps = Array[ApplicationInfo]()
     val activeDrivers = Array(createDriverInfo())
     val completedDrivers = Array(createDriverInfo())
-    val stateResponse = new MasterStateResponse("host", 8080, workers, activeApps, completedApps,
+    val stateResponse = new MasterStateResponse(
+      "host", 8080, None, workers, activeApps, completedApps,
       activeDrivers, completedDrivers, RecoveryState.ALIVE)
     val output = JsonProtocol.writeMasterState(stateResponse)
     assertValidJson(output)

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -197,6 +197,33 @@ class SparkSubmitSuite extends FunSuite with Matchers with ResetSystemProperties
     sysProps("spark.shuffle.spill") should be ("false")
   }
 
+  test("handles standalone cluster mode") {
+    val clArgs = Seq(
+      "--deploy-mode", "cluster",
+      "--master", "spark://h:p",
+      "--class", "org.SomeClass",
+      "--supervise",
+      "--driver-memory", "4g",
+      "--driver-cores", "5",
+      "--conf", "spark.shuffle.spill=false",
+      "thejar.jar",
+      "arg1", "arg2")
+    val appArgs = new SparkSubmitArguments(clArgs)
+    val (childArgs, classpath, sysProps, mainClass) = prepareSubmitEnvironment(appArgs)
+    val childArgsStr = childArgs.mkString(" ")
+    childArgsStr should startWith ("--memory 4g --cores 5 --supervise")
+    childArgsStr should include regex ("launch spark://h:p .*thejar.jar org.SomeClass arg1 arg2")
+    mainClass should be ("org.apache.spark.deploy.Client")
+    classpath should have size (0)
+    sysProps should have size (5)
+    sysProps.keys should contain ("SPARK_SUBMIT")
+    sysProps.keys should contain ("spark.master")
+    sysProps.keys should contain ("spark.app.name")
+    sysProps.keys should contain ("spark.jars")
+    sysProps.keys should contain ("spark.shuffle.spill")
+    sysProps("spark.shuffle.spill") should be ("false")
+  }
+
   test("handles standalone client mode") {
     val clArgs = Seq(
       "--deploy-mode", "client",
@@ -279,6 +306,7 @@ class SparkSubmitSuite extends FunSuite with Matchers with ResetSystemProperties
       "--master", "local-cluster[2,1,512]",
       "--jars", jarsString,
       "--conf", "spark.ui.enabled=false",
+      "--conf", "spark.master.rest.enabled=false",
       unusedJar.toString)
     runSparkSubmit(args)
   }