[SPARK-5338][MESOS] Add cluster mode support for Mesos #5144
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
There was a problem hiding this comment. This will cause a compilation warning complaining that the match is not exhaustive. If the user provides a random string then this will fail at run time with a bad message. Did you mean to make |
||
| 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() | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
technically not needed since we already do another level of validation in the
RestSubmissionClient, but OK to keep