From a2d75722a13088d94c6cc986d753d944adf7ab10 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Mon, 6 Apr 2015 20:29:45 +0900 Subject: [PATCH 1/9] Initial commit --- .../CoarseGrainedExecutorBackend.scala | 6 ++++- .../spark/scheduler/SparkListener.scala | 2 +- .../cluster/CoarseGrainedClusterMessage.scala | 3 ++- .../CoarseGrainedSchedulerBackend.scala | 4 ++-- .../apache/spark/ui/exec/ExecutorsPage.scala | 23 +++++++++++++++---- .../apache/spark/ui/exec/ExecutorsTab.scala | 2 ++ 6 files changed, 31 insertions(+), 9 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala index 8300f9f2190b9..7f5e3a18834cd 100644 --- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala +++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala @@ -49,11 +49,15 @@ private[spark] class CoarseGrainedExecutorBackend( override def onStart() { import scala.concurrent.ExecutionContext.Implicits.global + import sun.misc.VMSupport + val agentProps = VMSupport.getAgentProperties + val debugPortOpt = Option(agentProps.get("sun.jdwp.listenerAddress").asInstanceOf[String]).map(_.split(":")(1).toInt) + logInfo("Connecting to driver: " + driverUrl) rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref => driver = Some(ref) ref.sendWithReply[RegisteredExecutor.type]( - RegisterExecutor(executorId, self, hostPort, cores, extractLogUrls)) + RegisterExecutor(executorId, self, hostPort, cores, extractLogUrls, debugPortOpt)) } onComplete { case Success(msg) => Utils.tryLogNonFatalError { Option(self).foreach(_.send(msg)) // msg must be RegisteredExecutor diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala index b711ff209af94..f00ee5bbb348d 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala @@ -91,7 +91,7 @@ case class SparkListenerBlockManagerRemoved(time: Long, blockManagerId: BlockMan case class SparkListenerUnpersistRDD(rddId: Int) extends SparkListenerEvent @DeveloperApi -case class SparkListenerExecutorAdded(time: Long, executorId: String, executorInfo: ExecutorInfo) +case class SparkListenerExecutorAdded(time: Long, executorId: String, executorInfo: ExecutorInfo, debugPortOpt: Option[Int] = None) extends SparkListenerEvent @DeveloperApi diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala index 70364cea62a80..171efd7a7b3a3 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala @@ -45,7 +45,8 @@ private[spark] object CoarseGrainedClusterMessages { executorRef: RpcEndpointRef, hostPort: String, cores: Int, - logUrls: Map[String, String]) + logUrls: Map[String, String], + debugPortOpt: Option[Int]) extends CoarseGrainedClusterMessage { Utils.checkHostPort(hostPort, "Expected host port") } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index 4c49da87af9dc..33fe3183af786 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -114,7 +114,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp } override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = { - case RegisterExecutor(executorId, executorRef, hostPort, cores, logUrls) => + case RegisterExecutor(executorId, executorRef, hostPort, cores, logUrls, debugPortOpt) => Utils.checkHostPort(hostPort, "Host port expected " + hostPort) if (executorDataMap.contains(executorId)) { context.reply(RegisterExecutorFailed("Duplicate executor ID: " + executorId)) @@ -137,7 +137,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp } } listenerBus.post( - SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data)) + SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data, debugPortOpt)) makeOffers() } diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala index 956608d7c0cbe..c8edfed650421 100644 --- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala @@ -42,7 +42,8 @@ private[ui] case class ExecutorSummaryInfo( totalShuffleRead: Long, totalShuffleWrite: Long, maxMemory: Long, - executorLogs: Map[String, String]) + executorLogs: Map[String, String], + debugPortOpt: Option[Int]) private[ui] class ExecutorsPage( parent: ExecutorsTab, @@ -58,6 +59,7 @@ private[ui] class ExecutorsPage( val execInfo = for (statusId <- 0 until storageStatusList.size) yield getExecInfo(statusId) val execInfoSorted = execInfo.sortBy(_.id) val logsExist = execInfo.filter(_.executorLogs.nonEmpty).nonEmpty + val debugPortDefined = execInfo.filter(_.debugPortOpt.isDefined).nonEmpty val execTable = @@ -84,9 +86,10 @@ private[ui] class ExecutorsPage( {if (logsExist) else Seq.empty} {if (threadDumpEnabled) else Seq.empty} + {if (debugPortDefined) else Seq.empty} - {execInfoSorted.map(execRow(_, logsExist))} + {execInfoSorted.map(execRow(_, logsExist, debugPortDefined))}
LogsThread DumpDebug Port
@@ -111,10 +114,11 @@ private[ui] class ExecutorsPage( } /** Render an HTML row representing an executor */ - private def execRow(info: ExecutorSummaryInfo, logsExist: Boolean): Seq[Node] = { + private def execRow(info: ExecutorSummaryInfo, logsExist: Boolean, debugPortDefined: Boolean): Seq[Node] = { val maximumMemory = info.maxMemory val memoryUsed = info.memoryUsed val diskUsed = info.diskUsed + {info.id} {info.hostPort} @@ -167,6 +171,15 @@ private[ui] class ExecutorsPage( Seq.empty } } + { + if (debugPortDefined) { + + { + info.debugPortOpt.getOrElse("") + } + + } + } } @@ -188,6 +201,7 @@ private[ui] class ExecutorsPage( val totalShuffleRead = listener.executorToShuffleRead.getOrElse(execId, 0L) val totalShuffleWrite = listener.executorToShuffleWrite.getOrElse(execId, 0L) val executorLogs = listener.executorToLogUrls.getOrElse(execId, Map.empty) + val debugPortOpt = listener.executorToDebugPort.get(execId) new ExecutorSummaryInfo( execId, @@ -204,7 +218,8 @@ private[ui] class ExecutorsPage( totalShuffleRead, totalShuffleWrite, maxMem, - executorLogs + executorLogs, + debugPortOpt ) } } diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala index 69053fe44d7e4..86e2b786ed6b2 100644 --- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala +++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala @@ -54,12 +54,14 @@ class ExecutorsListener(storageStatusListener: StorageStatusListener) extends Sp val executorToShuffleRead = HashMap[String, Long]() val executorToShuffleWrite = HashMap[String, Long]() val executorToLogUrls = HashMap[String, Map[String, String]]() + val executorToDebugPort = HashMap[String, Int]() def storageStatusList: Seq[StorageStatus] = storageStatusListener.storageStatusList override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded): Unit = synchronized { val eid = executorAdded.executorId executorToLogUrls(eid) = executorAdded.executorInfo.logUrlMap + executorAdded.debugPortOpt.foreach(port => executorToDebugPort(eid) = port) } override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = synchronized { From 5c4982c9413c9ed6985e8e8f614730c5cd2c1023 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Tue, 7 Apr 2015 18:08:41 +0900 Subject: [PATCH 2/9] Improved code --- .../CoarseGrainedExecutorBackend.scala | 18 ++++++++++++++---- .../apache/spark/scheduler/SparkListener.scala | 6 +++++- .../cluster/SparkDeploySchedulerBackend.scala | 8 ++++++-- .../mesos/CoarseMesosSchedulerBackend.scala | 12 +++++++++--- .../cluster/mesos/MesosSchedulerBackend.scala | 10 ++++++++-- .../spark/deploy/yarn/ExecutorRunnable.scala | 4 ++++ 6 files changed, 46 insertions(+), 12 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala index 7f5e3a18834cd..1085cbfdcaeac 100644 --- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala +++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala @@ -20,6 +20,8 @@ package org.apache.spark.executor import java.net.URL import java.nio.ByteBuffer +import sun.misc.VMSupport + import scala.collection.mutable import scala.util.{Failure, Success} @@ -39,6 +41,7 @@ private[spark] class CoarseGrainedExecutorBackend( hostPort: String, cores: Int, userClassPath: Seq[URL], + debugPortOpt: Option[Int], env: SparkEnv) extends ThreadSafeRpcEndpoint with ExecutorBackend with Logging { @@ -49,9 +52,6 @@ private[spark] class CoarseGrainedExecutorBackend( override def onStart() { import scala.concurrent.ExecutionContext.Implicits.global - import sun.misc.VMSupport - val agentProps = VMSupport.getAgentProperties - val debugPortOpt = Option(agentProps.get("sun.jdwp.listenerAddress").asInstanceOf[String]).map(_.split(":")(1).toInt) logInfo("Connecting to driver: " + driverUrl) rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref => @@ -168,6 +168,16 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging { driverConf.set(key, value) } } + + debugPortOpt = + if (driverConf.getBoolean("spark.executor.jdwp.enabled", false)) { + val agentProps = VMSupport.getAgentProperties + Option(agentProps.get("sun.jdwp.listenerAddress").asInstanceOf[String]) + .map(_.split(":")(1).toInt) + } else { + None + } + val env = SparkEnv.createExecutorEnv( driverConf, executorId, hostname, port, cores, isLocal = false) @@ -178,7 +188,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging { // Start the CoarseGrainedExecutorBackend endpoint. val sparkHostPort = hostname + ":" + boundPort env.rpcEnv.setupEndpoint("Executor", new CoarseGrainedExecutorBackend( - env.rpcEnv, driverUrl, executorId, sparkHostPort, cores, userClassPath, env)) + env.rpcEnv, driverUrl, executorId, sparkHostPort, cores, userClassPath, debugPortOpt, env)) workerUrl.foreach { url => env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url)) } diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala index f00ee5bbb348d..0e606b849cd5a 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala @@ -91,7 +91,11 @@ case class SparkListenerBlockManagerRemoved(time: Long, blockManagerId: BlockMan case class SparkListenerUnpersistRDD(rddId: Int) extends SparkListenerEvent @DeveloperApi -case class SparkListenerExecutorAdded(time: Long, executorId: String, executorInfo: ExecutorInfo, debugPortOpt: Option[Int] = None) +case class SparkListenerExecutorAdded( + time: Long, + executorId: String, + executorInfo: ExecutorInfo, + debugPortOpt: Option[Int] = None) extends SparkListenerEvent @DeveloperApi diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala index 7eb3fdc19b5b8..ad7bf20a7b5cc 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala @@ -65,7 +65,11 @@ private[spark] class SparkDeploySchedulerBackend( .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil) val libraryPathEntries = sc.conf.getOption("spark.executor.extraLibraryPath") .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil) - + val jdwpOpts = if (sc.conf.getBoolean("spark.executor.jdwp.enabled", false)) { + Seq("-agentlib:jdwp=transport=dt_socket,suspend=n,server=y,address=0") + } else { + Seq.empty + } // When testing, expose the parent class path to the child. This is processed by // compute-classpath.{cmd,sh} and makes all needed jars available to child processes // when the assembly is built with the "*-provided" profiles enabled. @@ -78,7 +82,7 @@ private[spark] class SparkDeploySchedulerBackend( // Start executors with a few necessary configs for registering with the scheduler val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf) - val javaOpts = sparkJavaOpts ++ extraJavaOpts + val javaOpts = sparkJavaOpts ++ extraJavaOpts ++ jdwpOpts val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend", args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts) val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("") diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala index b037a4966ced0..2d0835c72d41d 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala @@ -22,7 +22,7 @@ import java.util.{List => JList} import java.util.Collections import scala.collection.JavaConversions._ -import scala.collection.mutable.{HashMap, HashSet} +import scala.collection.mutable.{ListBuffer, HashMap, HashSet} import org.apache.mesos.{Scheduler => MScheduler} import org.apache.mesos._ @@ -121,7 +121,13 @@ private[spark] class CoarseMesosSchedulerBackend( environment.addVariables( Environment.Variable.newBuilder().setName("SPARK_CLASSPATH").setValue(cp).build()) } - val extraJavaOpts = conf.get("spark.executor.extraJavaOptions", "") + val extraJavaOpts = new ListBuffer[String] + extraJavaOpts ++= conf.getOption("spark.executor.extraJavaOptions") + .map(Utils.splitCommandString).getOrElse(Seq.empty) + + if (conf.getBoolean("spark.executor.jdwp.enabled", false)) { + extraJavaOpts += "-agentlib:jdwp=transport=dt_socket,suspend=n,server=y,address=0" + } // Set the environment variable through a command prefix // to append to the existing value of the variable @@ -132,7 +138,7 @@ private[spark] class CoarseMesosSchedulerBackend( environment.addVariables( Environment.Variable.newBuilder() .setName("SPARK_EXECUTOR_OPTS") - .setValue(extraJavaOpts) + .setValue(extraJavaOpts.toString) .build()) sc.executorEnvs.foreach { case (key, value) => diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala index b381436839227..3d975389d5d40 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala @@ -22,7 +22,7 @@ import java.util.{ArrayList => JArrayList, List => JList} import java.util.Collections import scala.collection.JavaConversions._ -import scala.collection.mutable.{HashMap, HashSet} +import scala.collection.mutable.{ListBuffer, HashMap, HashSet} import org.apache.mesos.protobuf.ByteString import org.apache.mesos.{Scheduler => MScheduler} @@ -104,7 +104,13 @@ private[spark] class MesosSchedulerBackend( environment.addVariables( Environment.Variable.newBuilder().setName("SPARK_CLASSPATH").setValue(cp).build()) } - val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions").getOrElse("") + val extraJavaOpts = new ListBuffer[String] + extraJavaOpts ++= sc.conf.getOption("spark.executor.extraJavaOptions") + .map(Utils.splitCommandString).getOrElse(Seq.empty) + + if (sc.conf.getBoolean("spark.executor.jdwp.enabled", false)) { + extraJavaOpts += + } val prefixEnv = sc.conf.getOption("spark.executor.extraLibraryPath").map { p => Utils.libraryPathEnvPrefix(Seq(p)) diff --git a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala index 1ce10d906ab23..3fc3cd90f5e5d 100644 --- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala +++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala @@ -192,6 +192,10 @@ class ExecutorRunnable( // For log4j configuration to reference javaOpts += ("-Dspark.yarn.app.container.log.dir=" + ApplicationConstants.LOG_DIR_EXPANSION_VAR) + if (sys.props.get("spark.executor.jdwp.enabled").map(_.toBoolean).getOrElse(false)) { + javaOpts += "-agentlib:jdwp=transport=dt_socket,suspend=n,server=y,address=0" + } + val userClassPath = Client.getUserClasspath(sparkConf).flatMap { uri => val absPath = if (new File(uri.getPath()).isAbsolute()) { From 602c335f7c382b1db78549386216adf2307ee040 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Wed, 8 Apr 2015 17:27:49 +0900 Subject: [PATCH 3/9] Added a test case to YarnClusterSuite for the debug port feature --- .../CoarseGrainedExecutorBackend.scala | 3 +- .../spark/scheduler/SparkListener.scala | 6 +- .../CoarseGrainedSchedulerBackend.scala | 5 +- .../scheduler/cluster/ExecutorData.scala | 5 +- .../scheduler/cluster/ExecutorInfo.scala | 8 +- .../cluster/mesos/MesosSchedulerBackend.scala | 13 +--- .../apache/spark/ui/exec/ExecutorsTab.scala | 2 +- .../org/apache/spark/util/JsonProtocol.scala | 8 +- .../ExecutorAllocationManagerSuite.scala | 12 +-- .../mesos/MesosSchedulerBackendSuite.scala | 4 +- .../apache/spark/util/JsonProtocolSuite.scala | 9 ++- .../spark/deploy/yarn/YarnClusterSuite.scala | 74 +++++++++++++++++-- 12 files changed, 105 insertions(+), 44 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala index 1085cbfdcaeac..060a42da12bcd 100644 --- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala +++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala @@ -52,7 +52,6 @@ private[spark] class CoarseGrainedExecutorBackend( override def onStart() { import scala.concurrent.ExecutionContext.Implicits.global - logInfo("Connecting to driver: " + driverUrl) rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref => driver = Some(ref) @@ -169,7 +168,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging { } } - debugPortOpt = + val debugPortOpt = if (driverConf.getBoolean("spark.executor.jdwp.enabled", false)) { val agentProps = VMSupport.getAgentProperties Option(agentProps.get("sun.jdwp.listenerAddress").asInstanceOf[String]) diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala index 0e606b849cd5a..b711ff209af94 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala @@ -91,11 +91,7 @@ case class SparkListenerBlockManagerRemoved(time: Long, blockManagerId: BlockMan case class SparkListenerUnpersistRDD(rddId: Int) extends SparkListenerEvent @DeveloperApi -case class SparkListenerExecutorAdded( - time: Long, - executorId: String, - executorInfo: ExecutorInfo, - debugPortOpt: Option[Int] = None) +case class SparkListenerExecutorAdded(time: Long, executorId: String, executorInfo: ExecutorInfo) extends SparkListenerEvent @DeveloperApi diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index 33fe3183af786..80439e0edef8e 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -126,7 +126,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp totalCoreCount.addAndGet(cores) totalRegisteredExecutors.addAndGet(1) val (host, _) = Utils.parseHostPort(hostPort) - val data = new ExecutorData(executorRef, executorRef.address, host, cores, cores, logUrls) + val data = new ExecutorData( + executorRef, executorRef.address, host, cores, cores, logUrls, debugPortOpt) // This must be synchronized because variables mutated // in this block are read when requesting executors CoarseGrainedSchedulerBackend.this.synchronized { @@ -137,7 +138,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp } } listenerBus.post( - SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data, debugPortOpt)) + SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data)) makeOffers() } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala index 26e72c0bff38d..b2f8dc0ced44c 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala @@ -34,5 +34,6 @@ private[cluster] class ExecutorData( override val executorHost: String, var freeCores: Int, override val totalCores: Int, - override val logUrlMap: Map[String, String] -) extends ExecutorInfo(executorHost, totalCores, logUrlMap) + override val logUrlMap: Map[String, String], + override val debugPortOpt: Option[Int] +) extends ExecutorInfo(executorHost, totalCores, logUrlMap, debugPortOpt) diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorInfo.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorInfo.scala index 7f218566146a1..6abb168400d60 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorInfo.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorInfo.scala @@ -26,7 +26,8 @@ import org.apache.spark.annotation.DeveloperApi class ExecutorInfo( val executorHost: String, val totalCores: Int, - val logUrlMap: Map[String, String]) { + val logUrlMap: Map[String, String], + val debugPortOpt: Option[Int]) { def canEqual(other: Any): Boolean = other.isInstanceOf[ExecutorInfo] @@ -35,12 +36,13 @@ class ExecutorInfo( (that canEqual this) && executorHost == that.executorHost && totalCores == that.totalCores && - logUrlMap == that.logUrlMap + logUrlMap == that.logUrlMap && + debugPortOpt == that.debugPortOpt case _ => false } override def hashCode(): Int = { - val state = Seq(executorHost, totalCores, logUrlMap) + val state = Seq(executorHost, totalCores, logUrlMap, debugPortOpt) state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b) } } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala index 3d975389d5d40..19a26ce0416b2 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala @@ -22,7 +22,7 @@ import java.util.{ArrayList => JArrayList, List => JList} import java.util.Collections import scala.collection.JavaConversions._ -import scala.collection.mutable.{ListBuffer, HashMap, HashSet} +import scala.collection.mutable.{HashMap, HashSet} import org.apache.mesos.protobuf.ByteString import org.apache.mesos.{Scheduler => MScheduler} @@ -104,13 +104,7 @@ private[spark] class MesosSchedulerBackend( environment.addVariables( Environment.Variable.newBuilder().setName("SPARK_CLASSPATH").setValue(cp).build()) } - val extraJavaOpts = new ListBuffer[String] - extraJavaOpts ++= sc.conf.getOption("spark.executor.extraJavaOptions") - .map(Utils.splitCommandString).getOrElse(Seq.empty) - - if (sc.conf.getBoolean("spark.executor.jdwp.enabled", false)) { - extraJavaOpts += - } + val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions").getOrElse("") val prefixEnv = sc.conf.getOption("spark.executor.extraLibraryPath").map { p => Utils.libraryPathEnvPrefix(Seq(p)) @@ -277,7 +271,8 @@ private[spark] class MesosSchedulerBackend( slaveIdToWorkerOffer.get(slaveId).foreach(o => listenerBus.post(SparkListenerExecutorAdded(System.currentTimeMillis(), slaveId, // TODO: Add support for log urls for Mesos - new ExecutorInfo(o.host, o.cores, Map.empty))) + // TODO: Add support for debug port for Mesos + new ExecutorInfo(o.host, o.cores, Map.empty, None))) ) d.launchTasks(Collections.singleton(slaveIdToOffer(slaveId).getId), tasks, filters) } diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala index 86e2b786ed6b2..4fcabd83a75c2 100644 --- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala +++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala @@ -61,7 +61,7 @@ class ExecutorsListener(storageStatusListener: StorageStatusListener) extends Sp override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded): Unit = synchronized { val eid = executorAdded.executorId executorToLogUrls(eid) = executorAdded.executorInfo.logUrlMap - executorAdded.debugPortOpt.foreach(port => executorToDebugPort(eid) = port) + executorAdded.executorInfo.debugPortOpt.foreach(port => executorToDebugPort(eid) = port) } override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = synchronized { diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala index 474f79fb756f6..330a3b67c3c90 100644 --- a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala +++ b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala @@ -396,7 +396,8 @@ private[spark] object JsonProtocol { def executorInfoToJson(executorInfo: ExecutorInfo): JValue = { ("Host" -> executorInfo.executorHost) ~ ("Total Cores" -> executorInfo.totalCores) ~ - ("Log Urls" -> mapToJson(executorInfo.logUrlMap)) + ("Log Urls" -> mapToJson(executorInfo.logUrlMap)) ~ + ("Debug Port" -> executorInfo.debugPortOpt.map(port => JInt(port)).getOrElse(JNothing)) } /** ------------------------------ * @@ -819,7 +820,10 @@ private[spark] object JsonProtocol { val executorHost = (json \ "Host").extract[String] val totalCores = (json \ "Total Cores").extract[Int] val logUrls = mapFromJson(json \ "Log Urls").toMap - new ExecutorInfo(executorHost, totalCores, logUrls) + val debugPortOpt = Utils.jsonOption(json \ "Debug Port").map { value => + value.extract[Int] + } + new ExecutorInfo(executorHost, totalCores, logUrls, debugPortOpt) } /** -------------------------------- * diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 3ded1e4af8742..0f878ef2bab4c 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -157,7 +157,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext wit // Verify that running a task reduces the cap sc.listenerBus.postToAll(SparkListenerStageSubmitted(createStageInfo(1, 3))) sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty))) + 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, None))) sc.listenerBus.postToAll(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-1"))) assert(numExecutorsPending(manager) === 4) assert(addExecutors(manager) === 1) @@ -616,13 +616,13 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext wit // New executors have registered sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty))) + 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, None))) assert(executorIds(manager).size === 1) assert(executorIds(manager).contains("executor-1")) assert(removeTimes(manager).size === 1) assert(removeTimes(manager).contains("executor-1")) sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-2", new ExecutorInfo("host2", 1, Map.empty))) + 0L, "executor-2", new ExecutorInfo("host2", 1, Map.empty, None))) assert(executorIds(manager).size === 2) assert(executorIds(manager).contains("executor-2")) assert(removeTimes(manager).size === 2) @@ -649,7 +649,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext wit sc.listenerBus.postToAll(SparkListenerTaskStart(0, 0, createTaskInfo(0, 0, "executor-1"))) sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty))) + 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, None))) assert(executorIds(manager).size === 1) assert(executorIds(manager).contains("executor-1")) assert(removeTimes(manager).size === 0) @@ -661,7 +661,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext wit assert(executorIds(manager).isEmpty) assert(removeTimes(manager).isEmpty) sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty))) + 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, None))) sc.listenerBus.postToAll(SparkListenerTaskStart(0, 0, createTaskInfo(0, 0, "executor-1"))) assert(executorIds(manager).size === 1) @@ -669,7 +669,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext wit assert(removeTimes(manager).size === 0) sc.listenerBus.postToAll(SparkListenerExecutorAdded( - 0L, "executor-2", new ExecutorInfo("host1", 1, Map.empty))) + 0L, "executor-2", new ExecutorInfo("host1", 1, Map.empty, None))) assert(executorIds(manager).size === 2) assert(executorIds(manager).contains("executor-2")) assert(removeTimes(manager).size === 1) diff --git a/core/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendSuite.scala index f1a4380d349b3..dce64cb410cd5 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendSuite.scala @@ -47,7 +47,7 @@ class MesosSchedulerBackendSuite extends FunSuite with LocalSparkContext with Mo val listenerBus = mock[LiveListenerBus] listenerBus.post( - SparkListenerExecutorAdded(anyLong, "s1", new ExecutorInfo("host1", 2, Map.empty))) + SparkListenerExecutorAdded(anyLong, "s1", new ExecutorInfo("host1", 2, Map.empty, None))) val sc = mock[SparkContext] when(sc.getSparkHome()).thenReturn(Option("/spark-home")) @@ -91,7 +91,7 @@ class MesosSchedulerBackendSuite extends FunSuite with LocalSparkContext with Mo val listenerBus = mock[LiveListenerBus] listenerBus.post( - SparkListenerExecutorAdded(anyLong, "s1", new ExecutorInfo("host1", 2, Map.empty))) + SparkListenerExecutorAdded(anyLong, "s1", new ExecutorInfo("host1", 2, Map.empty, None))) val sc = mock[SparkContext] when(sc.executorMemory).thenReturn(100) diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala index a2be724254d7c..2b575efdc29ef 100644 --- a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala @@ -77,8 +77,9 @@ class JsonProtocolSuite extends FunSuite { val applicationStart = SparkListenerApplicationStart("The winner of all", None, 42L, "Garfield") val applicationEnd = SparkListenerApplicationEnd(42L) val logUrlMap = Map("stderr" -> "mystderr", "stdout" -> "mystdout").toMap + val debugPortOpt = Some(9876) val executorAdded = SparkListenerExecutorAdded(executorAddedTime, "exec1", - new ExecutorInfo("Hostee.awesome.com", 11, logUrlMap)) + new ExecutorInfo("Hostee.awesome.com", 11, logUrlMap, debugPortOpt)) val executorRemoved = SparkListenerExecutorRemoved(executorRemovedTime, "exec2", "test reason") testEvent(stageSubmitted, stageSubmittedJsonString) @@ -102,13 +103,14 @@ class JsonProtocolSuite extends FunSuite { test("Dependent Classes") { val logUrlMap = Map("stderr" -> "mystderr", "stdout" -> "mystdout").toMap + val debugPortOpt = Some(9876) testRDDInfo(makeRddInfo(2, 3, 4, 5L, 6L)) testStageInfo(makeStageInfo(10, 20, 30, 40L, 50L)) testTaskInfo(makeTaskInfo(999L, 888, 55, 777L, false)) testTaskMetrics(makeTaskMetrics( 33333L, 44444L, 55555L, 66666L, 7, 8, hasHadoopInput = false, hasOutput = false)) testBlockManagerId(BlockManagerId("Hong", "Kong", 500)) - testExecutorInfo(new ExecutorInfo("host", 43, logUrlMap)) + testExecutorInfo(new ExecutorInfo("host", 43, logUrlMap, debugPortOpt)) // StorageLevel testStorageLevel(StorageLevel.NONE) @@ -1522,7 +1524,8 @@ class JsonProtocolSuite extends FunSuite { | "Log Urls" : { | "stderr" : "mystderr", | "stdout" : "mystdout" - | } + | }, + | "Debug Port" : 9876 | } |} """ diff --git a/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala b/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala index c06c0105670c0..099227aee1aa5 100644 --- a/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala +++ b/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala @@ -33,7 +33,7 @@ import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers} import org.apache.spark.{Logging, SparkConf, SparkContext, SparkException, TestUtils} import org.apache.spark.scheduler.cluster.ExecutorInfo -import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorAdded} +import org.apache.spark.scheduler.{SparkListenerJobStart, SparkListener, SparkListenerExecutorAdded} import org.apache.spark.util.Utils /** @@ -173,6 +173,14 @@ class YarnClusterSuite extends FunSuite with BeforeAndAfterAll with Matchers wit testUseClassPathFirst(false) } + test("executors' debug ports are notified in client mode") { + testExecutorDebugPortNotified(true) + } + + test("executors' debug ports are notified in cluster mode") { + testExecutorDebugPortNotified(false) + } + private def testBasicYarnApp(clientMode: Boolean): Unit = { var result = File.createTempFile("result", null, tempDir) runSpark(clientMode, mainClassName(YarnClusterDriver.getClass), @@ -197,6 +205,14 @@ class YarnClusterSuite extends FunSuite with BeforeAndAfterAll with Matchers wit checkResult(executorResult, "OVERRIDDEN") } + private def testExecutorDebugPortNotified(clientMode: Boolean): Unit = { + val result = File.createTempFile("result", null, tempDir) + runSpark(clientMode, mainClassName(YarnExecutorDebugPortTest.getClass), + appArgs = Seq(result.getAbsolutePath()), + extraConf = Map("spark.executor.jdwp.enabled" -> "true")) + checkResult(result) + } + private def runSpark( clientMode: Boolean, klass: String, @@ -282,10 +298,10 @@ class YarnClusterSuite extends FunSuite with BeforeAndAfterAll with Matchers wit } -private class SaveExecutorInfo extends SparkListener { +private[spark] class SaveExecutorInfo extends SparkListener { val addedExecutorInfos = mutable.Map[String, ExecutorInfo]() - override def onExecutorAdded(executor : SparkListenerExecutorAdded) { + override def onExecutorAdded(executor: SparkListenerExecutorAdded) { addedExecutorInfos(executor.executorId) = executor.executorInfo } } @@ -293,7 +309,6 @@ private class SaveExecutorInfo extends SparkListener { private object YarnClusterDriver extends Logging with Matchers { val WAIT_TIMEOUT_MILLIS = 10000 - var listener: SaveExecutorInfo = null def main(args: Array[String]): Unit = { if (args.length != 1) { @@ -306,10 +321,9 @@ private object YarnClusterDriver extends Logging with Matchers { System.exit(1) } - listener = new SaveExecutorInfo val sc = new SparkContext(new SparkConf() + .set("spark.extraListeners", "org.apache.spark.deploy.yarn.SaveExecutorInfo") .setAppName("yarn \"test app\" 'with quotes' and \\back\\slashes and $dollarSigns")) - sc.addSparkListener(listener) val status = new File(args(0)) var result = "failure" try { @@ -323,7 +337,13 @@ private object YarnClusterDriver extends Logging with Matchers { } // verify log urls are present - listener.addedExecutorInfos.values.foreach { info => + val listener = sc.listenerBus.listeners.filter { + case _: SaveExecutorInfo => true + case _ => false + }(0).asInstanceOf[SaveExecutorInfo] + val executorInfos = listener.addedExecutorInfos.values + assert(executorInfos.nonEmpty) + executorInfos.foreach { info => assert(info.logUrlMap.nonEmpty) } } @@ -365,3 +385,43 @@ private object YarnClasspathTest { } } + +private object YarnExecutorDebugPortTest { + + val WAIT_TIMEOUT_MILLIS = 10000 + + def main(args: Array[String]): Unit = { + if (args.length != 1) { + System.err.println( + s""" + |Invalid command line: ${args.mkString(" ")} + | + |Usage: YarnExecutorDebugPortTest [result file] + """.stripMargin) + System.exit(1) + } + + val resultPath = args(0) + val sc = new SparkContext( + new SparkConf().set("spark.extraListeners", "org.apache.spark.deploy.yarn.SaveExecutorInfo")) + try { + sc.parallelize(Seq(1)).collect + assert(sc.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)) + } finally { + sc.stop() + var result = "failure" + val listener = sc.listenerBus.listeners.filter { + case _: SaveExecutorInfo => true + case _ => false + }(0).asInstanceOf[SaveExecutorInfo] + + val executorInfos = listener.addedExecutorInfos.values + if (executorInfos.nonEmpty && executorInfos.forall(info => info.debugPortOpt.nonEmpty)) { + result = "success" + } + Files.write(result, new File(resultPath), UTF_8) + } + } + +} + From 4c16e9cb4f0d84927494279650b5b75295f61478 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Wed, 8 Apr 2015 18:05:54 +0900 Subject: [PATCH 4/9] Temporary commit --- .../org/apache/spark/deploy/yarn/YarnClusterSuite.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala b/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala index 099227aee1aa5..7efe2092d2095 100644 --- a/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala +++ b/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala @@ -337,10 +337,12 @@ private object YarnClusterDriver extends Logging with Matchers { } // verify log urls are present - val listener = sc.listenerBus.listeners.filter { + val listenerOpt = sc.listenerBus.listeners.find { case _: SaveExecutorInfo => true case _ => false - }(0).asInstanceOf[SaveExecutorInfo] + }.map(_.asInstanceOf[SaveExecutorInfo]) + assert(listenerOpt.isDefined) + val listener = listenerOpt.get val executorInfos = listener.addedExecutorInfos.values assert(executorInfos.nonEmpty) executorInfos.foreach { info => From 654500d89ffdb94f9e2c83bc18c7e92793fb524b Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Thu, 9 Apr 2015 16:11:11 +0900 Subject: [PATCH 5/9] Added test cases to UISeleniumSuite --- .../org/apache/spark/ui/UISeleniumSuite.scala | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala index 0d155982a8c54..8d8033e6b66b6 100644 --- a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala +++ b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala @@ -356,4 +356,33 @@ class UISeleniumSuite extends FunSuite with WebBrowser with Matchers with Before } } } + + test("debug port numbers for each executor should be appeared when spark.executor.jdwp.enabled is true") { + def getSparkContext(jdwpEnabled: Boolean): SparkContext = { + val conf = new SparkConf() + .setMaster("local-cluster[2,1,512]") + .setAppName("test") + .set("spark.ui.enabled", "true") + .set("spark.executor.jdwp.enabled", jdwpEnabled.toString) + new SparkContext(conf) + } + + withSpark(getSparkContext(jdwpEnabled = true)) { sc => + sc.parallelize(1 to 10, 10).collect + eventually(timeout(5 seconds), interval(50 milliseconds)) { + go to (sc.ui.get.appUIAddress.stripSuffix("/") + "/executors") + val tableHeaders = findAll(cssSelector("th")).map(_.text).toSeq + tableHeaders should contain("Debug Port") + } + } + + withSpark(getSparkContext(jdwpEnabled = false)) { sc => + sc.parallelize(1 to 10, 10).collect + eventually(timeout(5 seconds), interval(50 milliseconds)) { + go to (sc.ui.get.appUIAddress.stripSuffix("/") + "/executors") + val tableHeaders = findAll(cssSelector("th")).map(_.text).toSeq + tableHeaders should not contain("Debug Port") + } + } + } } From dbf69f39b0443dc7eff49cf985b2ac9cc658c954 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Thu, 9 Apr 2015 16:11:36 +0900 Subject: [PATCH 6/9] Added ExecutorDebugPortsuite --- .../spark/deploy/ExecutorDebugPortSuite.scala | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala diff --git a/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala b/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala new file mode 100644 index 0000000000000..d42575c80f9ee --- /dev/null +++ b/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala @@ -0,0 +1,54 @@ +/* + * 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.io.Source + +import org.scalatest.FunSuite + +import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext} + +class ExecutorDebugPortSuite extends FunSuite with LocalSparkContext { + + /** Length of time to wait while draining listener events. */ + private val WAIT_TIMEOUT_MILLIS = 10000 + + test("verify that correct log urls get propagated from workers") { + try { + val conf = new SparkConf() + .set("spark.executor.jdwp.enabled", "true") + .set("spark.ui.enabled", "true") + .set("spark.ui.port", "0") + sc = new SparkContext("local-cluster[2,1,512]", "test", conf) + sc.parallelize(1 to 10, 10).collect + val url = sc.ui.get.appUIAddress.stripSuffix("/") + "/executors" + val html = Source.fromURL(url).mkString + println(html) + /* listener.addedExecutorInfos.values.foreach { info => + assert(info.logUrlMap.nonEmpty) + // Browse to each URL to check that it's valid + info.logUrlMap.foreach { case (logType, logUrl) => + val html = Source.fromURL(logUrl).mkString + assert(html.contains(s"$logType log page")) + } + }*/ + } finally { + sc.stop() + } + } +} From 41aa986296dc9c0745d1700ae89963e633302694 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Thu, 9 Apr 2015 16:52:00 +0900 Subject: [PATCH 7/9] Added description to configuration.md --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 7fe11475212b3..cf656f7e54ce7 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -253,6 +253,14 @@ Apart from these, the following properties are also available, and may be useful Set a special library path to use when launching executor JVM's. + + spark.executor.jdwp.enabled + false + + When enabling this option, unique JDWP ports will open for each Executor process and + which port is opened for each executor is displayed on Web UI. + + spark.executor.logs.rolling.maxRetainedFiles (none) From 79ee2f9e77e0dd7ec7da1c692c2d283f5794e17f Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Thu, 9 Apr 2015 16:55:30 +0900 Subject: [PATCH 8/9] Fixed style --- .../main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala index c8edfed650421..dcf0da1305e7b 100644 --- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala @@ -114,7 +114,10 @@ private[ui] class ExecutorsPage( } /** Render an HTML row representing an executor */ - private def execRow(info: ExecutorSummaryInfo, logsExist: Boolean, debugPortDefined: Boolean): Seq[Node] = { + private def execRow( + info: ExecutorSummaryInfo, + logsExist: Boolean, + debugPortDefined: Boolean): Seq[Node] = { val maximumMemory = info.maxMemory val memoryUsed = info.memoryUsed val diskUsed = info.diskUsed From e3ba120e661735909c3ded2130dcb85d7598cf75 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Thu, 9 Apr 2015 17:07:09 +0900 Subject: [PATCH 9/9] Modified ExexutorDebugPortSuite.scala --- .../spark/deploy/ExecutorDebugPortSuite.scala | 28 ++++++++++++------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala b/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala index d42575c80f9ee..b38d591389037 100644 --- a/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/ExecutorDebugPortSuite.scala @@ -28,7 +28,7 @@ class ExecutorDebugPortSuite extends FunSuite with LocalSparkContext { /** Length of time to wait while draining listener events. */ private val WAIT_TIMEOUT_MILLIS = 10000 - test("verify that correct log urls get propagated from workers") { + test("verify that debug port is opened when spark.executor.jdwp.enabled is true") { try { val conf = new SparkConf() .set("spark.executor.jdwp.enabled", "true") @@ -38,15 +38,23 @@ class ExecutorDebugPortSuite extends FunSuite with LocalSparkContext { sc.parallelize(1 to 10, 10).collect val url = sc.ui.get.appUIAddress.stripSuffix("/") + "/executors" val html = Source.fromURL(url).mkString - println(html) - /* listener.addedExecutorInfos.values.foreach { info => - assert(info.logUrlMap.nonEmpty) - // Browse to each URL to check that it's valid - info.logUrlMap.foreach { case (logType, logUrl) => - val html = Source.fromURL(logUrl).mkString - assert(html.contains(s"$logType log page")) - } - }*/ + assert(html.contains("Debug Port")) + } finally { + sc.stop() + } + } + + test("verify that debug port is not opened when spark.executor.jdwp.enabled is false") { + try { + val conf = new SparkConf() + .set("spark.executor.jdwp.enabled", "false") + .set("spark.ui.enabled", "true") + .set("spark.ui.port", "0") + sc = new SparkContext("local-cluster[2,1,512]", "test", conf) + sc.parallelize(1 to 10, 10).collect + val url = sc.ui.get.appUIAddress.stripSuffix("/") + "/executors" + val html = Source.fromURL(url).mkString + assert(!html.contains("Debug Port")) } finally { sc.stop() }