submit app to HA cluster in standalone cluster mode

WangTaoTheTonic · WangTaoTheTonic · commit fa1fa806460d · 2015-03-25T11:46:53.000+08:00
diff --git a/core/src/main/scala/org/apache/spark/deploy/Client.scala b/core/src/main/scala/org/apache/spark/deploy/Client.scala
@@ -35,16 +35,15 @@ import org.apache.spark.util.{ActorLogReceive, AkkaUtils, Utils}
 private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
   extends Actor with ActorLogReceive with Logging {
 
-  var masterActor: ActorSelection = _
+  var mastersActor = driverArgs.masters.map { m =>
+    context.actorSelection(Master.toAkkaUrl(m, AkkaUtils.protocol(context.system)))
+  }
   val timeout = AkkaUtils.askTimeout(conf)
 
   override def preStart() = {
-    masterActor = context.actorSelection(
-      Master.toAkkaUrl(driverArgs.master, AkkaUtils.protocol(context.system)))
-
     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
 
-    println(s"Sending ${driverArgs.cmd} command to ${driverArgs.master}")
+    println(s"Sending ${driverArgs.cmd} command to ${driverArgs.masters}")
 
     driverArgs.cmd match {
       case "launch" =>
@@ -79,10 +78,12 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
           driverArgs.supervise,
           command)
 
+      for (masterActor <- mastersActor)
         masterActor ! RequestSubmitDriver(driverDescription)
 
       case "kill" =>
         val driverId = driverArgs.driverId
+      for (masterActor <- mastersActor)
         masterActor ! RequestKillDriver(driverId)
     }
   }
@@ -92,48 +93,61 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
     println(s"... waiting before polling master for driver state")
     Thread.sleep(5000)
     println("... polling master for driver state")
-    val statusFuture = (masterActor ? RequestDriverStatus(driverId))(timeout)
-      .mapTo[DriverStatusResponse]
-    val statusResponse = Await.result(statusFuture, timeout)
-
-    statusResponse.found match {
-      case false =>
-        println(s"ERROR: Cluster master did not recognize $driverId")
-        System.exit(-1)
-      case true =>
-        println(s"State of $driverId is ${statusResponse.state.get}")
-        // Worker node, if present
-        (statusResponse.workerId, statusResponse.workerHostPort, statusResponse.state) match {
-          case (Some(id), Some(hostPort), Some(DriverState.RUNNING)) =>
-            println(s"Driver running on $hostPort ($id)")
-          case _ =>
-        }
-        // Exception, if present
-        statusResponse.exception.map { e =>
-          println(s"Exception from cluster was: $e")
-          e.printStackTrace()
-          System.exit(-1)
-        }
-        System.exit(0)
+    for (masterActor <- mastersActor) {
+      val statusFuture = (masterActor ? RequestDriverStatus(driverId))(timeout)
+        .mapTo[DriverStatusResponse]
+      val statusResponse = Await.result(statusFuture, timeout)
+
+      statusResponse.found match {
+        case false =>
+          statusResponse.exception.getOrElse {
+            println(s"ERROR: Cluster master did not recognize $driverId")
+            System.exit(-1)
+          }
+        case true =>
+          println(s"State of $driverId is ${statusResponse.state.get}")
+          // Worker node, if present
+          (statusResponse.workerId, statusResponse.workerHostPort, statusResponse.state) match {
+            case (Some(id), Some(hostPort), Some(DriverState.RUNNING)) =>
+              println(s"Driver running on $hostPort ($id)")
+            case _ =>
+          }
+          // Exception, if present
+          statusResponse.exception.map { e =>
+            println(s"Exception from cluster was: $e")
+            e.printStackTrace()
+            System.exit(-1)
+          }
+          System.exit(0)
+      }
     }
   }
 
   override def receiveWithLogging = {
 
     case SubmitDriverResponse(success, driverId, message) =>
       println(message)
-      if (success) pollAndReportStatus(driverId.get) else System.exit(-1)
+      if (success) {
+        pollAndReportStatus(driverId.get)
+      } else if (!message.contains("Can only")) {
+        System.exit(-1)
+      }
+
 
     case KillDriverResponse(driverId, success, message) =>
       println(message)
-      if (success) pollAndReportStatus(driverId) else System.exit(-1)
+      if (success) {
+        pollAndReportStatus(driverId)
+      } else if (!message.contains("Can only")) {
+        System.exit(-1)
+      }
 
     case DisassociatedEvent(_, remoteAddress, _) =>
-      println(s"Error connecting to master ${driverArgs.master} ($remoteAddress), exiting.")
+      println(s"Error connecting to master ${driverArgs.masters} ($remoteAddress), exiting.")
       System.exit(-1)
 
     case AssociationErrorEvent(cause, _, remoteAddress, _, _) =>
-      println(s"Error connecting to master ${driverArgs.master} ($remoteAddress), exiting.")
+      println(s"Error connecting to master ${driverArgs.masters} ($remoteAddress), exiting.")
       println(s"Cause was: $cause")
       System.exit(-1)
   }
@@ -163,7 +177,9 @@ object Client {
       "driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))
 
     // Verify driverArgs.master is a valid url so that we can use it in ClientActor safely
-    Master.toAkkaUrl(driverArgs.master, AkkaUtils.protocol(actorSystem))
+    for (m <- driverArgs.masters) {
+      Master.toAkkaUrl(m, AkkaUtils.protocol(actorSystem))
+    }
     actorSystem.actorOf(Props(classOf[ClientActor], driverArgs, conf))
 
     actorSystem.awaitTermination()
diff --git a/core/src/main/scala/org/apache/spark/deploy/ClientArguments.scala b/core/src/main/scala/org/apache/spark/deploy/ClientArguments.scala
@@ -35,7 +35,7 @@ private[deploy] class ClientArguments(args: Array[String]) {
   var logLevel = Level.WARN
 
   // launch parameters
-  var master: String = ""
+  var masters: Array[String] = null
   var jarUrl: String = ""
   var mainClass: String = ""
   var supervise: Boolean = DEFAULT_SUPERVISE
@@ -80,13 +80,13 @@ private[deploy] class ClientArguments(args: Array[String]) {
       }
 
       jarUrl = _jarUrl
-      master = _master
+      masters = _master.stripPrefix("spark://").split(",").map("spark://" + _)
       mainClass = _mainClass
       _driverOptions ++= tail
 
     case "kill" :: _master :: _driverId :: tail =>
       cmd = "kill"
-      master = _master
+      masters = _master.stripPrefix("spark://").split(",").map("spark://" + _)
       driverId = _driverId
 
     case _ =>
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -115,17 +115,17 @@ object SparkSubmit {
 
   /** Kill an existing submission using the REST protocol. Standalone cluster mode only. */
   private def kill(args: SparkSubmitArguments): Unit = {
-    new StandaloneRestClient()
-      .killSubmission(args.master, args.submissionToKill)
+    new StandaloneRestClient(args.master)
+      .killSubmission(args.submissionToKill)
   }
 
   /**
    * Request the status of an existing submission using the REST protocol.
    * Standalone cluster mode only.
    */
   private def requestStatus(args: SparkSubmitArguments): Unit = {
-    new StandaloneRestClient()
-      .requestSubmissionStatus(args.master, args.submissionToRequestStatusFor)
+    new StandaloneRestClient(args.master)
+      .requestSubmissionStatus(args.submissionToRequestStatusFor)
   }
 
   /**
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -305,12 +305,17 @@ private[master] class Master(
     }
 
     case RequestDriverStatus(driverId) => {
-      (drivers ++ completedDrivers).find(_.id == driverId) match {
-        case Some(driver) =>
-          sender ! DriverStatusResponse(found = true, Some(driver.state),
-            driver.worker.map(_.id), driver.worker.map(_.hostPort), driver.exception)
-        case None =>
-          sender ! DriverStatusResponse(found = false, None, None, None, None)
+      if (state != RecoveryState.ALIVE) {
+        val msg = s"Can only request driver status in ALIVE state. Current state: $state."
+        sender ! DriverStatusResponse(found = false, None, None, None, Some(new Exception(msg)))
+      } else {
+        (drivers ++ completedDrivers).find(_.id == driverId) match {
+          case Some(driver) =>
+            sender ! DriverStatusResponse(found = true, Some(driver.state),
+              driver.worker.map(_.id), driver.worker.map(_.hostPort), driver.exception)
+          case None =>
+            sender ! DriverStatusResponse(found = false, None, None, None, None)
+        }
       }
     }
 
diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestClient.scala b/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestClient.scala
@@ -52,55 +52,82 @@ import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
  * implementation of this client can use that information to retry using the version specified
  * by the server.
  */
-private[deploy] class StandaloneRestClient extends Logging {
+private[deploy] class StandaloneRestClient(master: String) extends Logging {
   import StandaloneRestClient._
 
+  val masters: Array[String] = master.stripPrefix("spark://").split(",").map("spark://" + _)
+
   /**
    * Submit an application specified by the parameters in the provided request.
    *
    * If the submission was successful, poll the status of the submission and report
    * it to the user. Otherwise, report the error message provided by the server.
    */
   private[rest] def createSubmission(
-      master: String,
       request: CreateSubmissionRequest): SubmitRestProtocolResponse = {
     logInfo(s"Submitting a request to launch an application in $master.")
-    validateMaster(master)
-    val url = getSubmitUrl(master)
-    val response = postJson(url, request.toJson)
-    response match {
-      case s: CreateSubmissionResponse =>
-        reportSubmissionStatus(master, s)
-        handleRestResponse(s)
-      case unexpected =>
-        handleUnexpectedRestResponse(unexpected)
+    var suc: Boolean = false
+    var response: SubmitRestProtocolResponse = null
+    for (m <- masters if !suc) {
+      validateMaster(m)
+      val url = getSubmitUrl(m)
+      response = postJson(url, request.toJson)
+      response match {
+        case s: CreateSubmissionResponse =>
+          if (s.success) {
+            reportSubmissionStatus(s)
+            handleRestResponse(s)
+            suc = true
+          }
+        case unexpected =>
+          handleUnexpectedRestResponse(unexpected)
+      }
     }
     response
   }
 
   /** Request that the server kill the specified submission. */
-  def killSubmission(master: String, submissionId: String): SubmitRestProtocolResponse = {
+  def killSubmission(submissionId: String): SubmitRestProtocolResponse = {
     logInfo(s"Submitting a request to kill submission $submissionId in $master.")
-    validateMaster(master)
-    val response = post(getKillUrl(master, submissionId))
-    response match {
-      case k: KillSubmissionResponse => handleRestResponse(k)
-      case unexpected => handleUnexpectedRestResponse(unexpected)
+    var suc: Boolean = false
+    var response: SubmitRestProtocolResponse = null
+    for (m <- masters if !suc) {
+      validateMaster(m)
+      response = post(getKillUrl(m, submissionId))
+      response match {
+        case k: KillSubmissionResponse =>
+          if (!k.message.contains("Can only")) {
+            handleRestResponse(k)
+            suc = true
+          }
+        case unexpected =>
+          handleUnexpectedRestResponse(unexpected)
+      }
     }
     response
   }
 
   /** Request the status of a submission from the server. */
   def requestSubmissionStatus(
-      master: String,
       submissionId: String,
       quiet: Boolean = false): SubmitRestProtocolResponse = {
     logInfo(s"Submitting a request for the status of submission $submissionId in $master.")
-    validateMaster(master)
-    val response = get(getStatusUrl(master, submissionId))
-    response match {
-      case s: SubmissionStatusResponse => if (!quiet) { handleRestResponse(s) }
-      case unexpected => handleUnexpectedRestResponse(unexpected)
+    var suc: Boolean = false
+    var response: SubmitRestProtocolResponse = null
+    for (m <- masters) {
+      validateMaster(m)
+      response = get(getStatusUrl(m, submissionId))
+      response match {
+        case s: SubmissionStatusResponse =>
+          if (!s.message.contains("Can only")) {
+            if (!quiet) {
+              handleRestResponse(s)
+            }
+            suc = true
+          }
+        case unexpected =>
+          handleUnexpectedRestResponse(unexpected)
+      }
     }
     response
   }
@@ -228,30 +255,24 @@ private[deploy] class StandaloneRestClient extends Logging {
 
   /** Report the status of a newly created submission. */
   private def reportSubmissionStatus(
-      master: String,
       submitResponse: CreateSubmissionResponse): Unit = {
-    if (submitResponse.success) {
-      val submissionId = submitResponse.submissionId
-      if (submissionId != null) {
-        logInfo(s"Submission successfully created as $submissionId. Polling submission state...")
-        pollSubmissionStatus(master, submissionId)
-      } else {
-        // should never happen
-        logError("Application successfully submitted, but submission ID was not provided!")
-      }
+    val submissionId = submitResponse.submissionId
+    if (submissionId != null) {
+      logInfo(s"Submission successfully created as $submissionId. Polling submission state...")
+      pollSubmissionStatus(submissionId)
     } else {
-      val failMessage = Option(submitResponse.message).map { ": " + _ }.getOrElse("")
-      logError("Application submission failed" + failMessage)
+      // should never happen
+      logError("Application successfully submitted, but submission ID was not provided!")
     }
   }
 
   /**
    * Poll the status of the specified submission and log it.
    * This retries up to a fixed number of times before giving up.
    */
-  private def pollSubmissionStatus(master: String, submissionId: String): Unit = {
+  private def pollSubmissionStatus(submissionId: String): Unit = {
     (1 to REPORT_DRIVER_STATUS_MAX_TRIES).foreach { _ =>
-      val response = requestSubmissionStatus(master, submissionId, quiet = true)
+      val response = requestSubmissionStatus(submissionId, quiet = true)
       val statusResponse = response match {
         case s: SubmissionStatusResponse => s
         case _ => return // unexpected type, let upstream caller handle it
@@ -311,10 +332,10 @@ private[rest] object StandaloneRestClient {
     }
     val sparkProperties = conf.getAll.toMap
     val environmentVariables = env.filter { case (k, _) => k.startsWith("SPARK_") }
-    val client = new StandaloneRestClient
+    val client = new StandaloneRestClient(master)
     val submitRequest = client.constructSubmitRequest(
       appResource, mainClass, appArgs, sparkProperties, environmentVariables)
-    client.createSubmission(master, submitRequest)
+    client.createSubmission(submitRequest)
   }
 
   def main(args: Array[String]): Unit = {
diff --git a/core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala b/core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala
