+
+ {backButton}
+ {range}
+ {nextButton}
+ +
+
+
+
+ UIUtils.basicSparkPage(content, logType + " log page for " + appId)
+ }
+
+ /** Determine the byte range for a log or log page. */
+ private def getByteRange(path: String, offset: Option[Long], byteLength: Int): (Long, Long) = {
+ val defaultBytes = 100 * 1024
+ val maxBytes = 1024 * 1024
+ val file = new File(path)
+ val logLength = file.length()
+ val getOffset = offset.getOrElse(logLength - defaultBytes)
+ val startByte =
+ if (getOffset < 0) 0L
+ else if (getOffset > logLength) logLength
+ else getOffset
+ val logPageLength = math.min(byteLength, maxBytes)
+ val endByte = math.min(startByte + logPageLength, logLength)
+ (startByte, endByte)
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerWebUI.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerWebUI.scala
index de76a5d5eb7bc..490a383be42e1 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerWebUI.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerWebUI.scala
@@ -20,11 +20,9 @@ package org.apache.spark.deploy.worker.ui
import java.io.File
import javax.servlet.http.HttpServletRequest
-import org.eclipse.jetty.servlet.ServletContextHandler
-
import org.apache.spark.Logging
import org.apache.spark.deploy.worker.Worker
-import org.apache.spark.ui.{JettyUtils, ServerInfo, SparkUI, UIUtils}
+import org.apache.spark.ui.{SparkUI, WebUI}
import org.apache.spark.ui.JettyUtils._
import org.apache.spark.util.{AkkaUtils, Utils}
@@ -33,164 +31,35 @@ import org.apache.spark.util.{AkkaUtils, Utils}
*/
private[spark]
class WorkerWebUI(val worker: Worker, val workDir: File, requestedPort: Option[Int] = None)
- extends Logging {
-
- val timeout = AkkaUtils.askTimeout(worker.conf)
+ extends WebUI(worker.securityMgr) with Logging {
private val host = Utils.localHostName()
private val port = requestedPort.getOrElse(
worker.conf.get("worker.ui.port", WorkerWebUI.DEFAULT_PORT).toInt)
- private val indexPage = new IndexPage(this)
- private var serverInfo: Option[ServerInfo] = None
+ val timeout = AkkaUtils.askTimeout(worker.conf)
- private val handlers: Seq[ServletContextHandler] = {
- worker.metricsSystem.getServletHandlers ++
- Seq[ServletContextHandler](
- createStaticHandler(WorkerWebUI.STATIC_RESOURCE_BASE, "/static"),
- createServletHandler("/log",
- (request: HttpServletRequest) => log(request), worker.securityMgr),
- createServletHandler("/logPage",
- (request: HttpServletRequest) => logPage(request), worker.securityMgr),
- createServletHandler("/json",
- (request: HttpServletRequest) => indexPage.renderJson(request), worker.securityMgr),
- createServletHandler("/",
- (request: HttpServletRequest) => indexPage.render(request), worker.securityMgr)
- )
+ /** Initialize all components of the server. */
+ def start() {
+ val logPage = new LogPage(this)
+ attachPage(logPage)
+ attachPage(new IndexPage(this))
+ attachHandler(createStaticHandler(WorkerWebUI.STATIC_RESOURCE_BASE, "/static"))
+ attachHandler(createServletHandler("/log",
+ (request: HttpServletRequest) => logPage.renderLog(request), worker.securityMgr))
+ worker.metricsSystem.getServletHandlers.foreach(attachHandler)
}
+ /** Bind to the HTTP server behind this web interface. */
def bind() {
try {
- serverInfo = Some(JettyUtils.startJettyServer(host, port, handlers, worker.conf))
+ serverInfo = Some(startJettyServer(host, port, handlers, worker.conf))
logInfo("Started Worker web UI at http://%s:%d".format(host, boundPort))
} catch {
case e: Exception =>
- logError("Failed to create Worker JettyUtils", e)
+ logError("Failed to create Worker web UI", e)
System.exit(1)
}
}
-
- def boundPort: Int = serverInfo.map(_.boundPort).getOrElse(-1)
-
- private def log(request: HttpServletRequest): String = {
- val defaultBytes = 100 * 1024
-
- val appId = Option(request.getParameter("appId"))
- val executorId = Option(request.getParameter("executorId"))
- val driverId = Option(request.getParameter("driverId"))
- val logType = request.getParameter("logType")
- val offset = Option(request.getParameter("offset")).map(_.toLong)
- val byteLength = Option(request.getParameter("byteLength")).map(_.toInt).getOrElse(defaultBytes)
-
- val path = (appId, executorId, driverId) match {
- case (Some(a), Some(e), None) =>
- s"${workDir.getPath}/$appId/$executorId/$logType"
- case (None, None, Some(d)) =>
- s"${workDir.getPath}/$driverId/$logType"
- case _ =>
- throw new Exception("Request must specify either application or driver identifiers")
- }
-
- val (startByte, endByte) = getByteRange(path, offset, byteLength)
- val file = new File(path)
- val logLength = file.length
-
- val pre = s"==== Bytes $startByte-$endByte of $logLength of $path ====\n"
- pre + Utils.offsetBytes(path, startByte, endByte)
- }
-
- private def logPage(request: HttpServletRequest): Seq[scala.xml.Node] = {
- val defaultBytes = 100 * 1024
- val appId = Option(request.getParameter("appId"))
- val executorId = Option(request.getParameter("executorId"))
- val driverId = Option(request.getParameter("driverId"))
- val logType = request.getParameter("logType")
- val offset = Option(request.getParameter("offset")).map(_.toLong)
- val byteLength = Option(request.getParameter("byteLength")).map(_.toInt).getOrElse(defaultBytes)
-
- val (path, params) = (appId, executorId, driverId) match {
- case (Some(a), Some(e), None) =>
- (s"${workDir.getPath}/$a/$e/$logType", s"appId=$a&executorId=$e")
- case (None, None, Some(d)) =>
- (s"${workDir.getPath}/$d/$logType", s"driverId=$d")
- case _ =>
- throw new Exception("Request must specify either application or driver identifiers")
- }
-
- val (startByte, endByte) = getByteRange(path, offset, byteLength)
- val file = new File(path)
- val logLength = file.length
- val logText = {logText}
+
-
- {backButton}
- {range}
- {nextButton}
- -
-
-
-
- UIUtils.basicSparkPage(content, logType + " log page for " + appId)
- }
-
- /** Determine the byte range for a log or log page. */
- private def getByteRange(path: String, offset: Option[Long], byteLength: Int): (Long, Long) = {
- val defaultBytes = 100 * 1024
- val maxBytes = 1024 * 1024
- val file = new File(path)
- val logLength = file.length()
- val getOffset = offset.getOrElse(logLength - defaultBytes)
- val startByte =
- if (getOffset < 0) 0L
- else if (getOffset > logLength) logLength
- else getOffset
- val logPageLength = math.min(byteLength, maxBytes)
- val endByte = math.min(startByte + logPageLength, logLength)
- (startByte, endByte)
- }
-
- def stop() {
- assert(serverInfo.isDefined, "Attempted to stop a Worker UI that was not bound to a server!")
- serverInfo.get.server.stop()
- }
}
private[spark] object WorkerWebUI {
diff --git a/core/src/main/scala/org/apache/spark/ui/FooTab.scala b/core/src/main/scala/org/apache/spark/ui/FooTab.scala
new file mode 100644
index 0000000000000..620fe8001a85c
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/ui/FooTab.scala
@@ -0,0 +1,105 @@
+/*
+ * 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.ui
+
+import javax.servlet.http.HttpServletRequest
+
+import scala.collection.mutable
+import scala.xml.Node
+
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.SparkContext._
+import org.apache.spark.scheduler.{SparkListener, SparkListenerJobEnd}
+
+/*
+ * This is an example of how to extend the SparkUI by adding new tabs to it. It is intended
+ * only as a demonstration and should be removed before merging into master!
+ *
+ * bin/spark-class org.apache.spark.ui.FooTab
+ */
+
+/** A tab that displays basic information about jobs seen so far. */
+private[spark] class FooTab(parent: SparkUI) extends UITab("foo") {
+ val appName = parent.appName
+ val basePath = parent.basePath
+
+ def start() {
+ listener = Some(new FooListener)
+ attachPage(new IndexPage(this))
+ }
+
+ def fooListener: FooListener = {
+ assert(listener.isDefined, "ExecutorsTab has not started yet!")
+ listener.get.asInstanceOf[FooListener]
+ }
+
+ def headerTabs: Seq[UITab] = parent.getTabs
+}
+
+/** A foo page. Enough said. */
+private[spark] class IndexPage(parent: FooTab) extends UIPage("") {
+ private val appName = parent.appName
+ private val basePath = parent.basePath
+ private val listener = parent.fooListener
+
+ override def render(request: HttpServletRequest): Seq[Node] = {
+ val results = listener.jobResultMap.toSeq.sortBy { case (k, _) => k }
+ val content =
+ {logText}
-
+
+ UIUtils.headerSparkPage(content, basePath, appName, "Foo", parent.headerTabs, parent)
+ }
+}
+
+/** A listener that maintains a mapping between job IDs and job results. */
+private[spark] class FooListener extends SparkListener {
+ val jobResultMap = mutable.Map[Int, String]()
+
+ override def onJobEnd(end: SparkListenerJobEnd) {
+ jobResultMap(end.jobId) = end.jobResult.toString
+ }
+}
+
+
+/**
+ * Start a SparkContext and a SparkUI with a FooTab attached.
+ */
+private[spark] object FooTab {
+ def main(args: Array[String]) {
+ val sc = new SparkContext("local", "Foo Tab", new SparkConf)
+ val fooTab = new FooTab(sc.ui)
+ sc.ui.attachTab(fooTab)
+
+ // Run a few jobs
+ sc.parallelize(1 to 1000).count()
+ sc.parallelize(1 to 2000).persist().count()
+ sc.parallelize(1 to 3000).map(i => (i/2, i)).groupByKey().count()
+ sc.parallelize(1 to 4000).map(i => (i/2, i)).groupByKey().persist().count()
+ sc.parallelize(1 to 5000).map(i => (i/2, i)).groupByKey().persist().count()
+ sc.parallelize(1 to 6000).map(i => (i/2, i)).groupByKey().persist().count()
+ sc.parallelize(1 to 7000).map(i => (i/2, i)).groupByKey().persist().count()
+
+ readLine("\n> Started SparkUI with a Foo tab...")
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/ui/Page.scala b/core/src/main/scala/org/apache/spark/ui/Page.scala
deleted file mode 100644
index b2a069a37552d..0000000000000
--- a/core/src/main/scala/org/apache/spark/ui/Page.scala
+++ /dev/null
@@ -1,22 +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.ui
-
-private[spark] object Page extends Enumeration {
- val Stages, Storage, Environment, Executors = Value
-}
diff --git a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
index ef1ad872c8ef7..ac22189f9f04f 100644
--- a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
@@ -17,103 +17,81 @@
package org.apache.spark.ui
-import org.eclipse.jetty.servlet.ServletContextHandler
-
-import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkContext, SparkEnv}
+import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkContext}
import org.apache.spark.scheduler._
import org.apache.spark.storage.StorageStatusListener
import org.apache.spark.ui.JettyUtils._
-import org.apache.spark.ui.env.EnvironmentUI
-import org.apache.spark.ui.exec.ExecutorsUI
-import org.apache.spark.ui.jobs.JobProgressUI
-import org.apache.spark.ui.storage.BlockManagerUI
+import org.apache.spark.ui.env.EnvironmentTab
+import org.apache.spark.ui.exec.ExecutorsTab
+import org.apache.spark.ui.jobs.JobProgressTab
+import org.apache.spark.ui.storage.BlockManagerTab
import org.apache.spark.util.Utils
-/** Top level user interface for Spark */
+/**
+ * Top level user interface for Spark.
+ */
private[spark] class SparkUI(
val sc: SparkContext,
conf: SparkConf,
+ val securityManager: SecurityManager,
val listenerBus: SparkListenerBus,
val appName: String,
val basePath: String = "")
- extends Logging {
+ extends WebUI(securityManager, basePath) with Logging {
- def this(sc: SparkContext) = this(sc, sc.conf, sc.listenerBus, sc.appName)
+ def this(sc: SparkContext) = this(sc, sc.conf, sc.env.securityManager, sc.listenerBus, sc.appName)
def this(conf: SparkConf, listenerBus: SparkListenerBus, appName: String, basePath: String) =
- this(null, conf, listenerBus, appName, basePath)
+ this(null, conf, new SecurityManager(conf), listenerBus, appName, basePath)
// If SparkContext is not provided, assume the associated application is not live
val live = sc != null
- val securityManager = if (live) sc.env.securityManager else new SecurityManager(conf)
-
private val bindHost = Utils.localHostName()
private val publicHost = Option(System.getenv("SPARK_PUBLIC_DNS")).getOrElse(bindHost)
private val port = conf.get("spark.ui.port", SparkUI.DEFAULT_PORT).toInt
- private var serverInfo: Option[ServerInfo] = None
- private val storage = new BlockManagerUI(this)
- private val jobs = new JobProgressUI(this)
- private val env = new EnvironmentUI(this)
- private val exec = new ExecutorsUI(this)
+ // Maintain executor storage status through Spark events
+ val storageStatusListener = new StorageStatusListener
+ listenerBus.addListener(storageStatusListener)
- val handlers: Seq[ServletContextHandler] = {
- val metricsServletHandlers = if (live) {
- SparkEnv.get.metricsSystem.getServletHandlers
- } else {
- Array[ServletContextHandler]()
+ /** Initialize all components of the server. */
+ def start() {
+ attachTab(new JobProgressTab(this))
+ attachTab(new BlockManagerTab(this))
+ attachTab(new EnvironmentTab(this))
+ attachTab(new ExecutorsTab(this))
+ attachHandler(createStaticHandler(SparkUI.STATIC_RESOURCE_DIR, "/static"))
+ attachHandler(createRedirectHandler("/", "/stages", basePath))
+ if (live) {
+ sc.env.metricsSystem.getServletHandlers.foreach(attachHandler)
}
- storage.getHandlers ++
- jobs.getHandlers ++
- env.getHandlers ++
- exec.getHandlers ++
- metricsServletHandlers ++
- Seq[ServletContextHandler] (
- createStaticHandler(SparkUI.STATIC_RESOURCE_DIR, "/static"),
- createRedirectHandler("/", "/stages", basePath)
- )
}
- // Maintain executor storage status through Spark events
- val storageStatusListener = new StorageStatusListener
-
- /** Bind the HTTP server which backs this web interface */
+ /** Bind to the HTTP server behind this web interface. */
def bind() {
try {
serverInfo = Some(startJettyServer(bindHost, port, handlers, sc.conf))
- logInfo("Started Spark Web UI at http://%s:%d".format(publicHost, boundPort))
+ logInfo("Started Spark web UI at http://%s:%d".format(publicHost, boundPort))
} catch {
case e: Exception =>
- logError("Failed to create Spark JettyUtils", e)
+ logError("Failed to create Spark web UI", e)
System.exit(1)
}
}
- def boundPort: Int = serverInfo.map(_.boundPort).getOrElse(-1)
-
- /** Initialize all components of the server */
- def start() {
- storage.start()
- jobs.start()
- env.start()
- exec.start()
-
- // Storage status listener must receive events first, as other listeners depend on its state
- listenerBus.addListener(storageStatusListener)
- listenerBus.addListener(storage.listener)
- listenerBus.addListener(jobs.listener)
- listenerBus.addListener(env.listener)
- listenerBus.addListener(exec.listener)
+ /** Attach a tab to this UI, along with its corresponding listener if it exists. */
+ override def attachTab(tab: UITab) {
+ super.attachTab(tab)
+ tab.listener.foreach(listenerBus.addListener)
}
- def stop() {
- assert(serverInfo.isDefined, "Attempted to stop a SparkUI that was not bound to a server!")
- serverInfo.get.server.stop()
- logInfo("Stopped Spark Web UI at %s".format(appUIAddress))
+ /** Stop the server behind this web interface. Only valid after bind(). */
+ override def stop() {
+ super.stop()
+ logInfo("Stopped Spark web UI at %s".format(appUIAddress))
}
private[spark] def appUIAddress = "http://" + publicHost + ":" + boundPort
-
}
private[spark] object SparkUI {
diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index a487924effbff..7cf16b5ed29b1 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -17,16 +17,40 @@
package org.apache.spark.ui
+import java.text.SimpleDateFormat
+import java.util.Date
+
import scala.xml.Node
/** Utility functions for generating XML pages with spark content. */
private[spark] object UIUtils {
- import Page._
+ // SimpleDateFormat is not thread-safe. Don't expose it to avoid improper use.
+ private val dateFormat = new ThreadLocal[SimpleDateFormat]() {
+ override def initialValue(): SimpleDateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
+ }
+
+ def formatDate(date: Date): String = dateFormat.get.format(date)
+
+ def formatDate(timestamp: Long): String = dateFormat.get.format(new Date(timestamp))
+
+ def formatDuration(milliseconds: Long): String = {
+ val seconds = milliseconds.toDouble / 1000
+ if (seconds < 60) {
+ return "%.0f s".format(seconds)
+ }
+ val minutes = seconds / 60
+ if (minutes < 10) {
+ return "%.1f min".format(minutes)
+ } else if (minutes < 60) {
+ return "%.0f min".format(minutes)
+ }
+ val hours = minutes / 60
+ "%.1f h".format(hours)
+ }
// Yarn has to go through a proxy so the base uri is provided and has to be on all links
- private[spark] val uiRoot : String = Option(System.getenv("APPLICATION_WEB_PROXY_BASE")).
- getOrElse("")
+ val uiRoot : String = Option(System.getenv("APPLICATION_WEB_PROXY_BASE")).getOrElse("")
def prependBaseUri(basePath: String = "", resource: String = "") = uiRoot + basePath + resource
@@ -36,26 +60,13 @@ private[spark] object UIUtils {
basePath: String,
appName: String,
title: String,
- page: Page.Value) : Seq[Node] = {
- val jobs = page match {
- case Stages =>
-
+ Foo Jobs:
+
+ -
+ {results.map { case (k, v) =>
- Job {k}: {v} }} +
![]({prependBaseUri("/static/spark-logo-77x50px-hd.png")}){kind=logo}
-
+
diff --git a/core/src/main/scala/org/apache/spark/ui/WebUI.scala b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
index a7b872f3445a4..126a7ff2f6080 100644
--- a/core/src/main/scala/org/apache/spark/ui/WebUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
@@ -17,34 +17,122 @@
package org.apache.spark.ui
-import java.text.SimpleDateFormat
-import java.util.Date
+import javax.servlet.http.HttpServletRequest
+
+import scala.collection.mutable.ArrayBuffer
+import scala.xml.Node
+
+import org.eclipse.jetty.servlet.ServletContextHandler
+import org.json4s.JsonAST.{JNothing, JValue}
+
+import org.apache.spark.SecurityManager
+import org.apache.spark.scheduler.SparkListener
+import org.apache.spark.ui.JettyUtils._
+import org.apache.spark.util.Utils
/**
- * Utilities used throughout the web UI.
+ * The top level component of the UI hierarchy that contains the server.
+ *
+ * Each WebUI represents a collection of tabs, each of which in turn represents a collection of
+ * pages. The use of tabs is optional, however; a WebUI may choose to include pages directly.
*/
-private[spark] object WebUI {
- // SimpleDateFormat is not thread-safe. Don't expose it to avoid improper use.
- private val dateFormat = new ThreadLocal[SimpleDateFormat]() {
- override def initialValue(): SimpleDateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
- }
+private[spark] abstract class WebUI(securityManager: SecurityManager, basePath: String = "") {
+ protected val tabs = ArrayBuffer[UITab]()
+ protected val handlers = ArrayBuffer[ServletContextHandler]()
+ protected var serverInfo: Option[ServerInfo] = None
- def formatDate(date: Date): String = dateFormat.get.format(date)
+ def getTabs: Seq[UITab] = tabs.toSeq
+ def getHandlers: Seq[ServletContextHandler] = handlers.toSeq
+ def getListeners: Seq[SparkListener] = tabs.flatMap(_.listener)
- def formatDate(timestamp: Long): String = dateFormat.get.format(new Date(timestamp))
+ /** Attach a tab to this UI, along with all of its attached pages. */
+ def attachTab(tab: UITab) {
+ tab.start()
+ tab.pages.foreach(attachPage)
+ tabs += tab
+ }
- def formatDuration(milliseconds: Long): String = {
- val seconds = milliseconds.toDouble / 1000
- if (seconds < 60) {
- return "%.0f s".format(seconds)
+ /** Attach a page to this UI. */
+ def attachPage(page: UIPage) {
+ val pagePath = "/" + page.prefix
+ attachHandler(createServletHandler(pagePath,
+ (request: HttpServletRequest) => page.render(request), securityManager, basePath))
+ if (page.includeJson) {
+ attachHandler(createServletHandler(pagePath.stripSuffix("/") + "/json",
+ (request: HttpServletRequest) => page.renderJson(request), securityManager, basePath))
}
- val minutes = seconds / 60
- if (minutes < 10) {
- return "%.1f min".format(minutes)
- } else if (minutes < 60) {
- return "%.0f min".format(minutes)
+ }
+
+ /** Attach a handler to this UI. */
+ def attachHandler(handler: ServletContextHandler) {
+ handlers += handler
+ serverInfo.foreach { info =>
+ info.rootHandler.addHandler(handler)
+ if (!handler.isStarted) {
+ handler.start()
+ }
}
- val hours = minutes / 60
- return "%.1f h".format(hours)
}
+
+ /** Detach a handler from this UI. */
+ def detachHandler(handler: ServletContextHandler) {
+ handlers -= handler
+ serverInfo.foreach { info =>
+ info.rootHandler.removeHandler(handler)
+ if (handler.isStarted) {
+ handler.stop()
+ }
+ }
+ }
+
+ /** Initialize all components of the server. */
+ def start()
+
+ /**
+ * Bind to the HTTP server behind this web interface.
+ * Overridden implementation should set serverInfo.
+ */
+ def bind()
+
+ /** Return the actual port to which this server is bound. Only valid after bind(). */
+ def boundPort: Int = serverInfo.map(_.boundPort).getOrElse(-1)
+
+ /** Stop the server behind this web interface. Only valid after bind(). */
+ def stop() {
+ assert(serverInfo.isDefined,
+ "Attempted to stop %s before binding to a server!".format(Utils.getFormattedClassName(this)))
+ serverInfo.get.server.stop()
+ }
+}
+
+
+/**
+ * A tab that represents a collection of pages and a unit of listening for Spark events.
+ * Associating each tab with a listener is arbitrary and need not be the case.
+ */
+private[spark] abstract class UITab(val prefix: String) {
+ val pages = ArrayBuffer[UIPage]()
+ var listener: Option[SparkListener] = None
+ var name = prefix.capitalize
+
+ /** Attach a page to this tab. This prepends the page's prefix with the tab's own prefix. */
+ def attachPage(page: UIPage) {
+ page.prefix = (prefix + "/" + page.prefix).stripSuffix("/")
+ pages += page
+ }
+
+ /** Initialize listener and attach pages. */
+ def start()
+}
+
+
+/**
+ * A page that represents the leaf node in the UI hierarchy.
+ *
+ * If includeJson is true, the parent WebUI (direct or indirect) creates handlers for both the
+ * HTML and the JSON content, rather than just the former.
+ */
+private[spark] abstract class UIPage(var prefix: String, val includeJson: Boolean = false) {
+ def render(request: HttpServletRequest): Seq[Node] = Seq[Node]()
+ def renderJson(request: HttpServletRequest): JValue = JNothing
}
diff --git a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentTab.scala b/core/src/main/scala/org/apache/spark/ui/env/EnvironmentTab.scala
new file mode 100644
index 0000000000000..6a2304f1ad42f
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/ui/env/EnvironmentTab.scala
@@ -0,0 +1,58 @@
+/*
+ * 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.ui.env
+
+import org.apache.spark.scheduler._
+import org.apache.spark.ui._
+
+private[ui] class EnvironmentTab(parent: SparkUI) extends UITab("environment") {
+ val appName = parent.appName
+ val basePath = parent.basePath
+
+ def start() {
+ listener = Some(new EnvironmentListener)
+ attachPage(new IndexPage(this))
+ }
+
+ def environmentListener: EnvironmentListener = {
+ assert(listener.isDefined, "EnvironmentTab has not started yet!")
+ listener.get.asInstanceOf[EnvironmentListener]
+ }
+
+ def headerTabs: Seq[UITab] = parent.getTabs
+}
+
+/**
+ * A SparkListener that prepares information to be displayed on the EnvironmentTab
+ */
+private[ui] class EnvironmentListener extends SparkListener {
+ var jvmInformation = Seq[(String, String)]()
+ var sparkProperties = Seq[(String, String)]()
+ var systemProperties = Seq[(String, String)]()
+ var classpathEntries = Seq[(String, String)]()
+
+ override def onEnvironmentUpdate(environmentUpdate: SparkListenerEnvironmentUpdate) {
+ synchronized {
+ val environmentDetails = environmentUpdate.environmentDetails
+ jvmInformation = environmentDetails("JVM Information")
+ sparkProperties = environmentDetails("Spark Properties")
+ systemProperties = environmentDetails("System Properties")
+ classpathEntries = environmentDetails("Classpath Entries")
+ }
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentUI.scala b/core/src/main/scala/org/apache/spark/ui/env/IndexPage.scala
similarity index 60%
rename from core/src/main/scala/org/apache/spark/ui/env/EnvironmentUI.scala
rename to core/src/main/scala/org/apache/spark/ui/env/IndexPage.scala
index 23e90c34d5b33..bde672909bbcc 100644
--- a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/env/IndexPage.scala
@@ -21,30 +21,14 @@ import javax.servlet.http.HttpServletRequest
import scala.xml.Node
-import org.eclipse.jetty.servlet.ServletContextHandler
+import org.apache.spark.ui.{UIUtils, UIPage}
-import org.apache.spark.scheduler._
-import org.apache.spark.ui._
-import org.apache.spark.ui.JettyUtils._
-import org.apache.spark.ui.Page.Environment
-
-private[ui] class EnvironmentUI(parent: SparkUI) {
+private[ui] class IndexPage(parent: EnvironmentTab) extends UIPage("") {
private val appName = parent.appName
private val basePath = parent.basePath
- private var _listener: Option[EnvironmentListener] = None
-
- lazy val listener = _listener.get
-
- def start() {
- _listener = Some(new EnvironmentListener)
- }
-
- def getHandlers = Seq[ServletContextHandler](
- createServletHandler("/environment",
- (request: HttpServletRequest) => render(request), parent.securityManager, basePath)
- )
+ private val listener = parent.environmentListener
- def render(request: HttpServletRequest): Seq[Node] = {
+ override def render(request: HttpServletRequest): Seq[Node] = {
val runtimeInformationTable = UIUtils.listingTable(
propertyHeader, jvmRow, listener.jvmInformation, fixedWidth = true)
val sparkPropertiesTable = UIUtils.listingTable(
@@ -61,7 +45,7 @@ private[ui] class EnvironmentUI(parent: SparkUI) {