Merge branch 'streaming-web-ui' of github.com:tdas/spark into streaming-web-ui

Conflicts:
	streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
	streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingTab.scala

Author: tdas

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -212,7 +212,6 @@ class SparkContext(config: SparkConf) extends Logging {
 
   // Initialize the Spark UI, registering all associated listeners
   private[spark] val ui = new SparkUI(this)
-  ui.start()
   ui.bind()
 
   // Optionally log Spark events

core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala

@@ -17,12 +17,9 @@
 
 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.scheduler._
@@ -45,15 +42,15 @@ import org.apache.spark.util.Utils
  */
 class HistoryServer(
     val baseLogDir: String,
+    securityManager: SecurityManager,
     conf: SparkConf)
-  extends WebUI(new SecurityManager(conf)) 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
 
   // A timestamp of when the disk was last accessed to check for log updates
   private var lastLogCheckTime = -1L
@@ -90,30 +87,20 @@ class HistoryServer(
   // 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 IndexPage(this))
     attachHandler(createStaticHandler(STATIC_RESOURCE_DIR, "/static"))
     logCheckingThread.start()
   }
 
-  /** Bind to the HTTP server behind this web interface. */
-  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.
@@ -179,12 +166,11 @@ class HistoryServer(
     val path = logDir.getPath
     val appId = path.getName
     val replayBus = new ReplayListenerBus(logInfo.logPaths, fileSystem, logInfo.compressionCodec)
-    val ui = new SparkUI(conf, 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)
@@ -267,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) }

core/src/main/scala/org/apache/spark/deploy/history/IndexPage.scala

@@ -21,9 +21,9 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
-import org.apache.spark.ui.{UIPage, UIUtils}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
 
-private[spark] class IndexPage(parent: HistoryServer) extends UIPage("") {
+private[spark] class IndexPage(parent: HistoryServer) extends WebUIPage("") {
 
   override def render(request: HttpServletRequest): Seq[Node] = {
     val appRows = parent.appIdToInfo.values.toSeq.sortBy { app => -app.lastUpdated }

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -118,7 +118,6 @@ private[spark] class Master(
     logInfo("Starting Spark master at " + masterUrl)
     // Listen for remote client disconnection events, since they don't go through Akka's watch()
     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
-    webUi.start()
     webUi.bind()
     masterWebUiUrl = "http://" + masterPublicAddress + ":" + webUi.boundPort
     context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut)
@@ -670,7 +669,6 @@ private[spark] class Master(
         val replayBus = new ReplayListenerBus(eventLogPaths, fileSystem, compressionCodec)
         val ui = new SparkUI(
           new SparkConf, replayBus, appName + " (completed)", "/history/" + app.id)
-        ui.start()
         replayBus.replay()
         app.desc.appUiUrl = ui.basePath
         appIdToUI(app.id) = ui

core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala

@@ -28,11 +28,11 @@ 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.{UIPage, UIUtils}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
 private[spark] class ApplicationPage(parent: MasterWebUI)
-  extends UIPage("app", includeJson = true) {
+  extends WebUIPage("app", includeJson = true) {
 
   private val master = parent.masterActorRef
   private val timeout = parent.timeout

core/src/main/scala/org/apache/spark/deploy/master/ui/IndexPage.scala

@@ -28,10 +28,10 @@ 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.{UIPage, UIUtils}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
-private[spark] class IndexPage(parent: MasterWebUI) extends UIPage("", includeJson = true) {
+private[spark] class IndexPage(parent: MasterWebUI) extends WebUIPage("", includeJson = true) {
   private val master = parent.masterActorRef
   private val timeout = parent.timeout
 

core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.deploy.master.ui
 
-import javax.servlet.http.HttpServletRequest
-
 import org.apache.spark.Logging
 import org.apache.spark.deploy.master.Master
 import org.apache.spark.ui.{SparkUI, WebUI}
@@ -30,34 +28,22 @@ import org.apache.spark.util.{AkkaUtils, Utils}
  */
 private[spark]
 class MasterWebUI(val master: Master, requestedPort: Int)
-  extends WebUI(master.securityMgr) with Logging {
+  extends WebUI(master.securityMgr, requestedPort, master.conf) with Logging {
 
-  private val host = Utils.localHostName()
-  private val port = requestedPort
   val masterActorRef = master.self
   val timeout = AkkaUtils.askTimeout(master.conf)
 
+  initialize()
+
   /** Initialize all components of the server. */
-  def start() {
+  def initialize() {
     attachPage(new ApplicationPage(this))
     attachPage(new IndexPage(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. */
-  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 attachUI(ui: SparkUI) {
     assert(serverInfo.isDefined, "Master UI must be bound to a server before attaching SparkUIs")

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -130,7 +130,6 @@ private[spark] class Worker(
     createWorkDir()
     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
     webUi = new WorkerWebUI(this, workDir, Some(webUiPort))
-    webUi.start()
     webUi.bind()
     registerWithMaster()
 

core/src/main/scala/org/apache/spark/deploy/worker/ui/IndexPage.scala

@@ -28,10 +28,10 @@ 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.{UIPage, UIUtils}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
-private[spark] class IndexPage(parent: WorkerWebUI) extends UIPage("", includeJson = true) {
+private[spark] class IndexPage(parent: WorkerWebUI) extends WebUIPage("", includeJson = true) {
   val workerActor = parent.worker.self
   val worker = parent.worker
   val timeout = parent.timeout

core/src/main/scala/org/apache/spark/deploy/worker/ui/LogPage.scala

@@ -22,10 +22,10 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
-import org.apache.spark.ui.{UIPage, UIUtils}
+import org.apache.spark.ui.{WebUIPage, UIUtils}
 import org.apache.spark.util.Utils
 
-private[spark] class LogPage(parent: WorkerWebUI) extends UIPage("logPage") {
+private[spark] class LogPage(parent: WorkerWebUI) extends WebUIPage("logPage") {
   private val worker = parent.worker
   private val workDir = parent.workDir