You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2014/04/12 08:34:06 UTC
[3/3] git commit: [SPARK-1386] Web UI for Spark Streaming
[SPARK-1386] Web UI for Spark Streaming
When debugging Spark Streaming applications it is necessary to monitor certain metrics that are not shown in the Spark application UI. For example, what is average processing time of batches? What is the scheduling delay? Is the system able to process as fast as it is receiving data? How many records I am receiving through my receivers?
While the StreamingListener interface introduced in the 0.9 provided some of this information, it could only be accessed programmatically. A UI that shows information specific to the streaming applications is necessary for easier debugging. This PR introduces such a UI. It shows various statistics related to the streaming application. Here is a screenshot of the UI running on my local machine.
http://i.imgur.com/1ooDGhm.png
This UI is integrated into the Spark UI running at 4040.
Author: Tathagata Das <ta...@gmail.com>
Author: Andrew Or <an...@gmail.com>
Closes #290 from tdas/streaming-web-ui and squashes the following commits:
fc73ca5 [Tathagata Das] Merge pull request #9 from andrewor14/ui-refactor
642dd88 [Andrew Or] Merge SparkUISuite.scala into UISuite.scala
eb30517 [Andrew Or] Merge github.com:apache/spark into ui-refactor
f4f4cbe [Tathagata Das] More minor fixes.
34bb364 [Tathagata Das] Merge branch 'streaming-web-ui' of github.com:tdas/spark into streaming-web-ui
252c566 [Tathagata Das] Merge pull request #8 from andrewor14/ui-refactor
e038b4b [Tathagata Das] Addressed Patrick's comments.
125a054 [Andrew Or] Disable serving static resources with gzip
90feb8d [Andrew Or] Address Patrick's comments
89dae36 [Tathagata Das] Merge branch 'streaming-web-ui' of github.com:tdas/spark into streaming-web-ui
72fe256 [Tathagata Das] Merge pull request #6 from andrewor14/ui-refactor
2fc09c8 [Tathagata Das] Added binary check exclusions
aa396d4 [Andrew Or] Rename tabs and pages (No more IndexPage.scala)
f8e1053 [Tathagata Das] Added Spark and Streaming UI unit tests.
caa5e05 [Tathagata Das] Merge branch 'streaming-web-ui' of github.com:tdas/spark into streaming-web-ui
585cd65 [Tathagata Das] Merge pull request #5 from andrewor14/ui-refactor
914b8ff [Tathagata Das] Moved utils functions to UIUtils.
548c98c [Andrew Or] Wide refactoring of WebUI, UITab, and UIPage (see commit message)
6de06b0 [Tathagata Das] Merge remote-tracking branch 'apache/master' into streaming-web-ui
ee6543f [Tathagata Das] Minor changes based on Andrew's comments.
fa760fe [Tathagata Das] Fixed long line.
1c0bcef [Tathagata Das] Refactored streaming UI into two files.
1af239b [Tathagata Das] Changed streaming UI to attach itself as a tab with the Spark UI.
827e81a [Tathagata Das] Merge branch 'streaming-web-ui' of github.com:tdas/spark into streaming-web-ui
168fe86 [Tathagata Das] Merge pull request #2 from andrewor14/ui-refactor
3e986f8 [Tathagata Das] Merge remote-tracking branch 'apache/master' into streaming-web-ui
c78c92d [Andrew Or] Remove outdated comment
8f7323b [Andrew Or] End of file new lines, indentation, and imports (minor)
0d61ee8 [Andrew Or] Merge branch 'streaming-web-ui' of github.com:tdas/spark into ui-refactor
9a48fa1 [Andrew Or] Allow adding tabs to SparkUI dynamically + add example
61358e3 [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into streaming-web-ui
53be2c5 [Tathagata Das] Minor style updates.
ed25dfc [Andrew Or] Generalize SparkUI header to display tabs dynamically
a37ad4f [Andrew Or] Comments, imports and formatting (minor)
cd000b0 [Andrew Or] Merge github.com:apache/spark into ui-refactor
7d57444 [Andrew Or] Refactoring the UI interface to add flexibility
aef4dd5 [Tathagata Das] Added Apache licenses.
db27bad [Tathagata Das] Added last batch processing time to StreamingUI.
4d86e98 [Tathagata Das] Added basic stats to the StreamingUI and refactored the UI to a Page to make it easier to transition to using SparkUI later.
93f1c69 [Tathagata Das] Added network receiver information to the Streaming UI.
56cc7fb [Tathagata Das] First cut implementation of Streaming UI.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6aa08c39
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6aa08c39
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6aa08c39
Branch: refs/heads/master
Commit: 6aa08c39cf30fa5c4ed97f4fff16371b9030a2e6
Parents: 165e06a
Author: Tathagata Das <ta...@gmail.com>
Authored: Fri Apr 11 23:33:49 2014 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Fri Apr 11 23:33:49 2014 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/SparkContext.scala | 1 -
.../apache/spark/deploy/SparkUIContainer.scala | 50 -----
.../spark/deploy/history/HistoryPage.scala | 82 +++++++
.../spark/deploy/history/HistoryServer.scala | 61 +++---
.../apache/spark/deploy/history/IndexPage.scala | 82 -------
.../org/apache/spark/deploy/master/Master.scala | 8 +-
.../deploy/master/ui/ApplicationPage.scala | 13 +-
.../spark/deploy/master/ui/IndexPage.scala | 195 -----------------
.../spark/deploy/master/ui/MasterPage.scala | 194 +++++++++++++++++
.../spark/deploy/master/ui/MasterWebUI.scala | 54 ++---
.../org/apache/spark/deploy/worker/Worker.scala | 2 +-
.../spark/deploy/worker/ui/IndexPage.scala | 165 --------------
.../apache/spark/deploy/worker/ui/LogPage.scala | 147 +++++++++++++
.../spark/deploy/worker/ui/WorkerPage.scala | 165 ++++++++++++++
.../spark/deploy/worker/ui/WorkerWebUI.scala | 180 +++-------------
.../scheduler/ApplicationEventListener.scala | 4 +-
.../org/apache/spark/storage/StorageUtils.scala | 16 +-
.../scala/org/apache/spark/ui/JettyUtils.scala | 1 +
.../main/scala/org/apache/spark/ui/Page.scala | 22 --
.../scala/org/apache/spark/ui/SparkUI.scala | 108 ++++------
.../scala/org/apache/spark/ui/UIUtils.scala | 172 +++++++++++----
.../main/scala/org/apache/spark/ui/WebUI.scala | 141 +++++++++---
.../apache/spark/ui/env/EnvironmentPage.scala | 56 +++++
.../apache/spark/ui/env/EnvironmentTab.scala | 50 +++++
.../org/apache/spark/ui/env/EnvironmentUI.scala | 93 --------
.../apache/spark/ui/exec/ExecutorsPage.scala | 141 ++++++++++++
.../org/apache/spark/ui/exec/ExecutorsTab.scala | 86 ++++++++
.../org/apache/spark/ui/exec/ExecutorsUI.scala | 213 -------------------
.../apache/spark/ui/jobs/ExecutorTable.scala | 7 +-
.../org/apache/spark/ui/jobs/IndexPage.scala | 100 ---------
.../spark/ui/jobs/JobProgressListener.scala | 10 +-
.../apache/spark/ui/jobs/JobProgressPage.scala | 98 +++++++++
.../apache/spark/ui/jobs/JobProgressTab.scala | 56 +++++
.../apache/spark/ui/jobs/JobProgressUI.scala | 77 -------
.../org/apache/spark/ui/jobs/PoolPage.scala | 14 +-
.../org/apache/spark/ui/jobs/PoolTable.scala | 7 +-
.../org/apache/spark/ui/jobs/StagePage.scala | 45 ++--
.../org/apache/spark/ui/jobs/StageTable.scala | 18 +-
.../spark/ui/storage/BlockManagerUI.scala | 99 ---------
.../org/apache/spark/ui/storage/IndexPage.scala | 70 ------
.../org/apache/spark/ui/storage/RDDPage.scala | 17 +-
.../apache/spark/ui/storage/StoragePage.scala | 67 ++++++
.../apache/spark/ui/storage/StorageTab.scala | 81 +++++++
.../org/apache/spark/util/JsonProtocol.scala | 12 +-
.../scala/org/apache/spark/SparkUISuite.scala | 35 ---
.../scala/org/apache/spark/ui/UISuite.scala | 81 ++++++-
.../apache/spark/util/JsonProtocolSuite.scala | 4 +-
project/MimaBuild.scala | 8 +-
.../spark/streaming/StreamingContext.scala | 23 +-
.../spark/streaming/dstream/DStream.scala | 9 -
.../streaming/dstream/NetworkInputDStream.scala | 79 +++++--
.../spark/streaming/scheduler/BatchInfo.scala | 1 +
.../streaming/scheduler/JobGenerator.scala | 9 +-
.../streaming/scheduler/JobScheduler.scala | 11 +-
.../spark/streaming/scheduler/JobSet.scala | 7 +-
.../scheduler/NetworkInputTracker.scala | 86 +++++---
.../streaming/scheduler/StreamingListener.scala | 18 +-
.../scheduler/StreamingListenerBus.scala | 4 +
.../ui/StreamingJobProgressListener.scala | 148 +++++++++++++
.../spark/streaming/ui/StreamingPage.scala | 180 ++++++++++++++++
.../spark/streaming/ui/StreamingTab.scala | 36 ++++
.../spark/streaming/InputStreamsSuite.scala | 6 +-
.../spark/streaming/StreamingContextSuite.scala | 1 -
.../org/apache/spark/streaming/UISuite.scala | 46 ++++
64 files changed, 2326 insertions(+), 1746 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 3bcc8ce..a764c17 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -213,7 +213,6 @@ class SparkContext(config: SparkConf) extends Logging {
// Initialize the Spark UI, registering all associated listeners
private[spark] val ui = new SparkUI(this)
ui.bind()
- ui.start()
// Optionally log Spark events
private[spark] val eventLogger: Option[EventLoggingListener] = {
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/SparkUIContainer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkUIContainer.scala b/core/src/main/scala/org/apache/spark/deploy/SparkUIContainer.scala
deleted file mode 100644
index 33fceae..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/SparkUIContainer.scala
+++ /dev/null
@@ -1,50 +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
-
-import org.apache.spark.ui.{SparkUI, WebUI}
-
-private[spark] abstract class SparkUIContainer(name: String) extends WebUI(name) {
-
- /** Attach a SparkUI to this container. Only valid after bind(). */
- def attachUI(ui: SparkUI) {
- assert(serverInfo.isDefined,
- "%s must be bound to a server before attaching SparkUIs".format(name))
- val rootHandler = serverInfo.get.rootHandler
- for (handler <- ui.handlers) {
- rootHandler.addHandler(handler)
- if (!handler.isStarted) {
- handler.start()
- }
- }
- }
-
- /** Detach a SparkUI from this container. Only valid after bind(). */
- def detachUI(ui: SparkUI) {
- assert(serverInfo.isDefined,
- "%s must be bound to a server before detaching SparkUIs".format(name))
- val rootHandler = serverInfo.get.rootHandler
- for (handler <- ui.handlers) {
- if (handler.isStarted) {
- handler.stop()
- }
- rootHandler.removeHandler(handler)
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala
new file mode 100644
index 0000000..180c853
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala
@@ -0,0 +1,82 @@
+/*
+ * 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.history
+
+import javax.servlet.http.HttpServletRequest
+
+import scala.xml.Node
+
+import org.apache.spark.ui.{WebUIPage, UIUtils}
+
+private[spark] class HistoryPage(parent: HistoryServer) extends WebUIPage("") {
+
+ def render(request: HttpServletRequest): Seq[Node] = {
+ val appRows = parent.appIdToInfo.values.toSeq.sortBy { app => -app.lastUpdated }
+ val appTable = UIUtils.listingTable(appHeader, appRow, appRows)
+ val content =
+ <div class="row-fluid">
+ <div class="span12">
+ <ul class="unstyled">
+ <li><strong>Event Log Location: </strong> {parent.baseLogDir}</li>
+ </ul>
+ {
+ if (parent.appIdToInfo.size > 0) {
+ <h4>
+ Showing {parent.appIdToInfo.size}/{parent.getNumApplications}
+ Completed Application{if (parent.getNumApplications > 1) "s" else ""}
+ </h4> ++
+ appTable
+ } else {
+ <h4>No Completed Applications Found</h4>
+ }
+ }
+ </div>
+ </div>
+ UIUtils.basicSparkPage(content, "History Server")
+ }
+
+ private val appHeader = Seq(
+ "App Name",
+ "Started",
+ "Completed",
+ "Duration",
+ "Spark User",
+ "Log Directory",
+ "Last Updated")
+
+ private def appRow(info: ApplicationHistoryInfo): Seq[Node] = {
+ val appName = if (info.started) info.name else info.logDirPath.getName
+ val uiAddress = parent.getAddress + info.ui.basePath
+ val startTime = if (info.started) UIUtils.formatDate(info.startTime) else "Not started"
+ val endTime = if (info.completed) UIUtils.formatDate(info.endTime) else "Not completed"
+ val difference = if (info.started && info.completed) info.endTime - info.startTime else -1L
+ val duration = if (difference > 0) UIUtils.formatDuration(difference) else "---"
+ val sparkUser = if (info.started) info.sparkUser else "Unknown user"
+ val logDirectory = info.logDirPath.getName
+ val lastUpdated = UIUtils.formatDate(info.lastUpdated)
+ <tr>
+ <td><a href={uiAddress}>{appName}</a></td>
+ <td>{startTime}</td>
+ <td>{endTime}</td>
+ <td>{duration}</td>
+ <td>{sparkUser}</td>
+ <td>{logDirectory}</td>
+ <td>{lastUpdated}</td>
+ </tr>
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala
index 97d2ba9..cf64700 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala
@@ -17,17 +17,13 @@
package org.apache.spark.deploy.history
-import javax.servlet.http.HttpServletRequest
-
import scala.collection.mutable
import org.apache.hadoop.fs.{FileStatus, Path}
-import org.eclipse.jetty.servlet.ServletContextHandler
import org.apache.spark.{Logging, SecurityManager, SparkConf}
-import org.apache.spark.deploy.SparkUIContainer
import org.apache.spark.scheduler._
-import org.apache.spark.ui.SparkUI
+import org.apache.spark.ui.{WebUI, SparkUI}
import org.apache.spark.ui.JettyUtils._
import org.apache.spark.util.Utils
@@ -46,17 +42,15 @@ import org.apache.spark.util.Utils
*/
class HistoryServer(
val baseLogDir: String,
+ securityManager: SecurityManager,
conf: SparkConf)
- extends SparkUIContainer("History Server") with Logging {
+ extends WebUI(securityManager, HistoryServer.WEB_UI_PORT, conf) with Logging {
import HistoryServer._
private val fileSystem = Utils.getHadoopFileSystem(baseLogDir)
private val localHost = Utils.localHostName()
private val publicHost = Option(System.getenv("SPARK_PUBLIC_DNS")).getOrElse(localHost)
- private val port = WEB_UI_PORT
- private val securityManager = new SecurityManager(conf)
- private val indexPage = new IndexPage(this)
// A timestamp of when the disk was last accessed to check for log updates
private var lastLogCheckTime = -1L
@@ -90,37 +84,23 @@ class HistoryServer(
}
}
- private val handlers = Seq[ServletContextHandler](
- createStaticHandler(STATIC_RESOURCE_DIR, "/static"),
- createServletHandler("/",
- (request: HttpServletRequest) => indexPage.render(request), securityMgr = securityManager)
- )
-
// A mapping of application ID to its history information, which includes the rendered UI
val appIdToInfo = mutable.HashMap[String, ApplicationHistoryInfo]()
+ initialize()
+
/**
- * Start the history server.
+ * Initialize the history server.
*
* This starts a background thread that periodically synchronizes information displayed on
* this UI with the event logs in the provided base directory.
*/
- def start() {
+ def initialize() {
+ attachPage(new HistoryPage(this))
+ attachHandler(createStaticHandler(STATIC_RESOURCE_DIR, "/static"))
logCheckingThread.start()
}
- /** Bind to the HTTP server behind this web interface. */
- override def bind() {
- try {
- serverInfo = Some(startJettyServer("0.0.0.0", port, handlers, conf))
- logInfo("Started HistoryServer at http://%s:%d".format(publicHost, boundPort))
- } catch {
- case e: Exception =>
- logError("Failed to bind HistoryServer", e)
- System.exit(1)
- }
- }
-
/**
* Check for any updates to event logs in the base directory. This is only effective once
* the server has been bound.
@@ -151,7 +131,7 @@ class HistoryServer(
// Remove any applications that should no longer be retained
appIdToInfo.foreach { case (appId, info) =>
if (!retainedAppIds.contains(appId)) {
- detachUI(info.ui)
+ detachSparkUI(info.ui)
appIdToInfo.remove(appId)
}
}
@@ -186,15 +166,14 @@ class HistoryServer(
val path = logDir.getPath
val appId = path.getName
val replayBus = new ReplayListenerBus(logInfo.logPaths, fileSystem, logInfo.compressionCodec)
- val ui = new SparkUI(replayBus, appId, "/history/" + appId)
val appListener = new ApplicationEventListener
replayBus.addListener(appListener)
+ val ui = new SparkUI(conf, replayBus, appId, "/history/" + appId)
// Do not call ui.bind() to avoid creating a new server for each application
- ui.start()
replayBus.replay()
if (appListener.applicationStarted) {
- attachUI(ui)
+ attachSparkUI(ui)
val appName = appListener.appName
val sparkUser = appListener.sparkUser
val startTime = appListener.startTime
@@ -213,6 +192,18 @@ class HistoryServer(
fileSystem.close()
}
+ /** Attach a reconstructed UI to this server. Only valid after bind(). */
+ private def attachSparkUI(ui: SparkUI) {
+ assert(serverInfo.isDefined, "HistoryServer must be bound before attaching SparkUIs")
+ ui.getHandlers.foreach(attachHandler)
+ }
+
+ /** Detach a reconstructed UI from this server. Only valid after bind(). */
+ private def detachSparkUI(ui: SparkUI) {
+ assert(serverInfo.isDefined, "HistoryServer must be bound before detaching SparkUIs")
+ ui.getHandlers.foreach(detachHandler)
+ }
+
/** Return the address of this server. */
def getAddress: String = "http://" + publicHost + ":" + boundPort
@@ -262,9 +253,9 @@ object HistoryServer {
def main(argStrings: Array[String]) {
val args = new HistoryServerArguments(argStrings)
- val server = new HistoryServer(args.logDir, conf)
+ val securityManager = new SecurityManager(conf)
+ val server = new HistoryServer(args.logDir, securityManager, conf)
server.bind()
- server.start()
// Wait until the end of the world... or if the HistoryServer process is manually stopped
while(true) { Thread.sleep(Int.MaxValue) }
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/history/IndexPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/IndexPage.scala b/core/src/main/scala/org/apache/spark/deploy/history/IndexPage.scala
deleted file mode 100644
index 54dffff..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/history/IndexPage.scala
+++ /dev/null
@@ -1,82 +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.history
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.xml.Node
-
-import org.apache.spark.ui.{UIUtils, WebUI}
-
-private[spark] class IndexPage(parent: HistoryServer) {
-
- def render(request: HttpServletRequest): Seq[Node] = {
- val appRows = parent.appIdToInfo.values.toSeq.sortBy { app => -app.lastUpdated }
- val appTable = UIUtils.listingTable(appHeader, appRow, appRows)
- val content =
- <div class="row-fluid">
- <div class="span12">
- <ul class="unstyled">
- <li><strong>Event Log Location: </strong> {parent.baseLogDir}</li>
- </ul>
- {
- if (parent.appIdToInfo.size > 0) {
- <h4>
- Showing {parent.appIdToInfo.size}/{parent.getNumApplications}
- Completed Application{if (parent.getNumApplications > 1) "s" else ""}
- </h4> ++
- appTable
- } else {
- <h4>No Completed Applications Found</h4>
- }
- }
- </div>
- </div>
- UIUtils.basicSparkPage(content, "History Server")
- }
-
- private val appHeader = Seq(
- "App Name",
- "Started",
- "Completed",
- "Duration",
- "Spark User",
- "Log Directory",
- "Last Updated")
-
- private def appRow(info: ApplicationHistoryInfo): Seq[Node] = {
- val appName = if (info.started) info.name else info.logDirPath.getName
- val uiAddress = parent.getAddress + info.ui.basePath
- val startTime = if (info.started) WebUI.formatDate(info.startTime) else "Not started"
- val endTime = if (info.completed) WebUI.formatDate(info.endTime) else "Not completed"
- val difference = if (info.started && info.completed) info.endTime - info.startTime else -1L
- val duration = if (difference > 0) WebUI.formatDuration(difference) else "---"
- val sparkUser = if (info.started) info.sparkUser else "Unknown user"
- val logDirectory = info.logDirPath.getName
- val lastUpdated = WebUI.formatDate(info.lastUpdated)
- <tr>
- <td><a href={uiAddress}>{appName}</a></td>
- <td>{startTime}</td>
- <td>{endTime}</td>
- <td>{duration}</td>
- <td>{sparkUser}</td>
- <td>{logDirectory}</td>
- <td>{lastUpdated}</td>
- </tr>
- }
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/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 2446e86..6c58e74 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
@@ -625,7 +625,7 @@ private[spark] class Master(
if (completedApps.size >= RETAINED_APPLICATIONS) {
val toRemove = math.max(RETAINED_APPLICATIONS / 10, 1)
completedApps.take(toRemove).foreach( a => {
- appIdToUI.remove(a.id).foreach { ui => webUi.detachUI(ui) }
+ appIdToUI.remove(a.id).foreach { ui => webUi.detachSparkUI(ui) }
applicationMetricsSystem.removeSource(a.appSource)
})
completedApps.trimStart(toRemove)
@@ -667,12 +667,12 @@ private[spark] class Master(
if (!eventLogPaths.isEmpty) {
try {
val replayBus = new ReplayListenerBus(eventLogPaths, fileSystem, compressionCodec)
- val ui = new SparkUI(replayBus, appName + " (completed)", "/history/" + app.id)
- ui.start()
+ val ui = new SparkUI(
+ new SparkConf, replayBus, appName + " (completed)", "/history/" + app.id)
replayBus.replay()
app.desc.appUiUrl = ui.basePath
appIdToUI(app.id) = ui
- webUi.attachUI(ui)
+ webUi.attachSparkUI(ui)
return true
} catch {
case t: Throwable =>
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
index cb092cb..b5cd4d2 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
@@ -28,15 +28,16 @@ import org.json4s.JValue
import org.apache.spark.deploy.JsonProtocol
import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, RequestMasterState}
import org.apache.spark.deploy.master.ExecutorInfo
-import org.apache.spark.ui.UIUtils
+import org.apache.spark.ui.{WebUIPage, UIUtils}
import org.apache.spark.util.Utils
-private[spark] class ApplicationPage(parent: MasterWebUI) {
- val master = parent.masterActorRef
- val timeout = parent.timeout
+private[spark] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app") {
+
+ private val master = parent.masterActorRef
+ private val timeout = parent.timeout
/** Executor details for a particular application */
- def renderJson(request: HttpServletRequest): JValue = {
+ override def renderJson(request: HttpServletRequest): JValue = {
val appId = request.getParameter("appId")
val stateFuture = (master ? RequestMasterState)(timeout).mapTo[MasterStateResponse]
val state = Await.result(stateFuture, timeout)
@@ -96,7 +97,7 @@ private[spark] class ApplicationPage(parent: MasterWebUI) {
UIUtils.basicSparkPage(content, "Application: " + app.desc.name)
}
- def executorRow(executor: ExecutorInfo): Seq[Node] = {
+ private def executorRow(executor: ExecutorInfo): Seq[Node] = {
<tr>
<td>{executor.id}</td>
<td>
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/master/ui/IndexPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/IndexPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/IndexPage.scala
deleted file mode 100644
index 8c1d6c7..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/IndexPage.scala
+++ /dev/null
@@ -1,195 +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.ui
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.concurrent.Await
-import scala.xml.Node
-
-import akka.pattern.ask
-import org.json4s.JValue
-
-import org.apache.spark.deploy.{JsonProtocol}
-import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, RequestMasterState}
-import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo, WorkerInfo}
-import org.apache.spark.ui.{WebUI, UIUtils}
-import org.apache.spark.util.Utils
-
-private[spark] class IndexPage(parent: MasterWebUI) {
- val master = parent.masterActorRef
- val timeout = parent.timeout
-
- def renderJson(request: HttpServletRequest): JValue = {
- val stateFuture = (master ? RequestMasterState)(timeout).mapTo[MasterStateResponse]
- val state = Await.result(stateFuture, timeout)
- JsonProtocol.writeMasterState(state)
- }
-
- /** Index view listing applications and executors */
- def render(request: HttpServletRequest): Seq[Node] = {
- val stateFuture = (master ? RequestMasterState)(timeout).mapTo[MasterStateResponse]
- val state = Await.result(stateFuture, timeout)
-
- val workerHeaders = Seq("Id", "Address", "State", "Cores", "Memory")
- val workers = state.workers.sortBy(_.id)
- val workerTable = UIUtils.listingTable(workerHeaders, workerRow, workers)
-
- val appHeaders = Seq("ID", "Name", "Cores", "Memory per Node", "Submitted Time", "User",
- "State", "Duration")
- val activeApps = state.activeApps.sortBy(_.startTime).reverse
- val activeAppsTable = UIUtils.listingTable(appHeaders, appRow, activeApps)
- val completedApps = state.completedApps.sortBy(_.endTime).reverse
- val completedAppsTable = UIUtils.listingTable(appHeaders, appRow, completedApps)
-
- val driverHeaders = Seq("ID", "Submitted Time", "Worker", "State", "Cores", "Memory",
- "Main Class")
- val activeDrivers = state.activeDrivers.sortBy(_.startTime).reverse
- val activeDriversTable = UIUtils.listingTable(driverHeaders, driverRow, activeDrivers)
- val completedDrivers = state.completedDrivers.sortBy(_.startTime).reverse
- val completedDriversTable = UIUtils.listingTable(driverHeaders, driverRow, completedDrivers)
-
- // For now we only show driver information if the user has submitted drivers to the cluster.
- // This is until we integrate the notion of drivers and applications in the UI.
- def hasDrivers = activeDrivers.length > 0 || completedDrivers.length > 0
-
- val content =
- <div class="row-fluid">
- <div class="span12">
- <ul class="unstyled">
- <li><strong>URL:</strong> {state.uri}</li>
- <li><strong>Workers:</strong> {state.workers.size}</li>
- <li><strong>Cores:</strong> {state.workers.map(_.cores).sum} Total,
- {state.workers.map(_.coresUsed).sum} Used</li>
- <li><strong>Memory:</strong>
- {Utils.megabytesToString(state.workers.map(_.memory).sum)} Total,
- {Utils.megabytesToString(state.workers.map(_.memoryUsed).sum)} Used</li>
- <li><strong>Applications:</strong>
- {state.activeApps.size} Running,
- {state.completedApps.size} Completed </li>
- <li><strong>Drivers:</strong>
- {state.activeDrivers.size} Running,
- {state.completedDrivers.size} Completed </li>
- <li><strong>Status:</strong> {state.status}</li>
- </ul>
- </div>
- </div>
-
- <div class="row-fluid">
- <div class="span12">
- <h4> Workers </h4>
- {workerTable}
- </div>
- </div>
-
- <div class="row-fluid">
- <div class="span12">
- <h4> Running Applications </h4>
- {activeAppsTable}
- </div>
- </div>
-
- <div>
- {if (hasDrivers) {
- <div class="row-fluid">
- <div class="span12">
- <h4> Running Drivers </h4>
- {activeDriversTable}
- </div>
- </div>
- }
- }
- </div>
-
- <div class="row-fluid">
- <div class="span12">
- <h4> Completed Applications </h4>
- {completedAppsTable}
- </div>
- </div>
-
- <div>
- {
- if (hasDrivers) {
- <div class="row-fluid">
- <div class="span12">
- <h4> Completed Drivers </h4>
- {completedDriversTable}
- </div>
- </div>
- }
- }
- </div>;
-
- UIUtils.basicSparkPage(content, "Spark Master at " + state.uri)
- }
-
- def workerRow(worker: WorkerInfo): Seq[Node] = {
- <tr>
- <td>
- <a href={worker.webUiAddress}>{worker.id}</a>
- </td>
- <td>{worker.host}:{worker.port}</td>
- <td>{worker.state}</td>
- <td>{worker.cores} ({worker.coresUsed} Used)</td>
- <td sorttable_customkey={"%s.%s".format(worker.memory, worker.memoryUsed)}>
- {Utils.megabytesToString(worker.memory)}
- ({Utils.megabytesToString(worker.memoryUsed)} Used)
- </td>
- </tr>
- }
-
-
- def appRow(app: ApplicationInfo): Seq[Node] = {
- <tr>
- <td>
- <a href={"app?appId=" + app.id}>{app.id}</a>
- </td>
- <td>
- <a href={app.desc.appUiUrl}>{app.desc.name}</a>
- </td>
- <td>
- {app.coresGranted}
- </td>
- <td sorttable_customkey={app.desc.memoryPerSlave.toString}>
- {Utils.megabytesToString(app.desc.memoryPerSlave)}
- </td>
- <td>{WebUI.formatDate(app.submitDate)}</td>
- <td>{app.desc.user}</td>
- <td>{app.state.toString}</td>
- <td>{WebUI.formatDuration(app.duration)}</td>
- </tr>
- }
-
- def driverRow(driver: DriverInfo): Seq[Node] = {
- <tr>
- <td>{driver.id} </td>
- <td>{driver.submitDate}</td>
- <td>{driver.worker.map(w => <a href={w.webUiAddress}>{w.id.toString}</a>).getOrElse("None")}
- </td>
- <td>{driver.state}</td>
- <td sorttable_customkey={driver.desc.cores.toString}>
- {driver.desc.cores}
- </td>
- <td sorttable_customkey={driver.desc.mem.toString}>
- {Utils.megabytesToString(driver.desc.mem.toLong)}
- </td>
- <td>{driver.desc.command.arguments(1)}</td>
- </tr>
- }
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
new file mode 100644
index 0000000..7ca3b08
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
@@ -0,0 +1,194 @@
+/*
+ * 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.ui
+
+import javax.servlet.http.HttpServletRequest
+
+import scala.concurrent.Await
+import scala.xml.Node
+
+import akka.pattern.ask
+import org.json4s.JValue
+
+import org.apache.spark.deploy.JsonProtocol
+import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, RequestMasterState}
+import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo, WorkerInfo}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
+import org.apache.spark.util.Utils
+
+private[spark] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
+ private val master = parent.masterActorRef
+ private val timeout = parent.timeout
+
+ override def renderJson(request: HttpServletRequest): JValue = {
+ val stateFuture = (master ? RequestMasterState)(timeout).mapTo[MasterStateResponse]
+ val state = Await.result(stateFuture, timeout)
+ JsonProtocol.writeMasterState(state)
+ }
+
+ /** Index view listing applications and executors */
+ def render(request: HttpServletRequest): Seq[Node] = {
+ val stateFuture = (master ? RequestMasterState)(timeout).mapTo[MasterStateResponse]
+ val state = Await.result(stateFuture, timeout)
+
+ val workerHeaders = Seq("Id", "Address", "State", "Cores", "Memory")
+ val workers = state.workers.sortBy(_.id)
+ val workerTable = UIUtils.listingTable(workerHeaders, workerRow, workers)
+
+ val appHeaders = Seq("ID", "Name", "Cores", "Memory per Node", "Submitted Time", "User",
+ "State", "Duration")
+ val activeApps = state.activeApps.sortBy(_.startTime).reverse
+ val activeAppsTable = UIUtils.listingTable(appHeaders, appRow, activeApps)
+ val completedApps = state.completedApps.sortBy(_.endTime).reverse
+ val completedAppsTable = UIUtils.listingTable(appHeaders, appRow, completedApps)
+
+ val driverHeaders = Seq("ID", "Submitted Time", "Worker", "State", "Cores", "Memory",
+ "Main Class")
+ val activeDrivers = state.activeDrivers.sortBy(_.startTime).reverse
+ val activeDriversTable = UIUtils.listingTable(driverHeaders, driverRow, activeDrivers)
+ val completedDrivers = state.completedDrivers.sortBy(_.startTime).reverse
+ val completedDriversTable = UIUtils.listingTable(driverHeaders, driverRow, completedDrivers)
+
+ // For now we only show driver information if the user has submitted drivers to the cluster.
+ // This is until we integrate the notion of drivers and applications in the UI.
+ def hasDrivers = activeDrivers.length > 0 || completedDrivers.length > 0
+
+ val content =
+ <div class="row-fluid">
+ <div class="span12">
+ <ul class="unstyled">
+ <li><strong>URL:</strong> {state.uri}</li>
+ <li><strong>Workers:</strong> {state.workers.size}</li>
+ <li><strong>Cores:</strong> {state.workers.map(_.cores).sum} Total,
+ {state.workers.map(_.coresUsed).sum} Used</li>
+ <li><strong>Memory:</strong>
+ {Utils.megabytesToString(state.workers.map(_.memory).sum)} Total,
+ {Utils.megabytesToString(state.workers.map(_.memoryUsed).sum)} Used</li>
+ <li><strong>Applications:</strong>
+ {state.activeApps.size} Running,
+ {state.completedApps.size} Completed </li>
+ <li><strong>Drivers:</strong>
+ {state.activeDrivers.size} Running,
+ {state.completedDrivers.size} Completed </li>
+ <li><strong>Status:</strong> {state.status}</li>
+ </ul>
+ </div>
+ </div>
+
+ <div class="row-fluid">
+ <div class="span12">
+ <h4> Workers </h4>
+ {workerTable}
+ </div>
+ </div>
+
+ <div class="row-fluid">
+ <div class="span12">
+ <h4> Running Applications </h4>
+ {activeAppsTable}
+ </div>
+ </div>
+
+ <div>
+ {if (hasDrivers) {
+ <div class="row-fluid">
+ <div class="span12">
+ <h4> Running Drivers </h4>
+ {activeDriversTable}
+ </div>
+ </div>
+ }
+ }
+ </div>
+
+ <div class="row-fluid">
+ <div class="span12">
+ <h4> Completed Applications </h4>
+ {completedAppsTable}
+ </div>
+ </div>
+
+ <div>
+ {
+ if (hasDrivers) {
+ <div class="row-fluid">
+ <div class="span12">
+ <h4> Completed Drivers </h4>
+ {completedDriversTable}
+ </div>
+ </div>
+ }
+ }
+ </div>;
+
+ UIUtils.basicSparkPage(content, "Spark Master at " + state.uri)
+ }
+
+ private def workerRow(worker: WorkerInfo): Seq[Node] = {
+ <tr>
+ <td>
+ <a href={worker.webUiAddress}>{worker.id}</a>
+ </td>
+ <td>{worker.host}:{worker.port}</td>
+ <td>{worker.state}</td>
+ <td>{worker.cores} ({worker.coresUsed} Used)</td>
+ <td sorttable_customkey={"%s.%s".format(worker.memory, worker.memoryUsed)}>
+ {Utils.megabytesToString(worker.memory)}
+ ({Utils.megabytesToString(worker.memoryUsed)} Used)
+ </td>
+ </tr>
+ }
+
+ private def appRow(app: ApplicationInfo): Seq[Node] = {
+ <tr>
+ <td>
+ <a href={"app?appId=" + app.id}>{app.id}</a>
+ </td>
+ <td>
+ <a href={app.desc.appUiUrl}>{app.desc.name}</a>
+ </td>
+ <td>
+ {app.coresGranted}
+ </td>
+ <td sorttable_customkey={app.desc.memoryPerSlave.toString}>
+ {Utils.megabytesToString(app.desc.memoryPerSlave)}
+ </td>
+ <td>{UIUtils.formatDate(app.submitDate)}</td>
+ <td>{app.desc.user}</td>
+ <td>{app.state.toString}</td>
+ <td>{UIUtils.formatDuration(app.duration)}</td>
+ </tr>
+ }
+
+ private def driverRow(driver: DriverInfo): Seq[Node] = {
+ <tr>
+ <td>{driver.id} </td>
+ <td>{driver.submitDate}</td>
+ <td>{driver.worker.map(w => <a href={w.webUiAddress}>{w.id.toString}</a>).getOrElse("None")}
+ </td>
+ <td>{driver.state}</td>
+ <td sorttable_customkey={driver.desc.cores.toString}>
+ {driver.desc.cores}
+ </td>
+ <td sorttable_customkey={driver.desc.mem.toString}>
+ {Utils.megabytesToString(driver.desc.mem.toLong)}
+ </td>
+ <td>{driver.desc.command.arguments(1)}</td>
+ </tr>
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
index 30c8ade..a18b39f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
@@ -17,14 +17,9 @@
package org.apache.spark.deploy.master.ui
-import javax.servlet.http.HttpServletRequest
-
-import org.eclipse.jetty.servlet.ServletContextHandler
-
import org.apache.spark.Logging
-import org.apache.spark.deploy.SparkUIContainer
import org.apache.spark.deploy.master.Master
-import org.apache.spark.ui.SparkUI
+import org.apache.spark.ui.{SparkUI, WebUI}
import org.apache.spark.ui.JettyUtils._
import org.apache.spark.util.{AkkaUtils, Utils}
@@ -33,44 +28,33 @@ import org.apache.spark.util.{AkkaUtils, Utils}
*/
private[spark]
class MasterWebUI(val master: Master, requestedPort: Int)
- extends SparkUIContainer("MasterWebUI") with Logging {
+ extends WebUI(master.securityMgr, requestedPort, master.conf) with Logging {
val masterActorRef = master.self
val timeout = AkkaUtils.askTimeout(master.conf)
- private val host = Utils.localHostName()
- private val port = requestedPort
- private val applicationPage = new ApplicationPage(this)
- private val indexPage = new IndexPage(this)
+ initialize()
- private val handlers: Seq[ServletContextHandler] = {
- master.masterMetricsSystem.getServletHandlers ++
- master.applicationMetricsSystem.getServletHandlers ++
- Seq[ServletContextHandler](
- createStaticHandler(MasterWebUI.STATIC_RESOURCE_DIR, "/static"),
- createServletHandler("/app/json",
- (request: HttpServletRequest) => applicationPage.renderJson(request), master.securityMgr),
- createServletHandler("/app",
- (request: HttpServletRequest) => applicationPage.render(request), master.securityMgr),
- createServletHandler("/json",
- (request: HttpServletRequest) => indexPage.renderJson(request), master.securityMgr),
- createServletHandler("/",
- (request: HttpServletRequest) => indexPage.render(request), master.securityMgr)
- )
+ /** Initialize all components of the server. */
+ def initialize() {
+ attachPage(new ApplicationPage(this))
+ attachPage(new MasterPage(this))
+ attachHandler(createStaticHandler(MasterWebUI.STATIC_RESOURCE_DIR, "/static"))
+ master.masterMetricsSystem.getServletHandlers.foreach(attachHandler)
+ master.applicationMetricsSystem.getServletHandlers.foreach(attachHandler)
}
- /** Bind to the HTTP server behind this web interface. */
- override def bind() {
- try {
- serverInfo = Some(startJettyServer("0.0.0.0", port, handlers, master.conf))
- logInfo("Started Master web UI at http://%s:%d".format(host, boundPort))
- } catch {
- case e: Exception =>
- logError("Failed to create Master web UI", e)
- System.exit(1)
- }
+ /** Attach a reconstructed UI to this Master UI. Only valid after bind(). */
+ def attachSparkUI(ui: SparkUI) {
+ assert(serverInfo.isDefined, "Master UI must be bound to a server before attaching SparkUIs")
+ ui.getHandlers.foreach(attachHandler)
}
+ /** Detach a reconstructed UI from this Master UI. Only valid after bind(). */
+ def detachSparkUI(ui: SparkUI) {
+ assert(serverInfo.isDefined, "Master UI must be bound to a server before detaching SparkUIs")
+ ui.getHandlers.foreach(detachHandler)
+ }
}
private[spark] object MasterWebUI {
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
index bf5a8d0..52c164c 100755
--- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -128,8 +128,8 @@ private[spark] class Worker(
host, port, cores, Utils.megabytesToString(memory)))
logInfo("Spark home: " + sparkHome)
createWorkDir()
- webUi = new WorkerWebUI(this, workDir, Some(webUiPort))
context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
+ webUi = new WorkerWebUI(this, workDir, Some(webUiPort))
webUi.bind()
registerWithMaster()
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/worker/ui/IndexPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ui/IndexPage.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ui/IndexPage.scala
deleted file mode 100644
index 49c1009..0000000
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ui/IndexPage.scala
+++ /dev/null
@@ -1,165 +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.worker.ui
-
-import scala.concurrent.Await
-import scala.xml.Node
-
-import akka.pattern.ask
-import javax.servlet.http.HttpServletRequest
-import org.json4s.JValue
-
-import org.apache.spark.deploy.JsonProtocol
-import org.apache.spark.deploy.DeployMessages.{RequestWorkerState, WorkerStateResponse}
-import org.apache.spark.deploy.master.DriverState
-import org.apache.spark.deploy.worker.{DriverRunner, ExecutorRunner}
-import org.apache.spark.ui.UIUtils
-import org.apache.spark.util.Utils
-
-private[spark] class IndexPage(parent: WorkerWebUI) {
- val workerActor = parent.worker.self
- val worker = parent.worker
- val timeout = parent.timeout
-
- def renderJson(request: HttpServletRequest): JValue = {
- val stateFuture = (workerActor ? RequestWorkerState)(timeout).mapTo[WorkerStateResponse]
- val workerState = Await.result(stateFuture, timeout)
- JsonProtocol.writeWorkerState(workerState)
- }
-
- def render(request: HttpServletRequest): Seq[Node] = {
- val stateFuture = (workerActor ? RequestWorkerState)(timeout).mapTo[WorkerStateResponse]
- val workerState = Await.result(stateFuture, timeout)
-
- val executorHeaders = Seq("ExecutorID", "Cores", "Memory", "Job Details", "Logs")
- val runningExecutorTable =
- UIUtils.listingTable(executorHeaders, executorRow, workerState.executors)
- val finishedExecutorTable =
- UIUtils.listingTable(executorHeaders, executorRow, workerState.finishedExecutors)
-
- val driverHeaders = Seq("DriverID", "Main Class", "State", "Cores", "Memory", "Logs", "Notes")
- val runningDrivers = workerState.drivers.sortBy(_.driverId).reverse
- val runningDriverTable = UIUtils.listingTable(driverHeaders, driverRow, runningDrivers)
- val finishedDrivers = workerState.finishedDrivers.sortBy(_.driverId).reverse
- def finishedDriverTable = UIUtils.listingTable(driverHeaders, driverRow, finishedDrivers)
-
- // For now we only show driver information if the user has submitted drivers to the cluster.
- // This is until we integrate the notion of drivers and applications in the UI.
- def hasDrivers = runningDrivers.length > 0 || finishedDrivers.length > 0
-
- val content =
- <div class="row-fluid"> <!-- Worker Details -->
- <div class="span12">
- <ul class="unstyled">
- <li><strong>ID:</strong> {workerState.workerId}</li>
- <li><strong>
- Master URL:</strong> {workerState.masterUrl}
- </li>
- <li><strong>Cores:</strong> {workerState.cores} ({workerState.coresUsed} Used)</li>
- <li><strong>Memory:</strong> {Utils.megabytesToString(workerState.memory)}
- ({Utils.megabytesToString(workerState.memoryUsed)} Used)</li>
- </ul>
- <p><a href={workerState.masterWebUiUrl}>Back to Master</a></p>
- </div>
- </div>
-
- <div class="row-fluid"> <!-- Running Executors -->
- <div class="span12">
- <h4> Running Executors {workerState.executors.size} </h4>
- {runningExecutorTable}
- </div>
- </div>
- // scalastyle:off
- <div>
- {if (hasDrivers)
- <div class="row-fluid"> <!-- Running Drivers -->
- <div class="span12">
- <h4> Running Drivers {workerState.drivers.size} </h4>
- {runningDriverTable}
- </div>
- </div>
- }
- </div>
-
- <div class="row-fluid"> <!-- Finished Executors -->
- <div class="span12">
- <h4> Finished Executors </h4>
- {finishedExecutorTable}
- </div>
- </div>
-
- <div>
- {if (hasDrivers)
- <div class="row-fluid"> <!-- Finished Drivers -->
- <div class="span12">
- <h4> Finished Drivers </h4>
- {finishedDriverTable}
- </div>
- </div>
- }
- </div>;
- // scalastyle:on
- UIUtils.basicSparkPage(content, "Spark Worker at %s:%s".format(
- workerState.host, workerState.port))
- }
-
- def executorRow(executor: ExecutorRunner): Seq[Node] = {
- <tr>
- <td>{executor.execId}</td>
- <td>{executor.cores}</td>
- <td sorttable_customkey={executor.memory.toString}>
- {Utils.megabytesToString(executor.memory)}
- </td>
- <td>
- <ul class="unstyled">
- <li><strong>ID:</strong> {executor.appId}</li>
- <li><strong>Name:</strong> {executor.appDesc.name}</li>
- <li><strong>User:</strong> {executor.appDesc.user}</li>
- </ul>
- </td>
- <td>
- <a href={"logPage?appId=%s&executorId=%s&logType=stdout"
- .format(executor.appId, executor.execId)}>stdout</a>
- <a href={"logPage?appId=%s&executorId=%s&logType=stderr"
- .format(executor.appId, executor.execId)}>stderr</a>
- </td>
- </tr>
-
- }
-
- def driverRow(driver: DriverRunner): Seq[Node] = {
- <tr>
- <td>{driver.driverId}</td>
- <td>{driver.driverDesc.command.arguments(1)}</td>
- <td>{driver.finalState.getOrElse(DriverState.RUNNING)}</td>
- <td sorttable_customkey={driver.driverDesc.cores.toString}>
- {driver.driverDesc.cores.toString}
- </td>
- <td sorttable_customkey={driver.driverDesc.mem.toString}>
- {Utils.megabytesToString(driver.driverDesc.mem)}
- </td>
- <td>
- <a href={s"logPage?driverId=${driver.driverId}&logType=stdout"}>stdout</a>
- <a href={s"logPage?driverId=${driver.driverId}&logType=stderr"}>stderr</a>
- </td>
- <td>
- {driver.finalException.getOrElse("")}
- </td>
- </tr>
- }
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/worker/ui/LogPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ui/LogPage.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ui/LogPage.scala
new file mode 100644
index 0000000..fec1207
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ui/LogPage.scala
@@ -0,0 +1,147 @@
+/*
+ * 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.worker.ui
+
+import java.io.File
+import javax.servlet.http.HttpServletRequest
+
+import scala.xml.Node
+
+import org.apache.spark.ui.{WebUIPage, UIUtils}
+import org.apache.spark.util.Utils
+
+private[spark] class LogPage(parent: WorkerWebUI) extends WebUIPage("logPage") {
+ private val worker = parent.worker
+ private val workDir = parent.workDir
+
+ def renderLog(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)
+ }
+
+ def render(request: HttpServletRequest): Seq[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 = <node>{Utils.offsetBytes(path, startByte, endByte)}</node>
+ val linkToMaster = <p><a href={worker.activeMasterWebUiUrl}>Back to Master</a></p>
+ val range = <span>Bytes {startByte.toString} - {endByte.toString} of {logLength}</span>
+
+ val backButton =
+ if (startByte > 0) {
+ <a href={"?%s&logType=%s&offset=%s&byteLength=%s"
+ .format(params, logType, math.max(startByte - byteLength, 0), byteLength)}>
+ <button type="button" class="btn btn-default">
+ Previous {Utils.bytesToString(math.min(byteLength, startByte))}
+ </button>
+ </a>
+ }
+ else {
+ <button type="button" class="btn btn-default" disabled="disabled">
+ Previous 0 B
+ </button>
+ }
+
+ val nextButton =
+ if (endByte < logLength) {
+ <a href={"?%s&logType=%s&offset=%s&byteLength=%s".
+ format(params, logType, endByte, byteLength)}>
+ <button type="button" class="btn btn-default">
+ Next {Utils.bytesToString(math.min(byteLength, logLength - endByte))}
+ </button>
+ </a>
+ }
+ else {
+ <button type="button" class="btn btn-default" disabled="disabled">
+ Next 0 B
+ </button>
+ }
+
+ val content =
+ <html>
+ <body>
+ {linkToMaster}
+ <div>
+ <div style="float:left; margin-right:10px">{backButton}</div>
+ <div style="float:left;">{range}</div>
+ <div style="float:right; margin-left:10px">{nextButton}</div>
+ </div>
+ <br />
+ <div style="height:500px; overflow:auto; padding:5px;">
+ <pre>{logText}</pre>
+ </div>
+ </body>
+ </html>
+ 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)
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
new file mode 100644
index 0000000..d451311
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
@@ -0,0 +1,165 @@
+/*
+ * 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.worker.ui
+
+import scala.concurrent.Await
+import scala.xml.Node
+
+import akka.pattern.ask
+import javax.servlet.http.HttpServletRequest
+import org.json4s.JValue
+
+import org.apache.spark.deploy.JsonProtocol
+import org.apache.spark.deploy.DeployMessages.{RequestWorkerState, WorkerStateResponse}
+import org.apache.spark.deploy.master.DriverState
+import org.apache.spark.deploy.worker.{DriverRunner, ExecutorRunner}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
+import org.apache.spark.util.Utils
+
+private[spark] class WorkerPage(parent: WorkerWebUI) extends WebUIPage("") {
+ val workerActor = parent.worker.self
+ val worker = parent.worker
+ val timeout = parent.timeout
+
+ override def renderJson(request: HttpServletRequest): JValue = {
+ val stateFuture = (workerActor ? RequestWorkerState)(timeout).mapTo[WorkerStateResponse]
+ val workerState = Await.result(stateFuture, timeout)
+ JsonProtocol.writeWorkerState(workerState)
+ }
+
+ def render(request: HttpServletRequest): Seq[Node] = {
+ val stateFuture = (workerActor ? RequestWorkerState)(timeout).mapTo[WorkerStateResponse]
+ val workerState = Await.result(stateFuture, timeout)
+
+ val executorHeaders = Seq("ExecutorID", "Cores", "Memory", "Job Details", "Logs")
+ val runningExecutorTable =
+ UIUtils.listingTable(executorHeaders, executorRow, workerState.executors)
+ val finishedExecutorTable =
+ UIUtils.listingTable(executorHeaders, executorRow, workerState.finishedExecutors)
+
+ val driverHeaders = Seq("DriverID", "Main Class", "State", "Cores", "Memory", "Logs", "Notes")
+ val runningDrivers = workerState.drivers.sortBy(_.driverId).reverse
+ val runningDriverTable = UIUtils.listingTable(driverHeaders, driverRow, runningDrivers)
+ val finishedDrivers = workerState.finishedDrivers.sortBy(_.driverId).reverse
+ def finishedDriverTable = UIUtils.listingTable(driverHeaders, driverRow, finishedDrivers)
+
+ // For now we only show driver information if the user has submitted drivers to the cluster.
+ // This is until we integrate the notion of drivers and applications in the UI.
+ def hasDrivers = runningDrivers.length > 0 || finishedDrivers.length > 0
+
+ val content =
+ <div class="row-fluid"> <!-- Worker Details -->
+ <div class="span12">
+ <ul class="unstyled">
+ <li><strong>ID:</strong> {workerState.workerId}</li>
+ <li><strong>
+ Master URL:</strong> {workerState.masterUrl}
+ </li>
+ <li><strong>Cores:</strong> {workerState.cores} ({workerState.coresUsed} Used)</li>
+ <li><strong>Memory:</strong> {Utils.megabytesToString(workerState.memory)}
+ ({Utils.megabytesToString(workerState.memoryUsed)} Used)</li>
+ </ul>
+ <p><a href={workerState.masterWebUiUrl}>Back to Master</a></p>
+ </div>
+ </div>
+
+ <div class="row-fluid"> <!-- Running Executors -->
+ <div class="span12">
+ <h4> Running Executors {workerState.executors.size} </h4>
+ {runningExecutorTable}
+ </div>
+ </div>
+ // scalastyle:off
+ <div>
+ {if (hasDrivers)
+ <div class="row-fluid"> <!-- Running Drivers -->
+ <div class="span12">
+ <h4> Running Drivers {workerState.drivers.size} </h4>
+ {runningDriverTable}
+ </div>
+ </div>
+ }
+ </div>
+
+ <div class="row-fluid"> <!-- Finished Executors -->
+ <div class="span12">
+ <h4> Finished Executors </h4>
+ {finishedExecutorTable}
+ </div>
+ </div>
+
+ <div>
+ {if (hasDrivers)
+ <div class="row-fluid"> <!-- Finished Drivers -->
+ <div class="span12">
+ <h4> Finished Drivers </h4>
+ {finishedDriverTable}
+ </div>
+ </div>
+ }
+ </div>;
+ // scalastyle:on
+ UIUtils.basicSparkPage(content, "Spark Worker at %s:%s".format(
+ workerState.host, workerState.port))
+ }
+
+ def executorRow(executor: ExecutorRunner): Seq[Node] = {
+ <tr>
+ <td>{executor.execId}</td>
+ <td>{executor.cores}</td>
+ <td sorttable_customkey={executor.memory.toString}>
+ {Utils.megabytesToString(executor.memory)}
+ </td>
+ <td>
+ <ul class="unstyled">
+ <li><strong>ID:</strong> {executor.appId}</li>
+ <li><strong>Name:</strong> {executor.appDesc.name}</li>
+ <li><strong>User:</strong> {executor.appDesc.user}</li>
+ </ul>
+ </td>
+ <td>
+ <a href={"logPage?appId=%s&executorId=%s&logType=stdout"
+ .format(executor.appId, executor.execId)}>stdout</a>
+ <a href={"logPage?appId=%s&executorId=%s&logType=stderr"
+ .format(executor.appId, executor.execId)}>stderr</a>
+ </td>
+ </tr>
+
+ }
+
+ def driverRow(driver: DriverRunner): Seq[Node] = {
+ <tr>
+ <td>{driver.driverId}</td>
+ <td>{driver.driverDesc.command.arguments(1)}</td>
+ <td>{driver.finalState.getOrElse(DriverState.RUNNING)}</td>
+ <td sorttable_customkey={driver.driverDesc.cores.toString}>
+ {driver.driverDesc.cores.toString}
+ </td>
+ <td sorttable_customkey={driver.driverDesc.mem.toString}>
+ {Utils.megabytesToString(driver.driverDesc.mem)}
+ </td>
+ <td>
+ <a href={s"logPage?driverId=${driver.driverId}&logType=stdout"}>stdout</a>
+ <a href={s"logPage?driverId=${driver.driverId}&logType=stderr"}>stderr</a>
+ </td>
+ <td>
+ {driver.finalException.getOrElse("")}
+ </td>
+ </tr>
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerWebUI.scala
----------------------------------------------------------------------
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 5625a44..0ad2edb 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,174 +20,44 @@ 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.{Logging, SparkConf}
import org.apache.spark.deploy.worker.Worker
-import org.apache.spark.ui.{SparkUI, UIUtils, WebUI}
+import org.apache.spark.ui.{SparkUI, WebUI}
import org.apache.spark.ui.JettyUtils._
-import org.apache.spark.util.{AkkaUtils, Utils}
+import org.apache.spark.util.AkkaUtils
/**
* Web UI server for the standalone worker.
*/
private[spark]
-class WorkerWebUI(val worker: Worker, val workDir: File, requestedPort: Option[Int] = None)
- extends WebUI("WorkerWebUI") with Logging {
+class WorkerWebUI(
+ val worker: Worker,
+ val workDir: File,
+ port: Option[Int] = None)
+ extends WebUI(worker.securityMgr, WorkerWebUI.getUIPort(port, worker.conf), worker.conf)
+ with Logging {
val timeout = AkkaUtils.askTimeout(worker.conf)
- private val host = Utils.localHostName()
- private val port = requestedPort.getOrElse(
- worker.conf.getInt("worker.ui.port", WorkerWebUI.DEFAULT_PORT))
- private val indexPage = new IndexPage(this)
-
- 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)
- )
- }
-
- /** Bind to the HTTP server behind this web interface. */
- override def bind() {
- try {
- serverInfo = Some(startJettyServer("0.0.0.0", 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 web UI", e)
- System.exit(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 = <node>{Utils.offsetBytes(path, startByte, endByte)}</node>
- val linkToMaster = <p><a href={worker.activeMasterWebUiUrl}>Back to Master</a></p>
- val range = <span>Bytes {startByte.toString} - {endByte.toString} of {logLength}</span>
-
- val backButton =
- if (startByte > 0) {
- <a href={"?%s&logType=%s&offset=%s&byteLength=%s"
- .format(params, logType, math.max(startByte - byteLength, 0), byteLength)}>
- <button type="button" class="btn btn-default">
- Previous {Utils.bytesToString(math.min(byteLength, startByte))}
- </button>
- </a>
- }
- else {
- <button type="button" class="btn btn-default" disabled="disabled">
- Previous 0 B
- </button>
- }
-
- val nextButton =
- if (endByte < logLength) {
- <a href={"?%s&logType=%s&offset=%s&byteLength=%s".
- format(params, logType, endByte, byteLength)}>
- <button type="button" class="btn btn-default">
- Next {Utils.bytesToString(math.min(byteLength, logLength - endByte))}
- </button>
- </a>
- }
- else {
- <button type="button" class="btn btn-default" disabled="disabled">
- Next 0 B
- </button>
- }
-
- val content =
- <html>
- <body>
- {linkToMaster}
- <div>
- <div style="float:left; margin-right:10px">{backButton}</div>
- <div style="float:left;">{range}</div>
- <div style="float:right; margin-left:10px">{nextButton}</div>
- </div>
- <br />
- <div style="height:500px; overflow:auto; padding:5px;">
- <pre>{logText}</pre>
- </div>
- </body>
- </html>
- UIUtils.basicSparkPage(content, logType + " log page for " + appId)
+ initialize()
+
+ /** Initialize all components of the server. */
+ def initialize() {
+ val logPage = new LogPage(this)
+ attachPage(logPage)
+ attachPage(new WorkerPage(this))
+ attachHandler(createStaticHandler(WorkerWebUI.STATIC_RESOURCE_BASE, "/static"))
+ attachHandler(createServletHandler("/log",
+ (request: HttpServletRequest) => logPage.renderLog(request), worker.securityMgr))
+ worker.metricsSystem.getServletHandlers.foreach(attachHandler)
}
-
- /** 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)
- }
-
}
private[spark] object WorkerWebUI {
- val DEFAULT_PORT=8081
+ val DEFAULT_PORT = 8081
val STATIC_RESOURCE_BASE = SparkUI.STATIC_RESOURCE_DIR
+
+ def getUIPort(requestedPort: Option[Int], conf: SparkConf): Int = {
+ requestedPort.getOrElse(conf.getInt("worker.ui.port", WorkerWebUI.DEFAULT_PORT))
+ }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala b/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
index affda13..c100122 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
@@ -31,11 +31,11 @@ private[spark] class ApplicationEventListener extends SparkListener {
def applicationStarted = startTime != -1
- def applicationFinished = endTime != -1
+ def applicationCompleted = endTime != -1
def applicationDuration: Long = {
val difference = endTime - startTime
- if (applicationStarted && applicationFinished && difference > 0) difference else -1L
+ if (applicationStarted && applicationCompleted && difference > 0) difference else -1L
}
override def onApplicationStart(applicationStart: SparkListenerApplicationStart) {
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala b/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
index 07255aa..7ed3713 100644
--- a/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
+++ b/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
@@ -42,24 +42,22 @@ class StorageStatus(
def memRemaining : Long = maxMem - memUsed()
- def rddBlocks = blocks.flatMap {
- case (rdd: RDDBlockId, status) => Some(rdd, status)
- case _ => None
- }
+ def rddBlocks = blocks.collect { case (rdd: RDDBlockId, status) => (rdd, status) }
}
@DeveloperApi
private[spark]
class RDDInfo(
- val id: Int,
- val name: String,
- val numPartitions: Int,
- val storageLevel: StorageLevel) extends Ordered[RDDInfo] {
+ val id: Int,
+ val name: String,
+ val numPartitions: Int,
+ val storageLevel: StorageLevel)
+ extends Ordered[RDDInfo] {
var numCachedPartitions = 0
var memSize = 0L
var diskSize = 0L
- var tachyonSize= 0L
+ var tachyonSize = 0L
override def toString = {
import Utils.bytesToString
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
index dd0818e..62a4e3d 100644
--- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
@@ -121,6 +121,7 @@ private[spark] object JettyUtils extends Logging {
/** Create a handler for serving files from a static directory */
def createStaticHandler(resourceBase: String, path: String): ServletContextHandler = {
val contextHandler = new ServletContextHandler
+ contextHandler.setInitParameter("org.eclipse.jetty.servlet.Default.gzip", "false")
val staticHandler = new DefaultServlet
val holder = new ServletHolder(staticHandler)
Option(getClass.getClassLoader.getResource(resourceBase)) match {
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/ui/Page.scala
----------------------------------------------------------------------
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 b2a069a..0000000
--- 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
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/6aa08c39/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
----------------------------------------------------------------------
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 7fa4fd3..2fef1a6 100644
--- a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
@@ -17,112 +17,86 @@
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.util.Utils
+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.StorageTab
-/** Top level user interface for Spark */
+/**
+ * Top level user interface for a Spark application.
+ */
private[spark] class SparkUI(
val sc: SparkContext,
val conf: SparkConf,
+ val securityManager: SecurityManager,
val listenerBus: SparkListenerBus,
var appName: String,
val basePath: String = "")
- extends WebUI("SparkUI") with Logging {
+ extends WebUI(securityManager, SparkUI.getUIPort(conf), conf, basePath)
+ with Logging {
- def this(sc: SparkContext) = this(sc, sc.conf, sc.listenerBus, sc.appName)
- def this(listenerBus: SparkListenerBus, appName: String, basePath: String) =
- this(null, new SparkConf, listenerBus, appName, basePath)
+ 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, 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 localHost = Utils.localHostName()
- private val publicHost = Option(System.getenv("SPARK_PUBLIC_DNS")).getOrElse(localHost)
- private val port = conf.getInt("spark.ui.port", SparkUI.DEFAULT_PORT)
+ // Maintain executor storage status through Spark events
+ val storageStatusListener = new StorageStatusListener
- private val storage = new BlockManagerUI(this)
- private val jobs = new JobProgressUI(this)
- private val env = new EnvironmentUI(this)
- private val exec = new ExecutorsUI(this)
+ initialize()
- val handlers: Seq[ServletContextHandler] = {
- val metricsServletHandlers = if (live) {
- SparkEnv.get.metricsSystem.getServletHandlers
- } else {
- Array[ServletContextHandler]()
+ /** Initialize all components of the server. */
+ def initialize() {
+ listenerBus.addListener(storageStatusListener)
+ val jobProgressTab = new JobProgressTab(this)
+ attachTab(jobProgressTab)
+ attachTab(new StorageTab(this))
+ attachTab(new EnvironmentTab(this))
+ attachTab(new ExecutorsTab(this))
+ attachHandler(createStaticHandler(SparkUI.STATIC_RESOURCE_DIR, "/static"))
+ attachHandler(createRedirectHandler("/", "/stages", basePath = basePath))
+ attachHandler(
+ createRedirectHandler("/stages/stage/kill", "/stages", jobProgressTab.handleKillRequest))
+ 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 = basePath)
- )
}
- // Maintain executor storage status through Spark events
- val storageStatusListener = new StorageStatusListener
-
+ /** Set the app name for this UI. */
def setAppName(name: String) {
appName = name
}
- /** 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)
- }
-
- /** Bind to the HTTP server behind this web interface. */
- override def bind() {
- try {
- serverInfo = Some(startJettyServer("0.0.0.0", port, handlers, sc.conf))
- logInfo("Started Spark web UI at http://%s:%d".format(publicHost, boundPort))
- } catch {
- case e: Exception =>
- logError("Failed to create Spark web UI", e)
- System.exit(1)
- }
+ /** Register the given listener with the listener bus. */
+ def registerListener(listener: SparkListener) {
+ listenerBus.addListener(listener)
}
/** 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))
+ logInfo("Stopped Spark web UI at %s".format(appUIAddress))
}
/**
* Return the application UI host:port. This does not include the scheme (http://).
*/
- private[spark] def appUIHostPort = publicHost + ":" + boundPort
+ private[spark] def appUIHostPort = publicHostName + ":" + boundPort
private[spark] def appUIAddress = s"http://$appUIHostPort"
-
}
private[spark] object SparkUI {
val DEFAULT_PORT = 4040
val STATIC_RESOURCE_DIR = "org/apache/spark/ui/static"
+
+ def getUIPort(conf: SparkConf): Int = {
+ conf.getInt("spark.ui.port", SparkUI.DEFAULT_PORT)
+ }
}