You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by an...@apache.org on 2015/04/28 22:34:10 UTC
[3/3] spark git commit: [SPARK-5338] [MESOS] Add cluster mode support
for Mesos
[SPARK-5338] [MESOS] Add cluster mode support for Mesos
This patch adds the support for cluster mode to run on Mesos.
It introduces a new Mesos framework dedicated to launch new apps/drivers, and can be called with the spark-submit script and specifying --master flag to the cluster mode REST interface instead of Mesos master.
Example:
./bin/spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master mesos://10.0.0.206:8077 --executor-memory 1G --total-executor-cores 100 examples/target/spark-examples_2.10-1.3.0-SNAPSHOT.jar 30
Part of this patch is also to abstract the StandaloneRestServer so it can have different implementations of the REST endpoints.
Features of the cluster mode in this PR:
- Supports supervise mode where scheduler will keep trying to reschedule exited job.
- Adds a new UI for the cluster mode scheduler to see all the running jobs, finished jobs, and supervise jobs waiting to be retried
- Supports state persistence to ZK, so when the cluster scheduler fails over it can pick up all the queued and running jobs
Author: Timothy Chen <tn...@gmail.com>
Author: Luc Bourlier <lu...@typesafe.com>
Closes #5144 from tnachen/mesos_cluster_mode and squashes the following commits:
069e946 [Timothy Chen] Fix rebase.
e24b512 [Timothy Chen] Persist submitted driver.
390c491 [Timothy Chen] Fix zk conf key for mesos zk engine.
e324ac1 [Timothy Chen] Fix merge.
fd5259d [Timothy Chen] Address review comments.
1553230 [Timothy Chen] Address review comments.
c6c6b73 [Timothy Chen] Pass spark properties to mesos cluster tasks.
f7d8046 [Timothy Chen] Change app name to spark cluster.
17f93a2 [Timothy Chen] Fix head of line blocking in scheduling drivers.
6ff8e5c [Timothy Chen] Address comments and add logging.
df355cd [Timothy Chen] Add metrics to mesos cluster scheduler.
20f7284 [Timothy Chen] Address review comments
7252612 [Timothy Chen] Fix tests.
a46ad66 [Timothy Chen] Allow zk cli param override.
920fc4b [Timothy Chen] Fix scala style issues.
862b5b5 [Timothy Chen] Support asking driver status when it's retrying.
7f214c2 [Timothy Chen] Fix RetryState visibility
e0f33f7 [Timothy Chen] Add supervise support and persist retries.
371ce65 [Timothy Chen] Handle cluster mode recovery and state persistence.
3d4dfa1 [Luc Bourlier] Adds support to kill submissions
febfaba [Timothy Chen] Bound the finished drivers in memory
543a98d [Timothy Chen] Schedule multiple jobs
6887e5e [Timothy Chen] Support looking at SPARK_EXECUTOR_URI env variable in schedulers
8ec76bc [Timothy Chen] Fix Mesos dispatcher UI.
d57d77d [Timothy Chen] Add documentation
825afa0 [Luc Bourlier] Supports more spark-submit parameters
b8e7181 [Luc Bourlier] Adds a shutdown latch to keep the deamon running
0fa7780 [Luc Bourlier] Launch task through the mesos scheduler
5b7a12b [Timothy Chen] WIP: Making a cluster mode a mesos framework.
4b2f5ef [Timothy Chen] Specify user jar in command to be replaced with local.
e775001 [Timothy Chen] Support fetching remote uris in driver runner.
7179495 [Timothy Chen] Change Driver page output and add logging
880bc27 [Timothy Chen] Add Mesos Cluster UI to display driver results
9986731 [Timothy Chen] Kill drivers when shutdown
67cbc18 [Timothy Chen] Rename StandaloneRestClient to RestClient and add sbin scripts
e3facdd [Timothy Chen] Add Mesos Cluster dispatcher
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/53befacc
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/53befacc
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/53befacc
Branch: refs/heads/master
Commit: 53befacced828bbac53c6e3a4976ec3f036bae9e
Parents: 8009810
Author: Timothy Chen <tn...@gmail.com>
Authored: Tue Apr 28 13:31:08 2015 -0700
Committer: Andrew Or <an...@databricks.com>
Committed: Tue Apr 28 13:33:57 2015 -0700
----------------------------------------------------------------------
.../spark/deploy/FaultToleranceTest.scala | 2 +-
.../apache/spark/deploy/SparkCuratorUtil.scala | 66 ++
.../org/apache/spark/deploy/SparkSubmit.scala | 48 +-
.../spark/deploy/SparkSubmitArguments.scala | 11 +-
.../org/apache/spark/deploy/master/Master.scala | 2 +-
.../spark/deploy/master/SparkCuratorUtil.scala | 64 --
.../master/ZooKeeperLeaderElectionAgent.scala | 1 +
.../master/ZooKeeperPersistenceEngine.scala | 1 +
.../deploy/mesos/MesosClusterDispatcher.scala | 116 ++++
.../mesos/MesosClusterDispatcherArguments.scala | 101 +++
.../deploy/mesos/MesosDriverDescription.scala | 65 ++
.../deploy/mesos/ui/MesosClusterPage.scala | 114 ++++
.../spark/deploy/mesos/ui/MesosClusterUI.scala | 48 ++
.../deploy/rest/RestSubmissionClient.scala | 344 +++++++++++
.../deploy/rest/RestSubmissionServer.scala | 318 ++++++++++
.../deploy/rest/StandaloneRestClient.scala | 335 ----------
.../deploy/rest/StandaloneRestServer.scala | 344 ++---------
.../deploy/rest/SubmitRestProtocolRequest.scala | 2 +-
.../rest/SubmitRestProtocolResponse.scala | 6 +-
.../deploy/rest/mesos/MesosRestServer.scala | 158 +++++
.../mesos/CoarseMesosSchedulerBackend.scala | 82 +--
.../mesos/MesosClusterPersistenceEngine.scala | 134 ++++
.../cluster/mesos/MesosClusterScheduler.scala | 608 +++++++++++++++++++
.../mesos/MesosClusterSchedulerSource.scala | 40 ++
.../cluster/mesos/MesosSchedulerBackend.scala | 85 +--
.../cluster/mesos/MesosSchedulerUtils.scala | 95 +++
.../apache/spark/deploy/SparkSubmitSuite.scala | 2 +-
.../deploy/rest/StandaloneRestSubmitSuite.scala | 46 +-
.../mesos/MesosClusterSchedulerSuite.scala | 76 +++
docs/running-on-mesos.md | 23 +-
sbin/start-mesos-dispatcher.sh | 40 ++
sbin/stop-mesos-dispatcher.sh | 27 +
32 files changed, 2529 insertions(+), 875 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala b/core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala
index a7c8927..c048b78 100644
--- a/core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala
+++ b/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
/**
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala b/core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala
new file mode 100644
index 0000000..b8d3993
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy
+
+import scala.collection.JavaConversions._
+
+import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
+import org.apache.curator.retry.ExponentialBackoffRetry
+import org.apache.zookeeper.KeeperException
+
+import org.apache.spark.{Logging, SparkConf}
+
+private[spark] object SparkCuratorUtil extends Logging {
+
+ private val ZK_CONNECTION_TIMEOUT_MILLIS = 15000
+ private val ZK_SESSION_TIMEOUT_MILLIS = 60000
+ private val RETRY_WAIT_MILLIS = 5000
+ private val MAX_RECONNECT_ATTEMPTS = 3
+
+ def newClient(
+ conf: SparkConf,
+ zkUrlConf: String = "spark.deploy.zookeeper.url"): CuratorFramework = {
+ val ZK_URL = conf.get(zkUrlConf)
+ val zk = CuratorFrameworkFactory.newClient(ZK_URL,
+ ZK_SESSION_TIMEOUT_MILLIS, ZK_CONNECTION_TIMEOUT_MILLIS,
+ new ExponentialBackoffRetry(RETRY_WAIT_MILLIS, MAX_RECONNECT_ATTEMPTS))
+ zk.start()
+ zk
+ }
+
+ def mkdir(zk: CuratorFramework, path: String) {
+ if (zk.checkExists().forPath(path) == null) {
+ try {
+ zk.create().creatingParentsIfNeeded().forPath(path)
+ } catch {
+ case nodeExist: KeeperException.NodeExistsException =>
+ // do nothing, ignore node existing exception.
+ case e: Exception => throw e
+ }
+ }
+ }
+
+ def deleteRecursive(zk: CuratorFramework, path: String) {
+ if (zk.checkExists().forPath(path) != null) {
+ for (child <- zk.getChildren.forPath(path)) {
+ zk.delete().forPath(path + "/" + child)
+ }
+ zk.delete().forPath(path)
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 296a076..f4f572e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -36,11 +36,11 @@ import org.apache.ivy.core.retrieve.RetrieveOptions
import org.apache.ivy.core.settings.IvySettings
import org.apache.ivy.plugins.matcher.GlobPatternMatcher
import org.apache.ivy.plugins.resolver.{ChainResolver, IBiblioResolver}
-
import org.apache.spark.SPARK_VERSION
import org.apache.spark.deploy.rest._
import org.apache.spark.util.{ChildFirstURLClassLoader, MutableURLClassLoader, Utils}
+
/**
* Whether to submit, kill, or request the status of an application.
* The latter two operations are currently supported only for standalone cluster mode.
@@ -114,18 +114,20 @@ object SparkSubmit {
}
}
- /** Kill an existing submission using the REST protocol. Standalone cluster mode only. */
+ /**
+ * Kill an existing submission using the REST protocol. Standalone and Mesos cluster mode only.
+ */
private def kill(args: SparkSubmitArguments): Unit = {
- new StandaloneRestClient()
+ new RestSubmissionClient()
.killSubmission(args.master, args.submissionToKill)
}
/**
* Request the status of an existing submission using the REST protocol.
- * Standalone cluster mode only.
+ * Standalone and Mesos cluster mode only.
*/
private def requestStatus(args: SparkSubmitArguments): Unit = {
- new StandaloneRestClient()
+ new RestSubmissionClient()
.requestSubmissionStatus(args.master, args.submissionToRequestStatusFor)
}
@@ -252,6 +254,7 @@ object SparkSubmit {
}
val isYarnCluster = clusterManager == YARN && deployMode == CLUSTER
+ val isMesosCluster = clusterManager == MESOS && deployMode == CLUSTER
// Resolve maven dependencies if there are any and add classpath to jars. Add them to py-files
// too for packages that include Python code
@@ -294,8 +297,9 @@ object SparkSubmit {
// The following modes are not supported or applicable
(clusterManager, deployMode) match {
- case (MESOS, CLUSTER) =>
- printErrorAndExit("Cluster deploy mode is currently not supported for Mesos clusters.")
+ case (MESOS, CLUSTER) if args.isPython =>
+ printErrorAndExit("Cluster deploy mode is currently not supported for python " +
+ "applications on Mesos clusters.")
case (STANDALONE, CLUSTER) if args.isPython =>
printErrorAndExit("Cluster deploy mode is currently not supported for python " +
"applications on standalone clusters.")
@@ -377,15 +381,6 @@ object SparkSubmit {
OptionAssigner(args.driverExtraLibraryPath, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,
sysProp = "spark.driver.extraLibraryPath"),
- // 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.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,
- sysProp = "spark.driver.supervise"),
-
// Yarn client only
OptionAssigner(args.queue, YARN, CLIENT, sysProp = "spark.yarn.queue"),
OptionAssigner(args.numExecutors, YARN, CLIENT, sysProp = "spark.executor.instances"),
@@ -413,7 +408,15 @@ object SparkSubmit {
OptionAssigner(args.totalExecutorCores, STANDALONE | MESOS, ALL_DEPLOY_MODES,
sysProp = "spark.cores.max"),
OptionAssigner(args.files, LOCAL | STANDALONE | MESOS, ALL_DEPLOY_MODES,
- sysProp = "spark.files")
+ sysProp = "spark.files"),
+ OptionAssigner(args.jars, STANDALONE | MESOS, CLUSTER, sysProp = "spark.jars"),
+ 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"),
+ OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy")
)
// In client mode, launch the application main class directly
@@ -452,7 +455,7 @@ object SparkSubmit {
// All Spark parameters are expected to be passed to the client through system properties.
if (args.isStandaloneCluster) {
if (args.useRest) {
- childMainClass = "org.apache.spark.deploy.rest.StandaloneRestClient"
+ childMainClass = "org.apache.spark.deploy.rest.RestSubmissionClient"
childArgs += (args.primaryResource, args.mainClass)
} else {
// In legacy standalone cluster mode, use Client as a wrapper around the user class
@@ -496,6 +499,15 @@ object SparkSubmit {
}
}
+ if (isMesosCluster) {
+ assert(args.useRest, "Mesos cluster mode is only supported through the REST submission API")
+ childMainClass = "org.apache.spark.deploy.rest.RestSubmissionClient"
+ childArgs += (args.primaryResource, args.mainClass)
+ if (args.childArgs != null) {
+ childArgs ++= args.childArgs
+ }
+ }
+
// Load any properties specified through --conf and the default properties file
for ((k, v) <- args.sparkProperties) {
sysProps.getOrElseUpdate(k, v)
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
index c896842..c621b8f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
@@ -241,8 +241,9 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
}
private def validateKillArguments(): Unit = {
- if (!master.startsWith("spark://")) {
- SparkSubmit.printErrorAndExit("Killing submissions is only supported in standalone mode!")
+ if (!master.startsWith("spark://") && !master.startsWith("mesos://")) {
+ SparkSubmit.printErrorAndExit(
+ "Killing submissions is only supported in standalone or Mesos mode!")
}
if (submissionToKill == null) {
SparkSubmit.printErrorAndExit("Please specify a submission to kill.")
@@ -250,9 +251,9 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
}
private def validateStatusRequestArguments(): Unit = {
- if (!master.startsWith("spark://")) {
+ if (!master.startsWith("spark://") && !master.startsWith("mesos://")) {
SparkSubmit.printErrorAndExit(
- "Requesting submission statuses is only supported in standalone mode!")
+ "Requesting submission statuses is only supported in standalone or Mesos mode!")
}
if (submissionToRequestStatusFor == null) {
SparkSubmit.printErrorAndExit("Please specify a submission to request status for.")
@@ -485,6 +486,8 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
|
| Spark standalone with cluster deploy mode only:
| --driver-cores NUM Cores for driver (Default: 1).
+ |
+ | Spark standalone or Mesos with cluster deploy mode only:
| --supervise If given, restarts the driver on failure.
| --kill SUBMISSION_ID If given, kills the driver specified.
| --status SUBMISSION_ID If given, requests the status of the driver specified.
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
----------------------------------------------------------------------
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
index ff2eed6..1c21c17 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+++ b/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
}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/master/SparkCuratorUtil.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/SparkCuratorUtil.scala b/core/src/main/scala/org/apache/spark/deploy/master/SparkCuratorUtil.scala
deleted file mode 100644
index 5b22481..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/master/SparkCuratorUtil.scala
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.deploy.master
-
-import scala.collection.JavaConversions._
-
-import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
-import org.apache.curator.retry.ExponentialBackoffRetry
-import org.apache.zookeeper.KeeperException
-
-import org.apache.spark.{Logging, SparkConf}
-
-private[deploy] object SparkCuratorUtil extends Logging {
-
- private val ZK_CONNECTION_TIMEOUT_MILLIS = 15000
- private val ZK_SESSION_TIMEOUT_MILLIS = 60000
- private val RETRY_WAIT_MILLIS = 5000
- private val MAX_RECONNECT_ATTEMPTS = 3
-
- def newClient(conf: SparkConf): CuratorFramework = {
- val ZK_URL = conf.get("spark.deploy.zookeeper.url")
- val zk = CuratorFrameworkFactory.newClient(ZK_URL,
- ZK_SESSION_TIMEOUT_MILLIS, ZK_CONNECTION_TIMEOUT_MILLIS,
- new ExponentialBackoffRetry(RETRY_WAIT_MILLIS, MAX_RECONNECT_ATTEMPTS))
- zk.start()
- zk
- }
-
- def mkdir(zk: CuratorFramework, path: String) {
- if (zk.checkExists().forPath(path) == null) {
- try {
- zk.create().creatingParentsIfNeeded().forPath(path)
- } catch {
- case nodeExist: KeeperException.NodeExistsException =>
- // do nothing, ignore node existing exception.
- case e: Exception => throw e
- }
- }
- }
-
- def deleteRecursive(zk: CuratorFramework, path: String) {
- if (zk.checkExists().forPath(path) != null) {
- for (child <- zk.getChildren.forPath(path)) {
- zk.delete().forPath(path + "/" + child)
- }
- zk.delete().forPath(path)
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala b/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala
index 4823fd7..52758d6 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperLeaderElectionAgent.scala
@@ -23,6 +23,7 @@ import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.deploy.master.MasterMessages._
import org.apache.curator.framework.CuratorFramework
import org.apache.curator.framework.recipes.leader.{LeaderLatchListener, LeaderLatch}
+import org.apache.spark.deploy.SparkCuratorUtil
private[master] class ZooKeeperLeaderElectionAgent(val masterActor: LeaderElectable,
conf: SparkConf) extends LeaderLatchListener with LeaderElectionAgent with Logging {
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala b/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala
index a285783..80db6d4 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala
@@ -26,6 +26,7 @@ import org.apache.curator.framework.CuratorFramework
import org.apache.zookeeper.CreateMode
import org.apache.spark.{Logging, SparkConf}
+import org.apache.spark.deploy.SparkCuratorUtil
private[master] class ZooKeeperPersistenceEngine(conf: SparkConf, val serialization: Serialization)
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
new file mode 100644
index 0000000..5d4e5b8
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos
+
+import java.util.concurrent.CountDownLatch
+
+import org.apache.spark.deploy.mesos.ui.MesosClusterUI
+import org.apache.spark.deploy.rest.mesos.MesosRestServer
+import org.apache.spark.scheduler.cluster.mesos._
+import org.apache.spark.util.SignalLogger
+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 is a long-running process started by the user in
+ * the cluster independently of Spark applications.
+ * It contains a [[MesosRestServer]] that listens for requests to submit drivers and a
+ * [[MesosClusterScheduler]] that processes these requests by negotiating with the Mesos master
+ * for resources.
+ *
+ * A typical new driver lifecycle is the following:
+ * - Driver submitted via spark-submit talking to the [[MesosRestServer]]
+ * - [[MesosRestServer]] queues the driver request to [[MesosClusterScheduler]]
+ * - [[MesosClusterScheduler]] gets resource offers and launches the drivers that are in queue
+ *
+ * This dispatcher supports both Mesos fine-grain or coarse-grain mode as the mode is configurable
+ * per driver launched.
+ * This class is needed since Mesos doesn't manage frameworks, so the dispatcher acts as
+ * a daemon to launch drivers as Mesos frameworks upon request. The dispatcher is also started and
+ * stopped by sbin/start-mesos-dispatcher and sbin/stop-mesos-dispatcher respectively.
+ */
+private[mesos] class MesosClusterDispatcher(
+ args: MesosClusterDispatcherArguments,
+ conf: SparkConf)
+ extends Logging {
+
+ private val publicAddress = Option(conf.getenv("SPARK_PUBLIC_DNS")).getOrElse(args.host)
+ private val recoveryMode = conf.get("spark.mesos.deploy.recoveryMode", "NONE").toUpperCase()
+ logInfo("Recovery mode in Mesos dispatcher set to: " + recoveryMode)
+
+ private val engineFactory = recoveryMode match {
+ case "NONE" => new BlackHoleMesosClusterPersistenceEngineFactory
+ case "ZOOKEEPER" => new ZookeeperMesosClusterPersistenceEngineFactory(conf)
+ case _ => throw new IllegalArgumentException("Unsupported recovery mode: " + recoveryMode)
+ }
+
+ private val scheduler = new MesosClusterScheduler(engineFactory, conf)
+
+ private val server = new MesosRestServer(args.host, args.port, conf, scheduler)
+ private val webUi = new MesosClusterUI(
+ new SecurityManager(conf),
+ args.webUiPort,
+ conf,
+ publicAddress,
+ scheduler)
+
+ private val shutdownLatch = new CountDownLatch(1)
+
+ def start(): Unit = {
+ webUi.bind()
+ scheduler.frameworkUrl = webUi.activeWebUiUrl
+ scheduler.start()
+ server.start()
+ }
+
+ def awaitShutdown(): Unit = {
+ shutdownLatch.await()
+ }
+
+ def stop(): Unit = {
+ webUi.stop()
+ server.stop()
+ scheduler.stop()
+ shutdownLatch.countDown()
+ }
+}
+
+private[mesos] object MesosClusterDispatcher extends Logging {
+ def main(args: Array[String]) {
+ SignalLogger.register(log)
+ val conf = new SparkConf
+ val dispatcherArgs = new MesosClusterDispatcherArguments(args, conf)
+ conf.setMaster(dispatcherArgs.masterUrl)
+ conf.setAppName(dispatcherArgs.name)
+ dispatcherArgs.zookeeperUrl.foreach { z =>
+ conf.set("spark.mesos.deploy.recoveryMode", "ZOOKEEPER")
+ conf.set("spark.mesos.deploy.zookeeper.url", z)
+ }
+ val dispatcher = new MesosClusterDispatcher(dispatcherArgs, conf)
+ dispatcher.start()
+ val shutdownHook = new Thread() {
+ override def run() {
+ logInfo("Shutdown hook is shutting down dispatcher")
+ dispatcher.stop()
+ dispatcher.awaitShutdown()
+ }
+ }
+ Runtime.getRuntime.addShutdownHook(shutdownHook)
+ dispatcher.awaitShutdown()
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
new file mode 100644
index 0000000..894cb78
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos
+
+import org.apache.spark.SparkConf
+import org.apache.spark.util.{IntParam, Utils}
+
+
+private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf: SparkConf) {
+ var host = Utils.localHostName()
+ var port = 7077
+ var name = "Spark Cluster"
+ var webUiPort = 8081
+ var masterUrl: String = _
+ var zookeeperUrl: Option[String] = None
+ var propertiesFile: String = _
+
+ parse(args.toList)
+
+ propertiesFile = Utils.loadDefaultSparkProperties(conf, propertiesFile)
+
+ private def parse(args: List[String]): Unit = args match {
+ case ("--host" | "-h") :: value :: tail =>
+ Utils.checkHost(value, "Please use hostname " + value)
+ host = value
+ parse(tail)
+
+ case ("--port" | "-p") :: IntParam(value) :: tail =>
+ port = value
+ parse(tail)
+
+ case ("--webui-port" | "-p") :: IntParam(value) :: tail =>
+ webUiPort = value
+ parse(tail)
+
+ case ("--zk" | "-z") :: value :: tail =>
+ zookeeperUrl = Some(value)
+ parse(tail)
+
+ case ("--master" | "-m") :: value :: tail =>
+ if (!value.startsWith("mesos://")) {
+ System.err.println("Cluster dispatcher only supports mesos (uri begins with mesos://)")
+ System.exit(1)
+ }
+ masterUrl = value.stripPrefix("mesos://")
+ parse(tail)
+
+ case ("--name") :: value :: tail =>
+ name = value
+ parse(tail)
+
+ case ("--properties-file") :: value :: tail =>
+ propertiesFile = value
+ parse(tail)
+
+ case ("--help") :: tail =>
+ printUsageAndExit(0)
+
+ case Nil => {
+ if (masterUrl == null) {
+ System.err.println("--master is required")
+ printUsageAndExit(1)
+ }
+ }
+
+ case _ =>
+ printUsageAndExit(1)
+ }
+
+ private def printUsageAndExit(exitCode: Int): Unit = {
+ System.err.println(
+ "Usage: MesosClusterDispatcher [options]\n" +
+ "\n" +
+ "Options:\n" +
+ " -h HOST, --host HOST Hostname to listen on\n" +
+ " -p PORT, --port PORT Port to listen on (default: 7077)\n" +
+ " --webui-port WEBUI_PORT WebUI Port to listen on (default: 8081)\n" +
+ " --name NAME Framework name to show in Mesos UI\n" +
+ " -m --master MASTER URI for connecting to Mesos master\n" +
+ " -z --zk ZOOKEEPER Comma delimited URLs for connecting to \n" +
+ " Zookeeper for persistence\n" +
+ " --properties-file FILE Path to a custom Spark properties file.\n" +
+ " Default is conf/spark-defaults.conf.")
+ System.exit(exitCode)
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/mesos/MesosDriverDescription.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosDriverDescription.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosDriverDescription.scala
new file mode 100644
index 0000000..1948226
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosDriverDescription.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos
+
+import java.util.Date
+
+import org.apache.spark.deploy.Command
+import org.apache.spark.scheduler.cluster.mesos.MesosClusterRetryState
+
+/**
+ * Describes a Spark driver that is submitted from the
+ * [[org.apache.spark.deploy.rest.mesos.MesosRestServer]], to be launched by
+ * [[org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler]].
+ * @param jarUrl URL to the application jar
+ * @param mem Amount of memory for the driver
+ * @param cores Number of cores for the driver
+ * @param supervise Supervise the driver for long running app
+ * @param command The command to launch the driver.
+ * @param schedulerProperties Extra properties to pass the Mesos scheduler
+ */
+private[spark] class MesosDriverDescription(
+ val name: String,
+ val jarUrl: String,
+ val mem: Int,
+ val cores: Double,
+ val supervise: Boolean,
+ val command: Command,
+ val schedulerProperties: Map[String, String],
+ val submissionId: String,
+ val submissionDate: Date,
+ val retryState: Option[MesosClusterRetryState] = None)
+ extends Serializable {
+
+ def copy(
+ name: String = name,
+ jarUrl: String = jarUrl,
+ mem: Int = mem,
+ cores: Double = cores,
+ supervise: Boolean = supervise,
+ command: Command = command,
+ schedulerProperties: Map[String, String] = schedulerProperties,
+ submissionId: String = submissionId,
+ submissionDate: Date = submissionDate,
+ retryState: Option[MesosClusterRetryState] = retryState): MesosDriverDescription = {
+ new MesosDriverDescription(name, jarUrl, mem, cores, supervise, command, schedulerProperties,
+ submissionId, submissionDate, retryState)
+ }
+
+ override def toString: String = s"MesosDriverDescription (${command.mainClass})"
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala
new file mode 100644
index 0000000..7b2005e
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos.ui
+
+import javax.servlet.http.HttpServletRequest
+
+import scala.xml.Node
+
+import org.apache.mesos.Protos.TaskStatus
+import org.apache.spark.deploy.mesos.MesosDriverDescription
+import org.apache.spark.scheduler.cluster.mesos.MesosClusterSubmissionState
+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.getSchedulerState()
+ val queuedHeaders = Seq("Driver ID", "Submit Date", "Main Class", "Driver Resources")
+ val driverHeaders = queuedHeaders ++
+ Seq("Start Date", "Mesos Slave ID", "State")
+ val retryHeaders = Seq("Driver ID", "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.pendingRetryDrivers)
+ val content =
+ <p>Mesos Framework ID: {state.frameworkId}</p>
+ <div class="row-fluid">
+ <div class="span12">
+ <h4>Queued Drivers:</h4>
+ {queuedTable}
+ <h4>Launched Drivers:</h4>
+ {launchedTable}
+ <h4>Finished Drivers:</h4>
+ {finishedTable}
+ <h4>Supervise drivers waiting for retry:</h4>
+ {retryTable}
+ </div>
+ </div>;
+ UIUtils.basicSparkPage(content, "Spark Drivers for Mesos cluster")
+ }
+
+ private def queuedRow(submission: MesosDriverDescription): Seq[Node] = {
+ <tr>
+ <td>{submission.submissionId}</td>
+ <td>{submission.submissionDate}</td>
+ <td>{submission.command.mainClass}</td>
+ <td>cpus: {submission.cores}, mem: {submission.mem}</td>
+ </tr>
+ }
+
+ private def driverRow(state: MesosClusterSubmissionState): Seq[Node] = {
+ <tr>
+ <td>{state.driverDescription.submissionId}</td>
+ <td>{state.driverDescription.submissionDate}</td>
+ <td>{state.driverDescription.command.mainClass}</td>
+ <td>cpus: {state.driverDescription.cores}, mem: {state.driverDescription.mem}</td>
+ <td>{state.startDate}</td>
+ <td>{state.slaveId.getValue}</td>
+ <td>{stateString(state.mesosTaskStatus)}</td>
+ </tr>
+ }
+
+ private def retryRow(submission: MesosDriverDescription): Seq[Node] = {
+ <tr>
+ <td>{submission.submissionId}</td>
+ <td>{submission.submissionDate}</td>
+ <td>{submission.command.mainClass}</td>
+ <td>{submission.retryState.get.lastFailureStatus}</td>
+ <td>{submission.retryState.get.nextRetry}</td>
+ <td>{submission.retryState.get.retries}</td>
+ </tr>
+ }
+
+ private def stateString(status: Option[TaskStatus]): String = {
+ if (status.isEmpty) {
+ return ""
+ }
+ val sb = new StringBuilder
+ val s = status.get
+ sb.append(s"State: ${s.getState}")
+ if (status.get.hasMessage) {
+ sb.append(s", Message: ${s.getMessage}")
+ }
+ if (status.get.hasHealthy) {
+ sb.append(s", Healthy: ${s.getHealthy}")
+ }
+ if (status.get.hasSource) {
+ sb.append(s", Source: ${s.getSource}")
+ }
+ if (status.get.hasReason) {
+ sb.append(s", Reason: ${s.getReason}")
+ }
+ if (status.get.hasTimestamp) {
+ sb.append(s", Time: ${s.getTimestamp}")
+ }
+ sb.toString()
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterUI.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterUI.scala
new file mode 100644
index 0000000..4865d46
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterUI.scala
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.mesos.ui
+
+import org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler
+import org.apache.spark.{SecurityManager, SparkConf}
+import org.apache.spark.ui.JettyUtils._
+import org.apache.spark.ui.{SparkUI, WebUI}
+
+/**
+ * UI that displays driver results from the [[org.apache.spark.deploy.mesos.MesosClusterDispatcher]]
+ */
+private[spark] class MesosClusterUI(
+ securityManager: SecurityManager,
+ port: Int,
+ conf: SparkConf,
+ dispatcherPublicAddress: String,
+ val scheduler: MesosClusterScheduler)
+ extends WebUI(securityManager, port, conf) {
+
+ initialize()
+
+ def activeWebUiUrl: String = "http://" + dispatcherPublicAddress + ":" + boundPort
+
+ override def initialize() {
+ attachPage(new MesosClusterPage(this))
+ attachHandler(createStaticHandler(MesosClusterUI.STATIC_RESOURCE_DIR, "/static"))
+ }
+}
+
+private object MesosClusterUI {
+ val STATIC_RESOURCE_DIR = SparkUI.STATIC_RESOURCE_DIR
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
new file mode 100644
index 0000000..307cebf
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.rest
+
+import java.io.{DataOutputStream, FileNotFoundException}
+import java.net.{HttpURLConnection, SocketException, URL}
+import javax.servlet.http.HttpServletResponse
+
+import scala.io.Source
+
+import com.fasterxml.jackson.core.JsonProcessingException
+import com.google.common.base.Charsets
+
+import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
+import org.apache.spark.util.Utils
+
+/**
+ * A client that submits applications to a [[RestSubmissionServer]].
+ *
+ * In protocol version v1, the REST URL takes the form http://[host:port]/v1/submissions/[action],
+ * where [action] can be one of create, kill, or status. Each type of request is represented in
+ * an HTTP message sent to the following prefixes:
+ * (1) submit - POST to /submissions/create
+ * (2) kill - POST /submissions/kill/[submissionId]
+ * (3) status - GET /submissions/status/[submissionId]
+ *
+ * In the case of (1), parameters are posted in the HTTP body in the form of JSON fields.
+ * Otherwise, the URL fully specifies the intended action of the client.
+ *
+ * Since the protocol is expected to be stable across Spark versions, existing fields cannot be
+ * added or removed, though new optional fields can be added. In the rare event that forward or
+ * backward compatibility is broken, Spark must introduce a new protocol version (e.g. v2).
+ *
+ * The client and the server must communicate using the same version of the protocol. If there
+ * is a mismatch, the server will respond with the highest protocol version it supports. A future
+ * implementation of this client can use that information to retry using the version specified
+ * by the server.
+ */
+private[spark] class RestSubmissionClient extends Logging {
+ import RestSubmissionClient._
+
+ private val supportedMasterPrefixes = Seq("spark://", "mesos://")
+
+ /**
+ * 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.
+ */
+ 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)
+ }
+ response
+ }
+
+ /** Request that the server kill the specified submission. */
+ def killSubmission(master: String, 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)
+ }
+ 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)
+ }
+ response
+ }
+
+ /** Construct a message that captures the specified parameters for submitting an application. */
+ def constructSubmitRequest(
+ appResource: String,
+ mainClass: String,
+ appArgs: Array[String],
+ sparkProperties: Map[String, String],
+ environmentVariables: Map[String, String]): CreateSubmissionRequest = {
+ val message = new CreateSubmissionRequest
+ message.clientSparkVersion = sparkVersion
+ message.appResource = appResource
+ message.mainClass = mainClass
+ message.appArgs = appArgs
+ message.sparkProperties = sparkProperties
+ message.environmentVariables = environmentVariables
+ message.validate()
+ message
+ }
+
+ /** Send a GET request to the specified URL. */
+ private def get(url: URL): SubmitRestProtocolResponse = {
+ logDebug(s"Sending GET request to server at $url.")
+ val conn = url.openConnection().asInstanceOf[HttpURLConnection]
+ conn.setRequestMethod("GET")
+ readResponse(conn)
+ }
+
+ /** Send a POST request to the specified URL. */
+ private def post(url: URL): SubmitRestProtocolResponse = {
+ logDebug(s"Sending POST request to server at $url.")
+ val conn = url.openConnection().asInstanceOf[HttpURLConnection]
+ conn.setRequestMethod("POST")
+ readResponse(conn)
+ }
+
+ /** Send a POST request with the given JSON as the body to the specified URL. */
+ private def postJson(url: URL, json: String): SubmitRestProtocolResponse = {
+ logDebug(s"Sending POST request to server at $url:\n$json")
+ val conn = url.openConnection().asInstanceOf[HttpURLConnection]
+ conn.setRequestMethod("POST")
+ conn.setRequestProperty("Content-Type", "application/json")
+ conn.setRequestProperty("charset", "utf-8")
+ conn.setDoOutput(true)
+ val out = new DataOutputStream(conn.getOutputStream)
+ Utils.tryWithSafeFinally {
+ out.write(json.getBytes(Charsets.UTF_8))
+ } {
+ out.close()
+ }
+ readResponse(conn)
+ }
+
+ /**
+ * Read the response from the server and return it as a validated [[SubmitRestProtocolResponse]].
+ * If the response represents an error, report the embedded message to the user.
+ * Exposed for testing.
+ */
+ private[rest] def readResponse(connection: HttpURLConnection): SubmitRestProtocolResponse = {
+ try {
+ val dataStream =
+ if (connection.getResponseCode == HttpServletResponse.SC_OK) {
+ connection.getInputStream
+ } else {
+ connection.getErrorStream
+ }
+ // If the server threw an exception while writing a response, it will not have a body
+ if (dataStream == null) {
+ throw new SubmitRestProtocolException("Server returned empty body")
+ }
+ val responseJson = Source.fromInputStream(dataStream).mkString
+ logDebug(s"Response from the server:\n$responseJson")
+ val response = SubmitRestProtocolMessage.fromJson(responseJson)
+ response.validate()
+ response match {
+ // If the response is an error, log the message
+ case error: ErrorResponse =>
+ logError(s"Server responded with error:\n${error.message}")
+ error
+ // Otherwise, simply return the response
+ case response: SubmitRestProtocolResponse => response
+ case unexpected =>
+ throw new SubmitRestProtocolException(
+ s"Message received from server was not a response:\n${unexpected.toJson}")
+ }
+ } catch {
+ case unreachable @ (_: FileNotFoundException | _: SocketException) =>
+ throw new SubmitRestConnectionException(
+ s"Unable to connect to server ${connection.getURL}", unreachable)
+ case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+ throw new SubmitRestProtocolException(
+ "Malformed response received from server", malformed)
+ }
+ }
+
+ /** Return the REST URL for creating a new submission. */
+ private def getSubmitUrl(master: String): URL = {
+ val baseUrl = getBaseUrl(master)
+ new URL(s"$baseUrl/create")
+ }
+
+ /** Return the REST URL for killing an existing submission. */
+ private def getKillUrl(master: String, submissionId: String): URL = {
+ val baseUrl = getBaseUrl(master)
+ new URL(s"$baseUrl/kill/$submissionId")
+ }
+
+ /** Return the REST URL for requesting the status of an existing submission. */
+ private def getStatusUrl(master: String, submissionId: String): URL = {
+ val baseUrl = getBaseUrl(master)
+ new URL(s"$baseUrl/status/$submissionId")
+ }
+
+ /** Return the base URL for communicating with the server, including the protocol version. */
+ private def getBaseUrl(master: String): String = {
+ var masterUrl = master
+ supportedMasterPrefixes.foreach { prefix =>
+ if (master.startsWith(prefix)) {
+ masterUrl = master.stripPrefix(prefix)
+ }
+ }
+ masterUrl = masterUrl.stripSuffix("/")
+ s"http://$masterUrl/$PROTOCOL_VERSION/submissions"
+ }
+
+ /** Throw an exception if this is not standalone mode. */
+ private def validateMaster(master: String): Unit = {
+ val valid = supportedMasterPrefixes.exists { prefix => master.startsWith(prefix) }
+ if (!valid) {
+ throw new IllegalArgumentException(
+ "This REST client only supports master URLs that start with " +
+ "one of the following: " + supportedMasterPrefixes.mkString(","))
+ }
+ }
+
+ /** 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!")
+ }
+ } else {
+ val failMessage = Option(submitResponse.message).map { ": " + _ }.getOrElse("")
+ logError(s"Application submission failed$failMessage")
+ }
+ }
+
+ /**
+ * 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 = {
+ (1 to REPORT_DRIVER_STATUS_MAX_TRIES).foreach { _ =>
+ val response = requestSubmissionStatus(master, submissionId, quiet = true)
+ val statusResponse = response match {
+ case s: SubmissionStatusResponse => s
+ case _ => return // unexpected type, let upstream caller handle it
+ }
+ if (statusResponse.success) {
+ val driverState = Option(statusResponse.driverState)
+ val workerId = Option(statusResponse.workerId)
+ val workerHostPort = Option(statusResponse.workerHostPort)
+ val exception = Option(statusResponse.message)
+ // Log driver state, if present
+ driverState match {
+ case Some(state) => logInfo(s"State of driver $submissionId is now $state.")
+ case _ => logError(s"State of driver $submissionId was not found!")
+ }
+ // Log worker node, if present
+ (workerId, workerHostPort) match {
+ case (Some(id), Some(hp)) => logInfo(s"Driver is running on worker $id at $hp.")
+ case _ =>
+ }
+ // Log exception stack trace, if present
+ exception.foreach { e => logError(e) }
+ return
+ }
+ Thread.sleep(REPORT_DRIVER_STATUS_INTERVAL)
+ }
+ logError(s"Error: Master did not recognize driver $submissionId.")
+ }
+
+ /** Log the response sent by the server in the REST application submission protocol. */
+ private def handleRestResponse(response: SubmitRestProtocolResponse): Unit = {
+ logInfo(s"Server responded with ${response.messageType}:\n${response.toJson}")
+ }
+
+ /** Log an appropriate error if the response sent by the server is not of the expected type. */
+ private def handleUnexpectedRestResponse(unexpected: SubmitRestProtocolResponse): Unit = {
+ logError(s"Error: Server responded with message of unexpected type ${unexpected.messageType}.")
+ }
+}
+
+private[spark] object RestSubmissionClient {
+ private val REPORT_DRIVER_STATUS_INTERVAL = 1000
+ private val REPORT_DRIVER_STATUS_MAX_TRIES = 10
+ val PROTOCOL_VERSION = "v1"
+
+ /**
+ * Submit an application, assuming Spark parameters are specified through the given config.
+ * This is abstracted to its own method for testing purposes.
+ */
+ def run(
+ appResource: String,
+ mainClass: String,
+ appArgs: Array[String],
+ conf: SparkConf,
+ env: Map[String, String] = sys.env): SubmitRestProtocolResponse = {
+ val master = conf.getOption("spark.master").getOrElse {
+ throw new IllegalArgumentException("'spark.master' must be set.")
+ }
+ val sparkProperties = conf.getAll.toMap
+ val environmentVariables = env.filter { case (k, _) => k.startsWith("SPARK_") }
+ val client = new RestSubmissionClient
+ val submitRequest = client.constructSubmitRequest(
+ appResource, mainClass, appArgs, sparkProperties, environmentVariables)
+ client.createSubmission(master, submitRequest)
+ }
+
+ def main(args: Array[String]): Unit = {
+ if (args.size < 2) {
+ sys.error("Usage: RestSubmissionClient [app resource] [main class] [app args*]")
+ sys.exit(1)
+ }
+ val appResource = args(0)
+ val mainClass = args(1)
+ val appArgs = args.slice(2, args.size)
+ val conf = new SparkConf
+ run(appResource, mainClass, appArgs, conf)
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
new file mode 100644
index 0000000..2e78d03
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.rest
+
+import java.net.InetSocketAddress
+import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}
+
+import scala.io.Source
+import com.fasterxml.jackson.core.JsonProcessingException
+import org.eclipse.jetty.server.Server
+import org.eclipse.jetty.servlet.{ServletHolder, ServletContextHandler}
+import org.eclipse.jetty.util.thread.QueuedThreadPool
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
+import org.apache.spark.util.Utils
+
+/**
+ * A server that responds to requests submitted by the [[RestSubmissionClient]].
+ *
+ * This server responds with different HTTP codes depending on the situation:
+ * 200 OK - Request was processed successfully
+ * 400 BAD REQUEST - Request was malformed, not successfully validated, or of unexpected type
+ * 468 UNKNOWN PROTOCOL VERSION - Request specified a protocol this server does not understand
+ * 500 INTERNAL SERVER ERROR - Server throws an exception internally while processing the request
+ *
+ * The server always includes a JSON representation of the relevant [[SubmitRestProtocolResponse]]
+ * in the HTTP body. If an error occurs, however, the server will include an [[ErrorResponse]]
+ * instead of the one expected by the client. If the construction of this error response itself
+ * fails, the response will consist of an empty body with a response code that indicates internal
+ * server error.
+ */
+private[spark] abstract class RestSubmissionServer(
+ val host: String,
+ val requestedPort: Int,
+ val masterConf: SparkConf) extends Logging {
+ protected val submitRequestServlet: SubmitRequestServlet
+ protected val killRequestServlet: KillRequestServlet
+ protected val 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"/${RestSubmissionServer.PROTOCOL_VERSION}/submissions"
+ protected lazy val contextToServlet = Map[String, RestServlet](
+ s"$baseContext/create/*" -> submitRequestServlet,
+ s"$baseContext/kill/*" -> killRequestServlet,
+ s"$baseContext/status/*" -> statusRequestServlet,
+ "/*" -> new ErrorServlet // default handler
+ )
+
+ /** Start the server and return the bound port. */
+ def start(): Int = {
+ val (server, boundPort) = Utils.startServiceOnPort[Server](requestedPort, doStart, masterConf)
+ _server = Some(server)
+ logInfo(s"Started REST server for submitting applications on port $boundPort")
+ boundPort
+ }
+
+ /**
+ * Map the servlets to their corresponding contexts and attach them to a server.
+ * Return a 2-tuple of the started server and the bound port.
+ */
+ private def doStart(startPort: Int): (Server, Int) = {
+ val server = new Server(new InetSocketAddress(host, startPort))
+ val threadPool = new QueuedThreadPool
+ threadPool.setDaemon(true)
+ server.setThreadPool(threadPool)
+ val mainHandler = new ServletContextHandler
+ mainHandler.setContextPath("/")
+ contextToServlet.foreach { case (prefix, servlet) =>
+ mainHandler.addServlet(new ServletHolder(servlet), prefix)
+ }
+ server.setHandler(mainHandler)
+ server.start()
+ val boundPort = server.getConnectors()(0).getLocalPort
+ (server, boundPort)
+ }
+
+ def stop(): Unit = {
+ _server.foreach(_.stop())
+ }
+}
+
+private[rest] object RestSubmissionServer {
+ val PROTOCOL_VERSION = RestSubmissionClient.PROTOCOL_VERSION
+ val SC_UNKNOWN_PROTOCOL_VERSION = 468
+}
+
+/**
+ * An abstract servlet for handling requests passed to the [[RestSubmissionServer]].
+ */
+private[rest] abstract class RestServlet extends HttpServlet with Logging {
+
+ /**
+ * Serialize the given response message to JSON and send it through the response servlet.
+ * This validates the response before sending it to ensure it is properly constructed.
+ */
+ protected def sendResponse(
+ responseMessage: SubmitRestProtocolResponse,
+ responseServlet: HttpServletResponse): Unit = {
+ val message = validateResponse(responseMessage, responseServlet)
+ responseServlet.setContentType("application/json")
+ responseServlet.setCharacterEncoding("utf-8")
+ responseServlet.getWriter.write(message.toJson)
+ }
+
+ /**
+ * Return any fields in the client request message that the server does not know about.
+ *
+ * The mechanism for this is to reconstruct the JSON on the server side and compare the
+ * diff between this JSON and the one generated on the client side. Any fields that are
+ * only in the client JSON are treated as unexpected.
+ */
+ protected def findUnknownFields(
+ requestJson: String,
+ requestMessage: SubmitRestProtocolMessage): Array[String] = {
+ val clientSideJson = parse(requestJson)
+ val serverSideJson = parse(requestMessage.toJson)
+ val Diff(_, _, unknown) = clientSideJson.diff(serverSideJson)
+ unknown match {
+ case j: JObject => j.obj.map { case (k, _) => k }.toArray
+ case _ => Array.empty[String] // No difference
+ }
+ }
+
+ /** Return a human readable String representation of the exception. */
+ protected def formatException(e: Throwable): String = {
+ val stackTraceString = e.getStackTrace.map { "\t" + _ }.mkString("\n")
+ s"$e\n$stackTraceString"
+ }
+
+ /** Construct an error message to signal the fact that an exception has been thrown. */
+ protected def handleError(message: String): ErrorResponse = {
+ val e = new ErrorResponse
+ e.serverSparkVersion = sparkVersion
+ e.message = message
+ e
+ }
+
+ /**
+ * Parse a submission ID from the relative path, assuming it is the first part of the path.
+ * For instance, we expect the path to take the form /[submission ID]/maybe/something/else.
+ * The returned submission ID cannot be empty. If the path is unexpected, return None.
+ */
+ protected def parseSubmissionId(path: String): Option[String] = {
+ if (path == null || path.isEmpty) {
+ None
+ } else {
+ path.stripPrefix("/").split("/").headOption.filter(_.nonEmpty)
+ }
+ }
+
+ /**
+ * Validate the response to ensure that it is correctly constructed.
+ *
+ * If it is, simply return the message as is. Otherwise, return an error response instead
+ * to propagate the exception back to the client and set the appropriate error code.
+ */
+ private def validateResponse(
+ responseMessage: SubmitRestProtocolResponse,
+ responseServlet: HttpServletResponse): SubmitRestProtocolResponse = {
+ try {
+ responseMessage.validate()
+ responseMessage
+ } catch {
+ case e: Exception =>
+ responseServlet.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR)
+ handleError("Internal server error: " + formatException(e))
+ }
+ }
+}
+
+/**
+ * A servlet for handling kill requests passed to the [[RestSubmissionServer]].
+ */
+private[rest] abstract class KillRequestServlet extends RestServlet {
+
+ /**
+ * If a submission ID is specified in the URL, have the Master kill the corresponding
+ * driver and return an appropriate response to the client. Otherwise, return error.
+ */
+ protected override def doPost(
+ request: HttpServletRequest,
+ response: HttpServletResponse): Unit = {
+ val submissionId = parseSubmissionId(request.getPathInfo)
+ val responseMessage = submissionId.map(handleKill).getOrElse {
+ response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+ handleError("Submission ID is missing in kill request.")
+ }
+ sendResponse(responseMessage, response)
+ }
+
+ protected def handleKill(submissionId: String): KillSubmissionResponse
+}
+
+/**
+ * A servlet for handling status requests passed to the [[RestSubmissionServer]].
+ */
+private[rest] abstract class StatusRequestServlet extends RestServlet {
+
+ /**
+ * If a submission ID is specified in the URL, request the status of the corresponding
+ * driver from the Master and include it in the response. Otherwise, return error.
+ */
+ protected override def doGet(
+ request: HttpServletRequest,
+ response: HttpServletResponse): Unit = {
+ val submissionId = parseSubmissionId(request.getPathInfo)
+ val responseMessage = submissionId.map(handleStatus).getOrElse {
+ response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+ handleError("Submission ID is missing in status request.")
+ }
+ sendResponse(responseMessage, response)
+ }
+
+ protected def handleStatus(submissionId: String): SubmissionStatusResponse
+}
+
+/**
+ * A servlet for handling submit requests passed to the [[RestSubmissionServer]].
+ */
+private[rest] abstract class SubmitRequestServlet extends RestServlet {
+
+ /**
+ * Submit an application to the Master with parameters specified in the request.
+ *
+ * The request is assumed to be a [[SubmitRestProtocolRequest]] in the form of JSON.
+ * If the request is successfully processed, return an appropriate response to the
+ * client indicating so. Otherwise, return error instead.
+ */
+ protected override def doPost(
+ requestServlet: HttpServletRequest,
+ responseServlet: HttpServletResponse): Unit = {
+ val responseMessage =
+ try {
+ val requestMessageJson = Source.fromInputStream(requestServlet.getInputStream).mkString
+ val requestMessage = SubmitRestProtocolMessage.fromJson(requestMessageJson)
+ // The response should have already been validated on the client.
+ // In case this is not true, validate it ourselves to avoid potential NPEs.
+ requestMessage.validate()
+ handleSubmit(requestMessageJson, requestMessage, responseServlet)
+ } catch {
+ // The client failed to provide a valid JSON, so this is not our fault
+ case e @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+ responseServlet.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+ handleError("Malformed request: " + formatException(e))
+ }
+ sendResponse(responseMessage, responseServlet)
+ }
+
+ protected def handleSubmit(
+ requestMessageJson: String,
+ requestMessage: SubmitRestProtocolMessage,
+ responseServlet: HttpServletResponse): SubmitRestProtocolResponse
+}
+
+/**
+ * A default servlet that handles error cases that are not captured by other servlets.
+ */
+private class ErrorServlet extends RestServlet {
+ private val serverVersion = RestSubmissionServer.PROTOCOL_VERSION
+
+ /** Service a faulty request by returning an appropriate error message to the client. */
+ protected override def service(
+ request: HttpServletRequest,
+ response: HttpServletResponse): Unit = {
+ val path = request.getPathInfo
+ val parts = path.stripPrefix("/").split("/").filter(_.nonEmpty).toList
+ var versionMismatch = false
+ var msg =
+ parts match {
+ case Nil =>
+ // http://host:port/
+ "Missing protocol version."
+ case `serverVersion` :: Nil =>
+ // http://host:port/correct-version
+ "Missing the /submissions prefix."
+ case `serverVersion` :: "submissions" :: tail =>
+ // http://host:port/correct-version/submissions/*
+ "Missing an action: please specify one of /create, /kill, or /status."
+ case unknownVersion :: tail =>
+ // http://host:port/unknown-version/*
+ versionMismatch = true
+ s"Unknown protocol version '$unknownVersion'."
+ case _ =>
+ // never reached
+ s"Malformed path $path."
+ }
+ msg += s" Please submit requests through http://[host]:[port]/$serverVersion/submissions/..."
+ val error = handleError(msg)
+ // If there is a version mismatch, include the highest protocol version that
+ // this server supports in case the client wants to retry with our version
+ if (versionMismatch) {
+ error.highestProtocolVersion = serverVersion
+ response.setStatus(RestSubmissionServer.SC_UNKNOWN_PROTOCOL_VERSION)
+ } else {
+ response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+ }
+ sendResponse(error, response)
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/53befacc/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestClient.scala
----------------------------------------------------------------------
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
deleted file mode 100644
index b8fd406..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestClient.scala
+++ /dev/null
@@ -1,335 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.deploy.rest
-
-import java.io.{DataOutputStream, FileNotFoundException}
-import java.net.{HttpURLConnection, SocketException, URL}
-import javax.servlet.http.HttpServletResponse
-
-import scala.io.Source
-
-import com.fasterxml.jackson.core.JsonProcessingException
-import com.google.common.base.Charsets
-
-import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
-import org.apache.spark.util.Utils
-
-/**
- * A client that submits applications to the standalone Master using a REST protocol.
- * This client is intended to communicate with the [[StandaloneRestServer]] and is
- * currently used for cluster mode only.
- *
- * In protocol version v1, the REST URL takes the form http://[host:port]/v1/submissions/[action],
- * where [action] can be one of create, kill, or status. Each type of request is represented in
- * an HTTP message sent to the following prefixes:
- * (1) submit - POST to /submissions/create
- * (2) kill - POST /submissions/kill/[submissionId]
- * (3) status - GET /submissions/status/[submissionId]
- *
- * In the case of (1), parameters are posted in the HTTP body in the form of JSON fields.
- * Otherwise, the URL fully specifies the intended action of the client.
- *
- * Since the protocol is expected to be stable across Spark versions, existing fields cannot be
- * added or removed, though new optional fields can be added. In the rare event that forward or
- * backward compatibility is broken, Spark must introduce a new protocol version (e.g. v2).
- *
- * The client and the server must communicate using the same version of the protocol. If there
- * is a mismatch, the server will respond with the highest protocol version it supports. A future
- * implementation of this client can use that information to retry using the version specified
- * by the server.
- */
-private[deploy] class StandaloneRestClient extends Logging {
- import StandaloneRestClient._
-
- /**
- * 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)
- }
- response
- }
-
- /** Request that the server kill the specified submission. */
- def killSubmission(master: String, 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)
- }
- 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)
- }
- response
- }
-
- /** Construct a message that captures the specified parameters for submitting an application. */
- private[rest] def constructSubmitRequest(
- appResource: String,
- mainClass: String,
- appArgs: Array[String],
- sparkProperties: Map[String, String],
- environmentVariables: Map[String, String]): CreateSubmissionRequest = {
- val message = new CreateSubmissionRequest
- message.clientSparkVersion = sparkVersion
- message.appResource = appResource
- message.mainClass = mainClass
- message.appArgs = appArgs
- message.sparkProperties = sparkProperties
- message.environmentVariables = environmentVariables
- message.validate()
- message
- }
-
- /** Send a GET request to the specified URL. */
- private def get(url: URL): SubmitRestProtocolResponse = {
- logDebug(s"Sending GET request to server at $url.")
- val conn = url.openConnection().asInstanceOf[HttpURLConnection]
- conn.setRequestMethod("GET")
- readResponse(conn)
- }
-
- /** Send a POST request to the specified URL. */
- private def post(url: URL): SubmitRestProtocolResponse = {
- logDebug(s"Sending POST request to server at $url.")
- val conn = url.openConnection().asInstanceOf[HttpURLConnection]
- conn.setRequestMethod("POST")
- readResponse(conn)
- }
-
- /** Send a POST request with the given JSON as the body to the specified URL. */
- private def postJson(url: URL, json: String): SubmitRestProtocolResponse = {
- logDebug(s"Sending POST request to server at $url:\n$json")
- val conn = url.openConnection().asInstanceOf[HttpURLConnection]
- conn.setRequestMethod("POST")
- conn.setRequestProperty("Content-Type", "application/json")
- conn.setRequestProperty("charset", "utf-8")
- conn.setDoOutput(true)
- val out = new DataOutputStream(conn.getOutputStream)
- Utils.tryWithSafeFinally {
- out.write(json.getBytes(Charsets.UTF_8))
- } {
- out.close()
- }
- readResponse(conn)
- }
-
- /**
- * Read the response from the server and return it as a validated [[SubmitRestProtocolResponse]].
- * If the response represents an error, report the embedded message to the user.
- * Exposed for testing.
- */
- private[rest] def readResponse(connection: HttpURLConnection): SubmitRestProtocolResponse = {
- try {
- val dataStream =
- if (connection.getResponseCode == HttpServletResponse.SC_OK) {
- connection.getInputStream
- } else {
- connection.getErrorStream
- }
- // If the server threw an exception while writing a response, it will not have a body
- if (dataStream == null) {
- throw new SubmitRestProtocolException("Server returned empty body")
- }
- val responseJson = Source.fromInputStream(dataStream).mkString
- logDebug(s"Response from the server:\n$responseJson")
- val response = SubmitRestProtocolMessage.fromJson(responseJson)
- response.validate()
- response match {
- // If the response is an error, log the message
- case error: ErrorResponse =>
- logError(s"Server responded with error:\n${error.message}")
- error
- // Otherwise, simply return the response
- case response: SubmitRestProtocolResponse => response
- case unexpected =>
- throw new SubmitRestProtocolException(
- s"Message received from server was not a response:\n${unexpected.toJson}")
- }
- } catch {
- case unreachable @ (_: FileNotFoundException | _: SocketException) =>
- throw new SubmitRestConnectionException(
- s"Unable to connect to server ${connection.getURL}", unreachable)
- case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
- throw new SubmitRestProtocolException(
- "Malformed response received from server", malformed)
- }
- }
-
- /** Return the REST URL for creating a new submission. */
- private def getSubmitUrl(master: String): URL = {
- val baseUrl = getBaseUrl(master)
- new URL(s"$baseUrl/create")
- }
-
- /** Return the REST URL for killing an existing submission. */
- private def getKillUrl(master: String, submissionId: String): URL = {
- val baseUrl = getBaseUrl(master)
- new URL(s"$baseUrl/kill/$submissionId")
- }
-
- /** Return the REST URL for requesting the status of an existing submission. */
- private def getStatusUrl(master: String, submissionId: String): URL = {
- val baseUrl = getBaseUrl(master)
- new URL(s"$baseUrl/status/$submissionId")
- }
-
- /** Return the base URL for communicating with the server, including the protocol version. */
- private def getBaseUrl(master: String): String = {
- val masterUrl = master.stripPrefix("spark://").stripSuffix("/")
- s"http://$masterUrl/$PROTOCOL_VERSION/submissions"
- }
-
- /** Throw an exception if this is not standalone mode. */
- private def validateMaster(master: String): Unit = {
- if (!master.startsWith("spark://")) {
- throw new IllegalArgumentException("This REST client is only supported in standalone mode.")
- }
- }
-
- /** 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!")
- }
- } else {
- val failMessage = Option(submitResponse.message).map { ": " + _ }.getOrElse("")
- logError(s"Application submission failed$failMessage")
- }
- }
-
- /**
- * 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 = {
- (1 to REPORT_DRIVER_STATUS_MAX_TRIES).foreach { _ =>
- val response = requestSubmissionStatus(master, submissionId, quiet = true)
- val statusResponse = response match {
- case s: SubmissionStatusResponse => s
- case _ => return // unexpected type, let upstream caller handle it
- }
- if (statusResponse.success) {
- val driverState = Option(statusResponse.driverState)
- val workerId = Option(statusResponse.workerId)
- val workerHostPort = Option(statusResponse.workerHostPort)
- val exception = Option(statusResponse.message)
- // Log driver state, if present
- driverState match {
- case Some(state) => logInfo(s"State of driver $submissionId is now $state.")
- case _ => logError(s"State of driver $submissionId was not found!")
- }
- // Log worker node, if present
- (workerId, workerHostPort) match {
- case (Some(id), Some(hp)) => logInfo(s"Driver is running on worker $id at $hp.")
- case _ =>
- }
- // Log exception stack trace, if present
- exception.foreach { e => logError(e) }
- return
- }
- Thread.sleep(REPORT_DRIVER_STATUS_INTERVAL)
- }
- logError(s"Error: Master did not recognize driver $submissionId.")
- }
-
- /** Log the response sent by the server in the REST application submission protocol. */
- private def handleRestResponse(response: SubmitRestProtocolResponse): Unit = {
- logInfo(s"Server responded with ${response.messageType}:\n${response.toJson}")
- }
-
- /** Log an appropriate error if the response sent by the server is not of the expected type. */
- private def handleUnexpectedRestResponse(unexpected: SubmitRestProtocolResponse): Unit = {
- logError(s"Error: Server responded with message of unexpected type ${unexpected.messageType}.")
- }
-}
-
-private[rest] object StandaloneRestClient {
- private val REPORT_DRIVER_STATUS_INTERVAL = 1000
- private val REPORT_DRIVER_STATUS_MAX_TRIES = 10
- val PROTOCOL_VERSION = "v1"
-
- /**
- * Submit an application, assuming Spark parameters are specified through the given config.
- * This is abstracted to its own method for testing purposes.
- */
- def run(
- appResource: String,
- mainClass: String,
- appArgs: Array[String],
- conf: SparkConf,
- env: Map[String, String] = sys.env): SubmitRestProtocolResponse = {
- val master = conf.getOption("spark.master").getOrElse {
- throw new IllegalArgumentException("'spark.master' must be set.")
- }
- val sparkProperties = conf.getAll.toMap
- val environmentVariables = env.filter { case (k, _) => k.startsWith("SPARK_") }
- val client = new StandaloneRestClient
- val submitRequest = client.constructSubmitRequest(
- appResource, mainClass, appArgs, sparkProperties, environmentVariables)
- client.createSubmission(master, submitRequest)
- }
-
- def main(args: Array[String]): Unit = {
- if (args.size < 2) {
- sys.error("Usage: StandaloneRestClient [app resource] [main class] [app args*]")
- sys.exit(1)
- }
- val appResource = args(0)
- val mainClass = args(1)
- val appArgs = args.slice(2, args.size)
- val conf = new SparkConf
- run(appResource, mainClass, appArgs, conf)
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org