link application page to master page

Author: nblintao

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

@@ -2157,7 +2157,8 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     // Note: this code assumes that the task scheduler has been initialized and has contacted
     // the cluster manager to get an application ID (in case the cluster manager provides one).
     listenerBus.post(SparkListenerApplicationStart(appName, Some(applicationId),
-      startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls))
+      startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls,
+      schedulerBackend.getMasterWebUiUrl))
   }
 
   /** Post the application end event */

core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala

@@ -127,7 +127,8 @@ private[deploy] object DeployMessages {
 
   // Master to AppClient
 
-  case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
+  case class RegisteredApplication(appId: String, master: RpcEndpointRef,
+    masterWebUiUrl: String) extends DeployMessage
 
   // TODO(matei): replace hostPort with host
   case class ExecutorAdded(id: Int, workerId: String, hostPort: String, cores: Int, memory: Int) {
@@ -176,6 +177,7 @@ private[deploy] object DeployMessages {
       host: String,
       port: Int,
       restPort: Option[Int],
+      webUiUrl: String,
       workers: Array[WorkerInfo],
       activeApps: Array[ApplicationInfo],
       completedApps: Array[ApplicationInfo],

core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala

@@ -156,7 +156,7 @@ private[spark] class StandaloneAppClient(
     }
 
     override def receive: PartialFunction[Any, Unit] = {
-      case RegisteredApplication(appId_, masterRef) =>
+      case RegisteredApplication(appId_, masterRef, masterWebUiUrl) =>
         // FIXME How to handle the following cases?
         // 1. A master receives multiple registrations and sends back multiple
         // RegisteredApplications due to an unstable network.
@@ -166,6 +166,7 @@ private[spark] class StandaloneAppClient(
         registered.set(true)
         master = Some(masterRef)
         listener.connected(appId.get)
+        listener.masterChanged(masterWebUiUrl)
 
       case ApplicationRemoved(message) =>
         markDead("Master removed our application: %s".format(message))
@@ -188,6 +189,7 @@ private[spark] class StandaloneAppClient(
       case MasterChanged(masterRef, masterWebUiUrl) =>
         logInfo("Master has changed, new master is at " + masterRef.address.toSparkURL)
         master = Some(masterRef)
+        listener.masterChanged(masterWebUiUrl)
         alreadyDisconnected = false
         masterRef.send(MasterChangeAcknowledged(appId.get))
     }

core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClientListener.scala

@@ -37,4 +37,6 @@ private[spark] trait StandaloneAppClientListener {
       fullId: String, workerId: String, hostPort: String, cores: Int, memory: Int): Unit
 
   def executorRemoved(fullId: String, message: String, exitStatus: Option[Int]): Unit
+
+  def masterChanged(masterWebUiUrl: String): Unit
 }

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

@@ -233,7 +233,7 @@ private[deploy] class Master(
         registerApplication(app)
         logInfo("Registered app " + description.name + " with ID " + app.id)
         persistenceEngine.addApplication(app)
-        driver.send(RegisteredApplication(app.id, self))
+        driver.send(RegisteredApplication(app.id, self, masterWebUiUrl))
         schedule()
       }
 
@@ -471,7 +471,7 @@ private[deploy] class Master(
 
     case RequestMasterState =>
       context.reply(MasterStateResponse(
-        address.host, address.port, restServerBoundPort,
+        address.host, address.port, restServerBoundPort, masterWebUiUrl,
         workers.toArray, apps.toArray, completedApps.toArray,
         drivers.toArray, completedDrivers.toArray, state))
 

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

@@ -35,6 +35,7 @@ private[ui] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app")
   def render(request: HttpServletRequest): Seq[Node] = {
     val appId = request.getParameter("appId")
     val state = master.askWithRetry[MasterStateResponse](RequestMasterState)
+    val masterWebUiUrl = state.webUiUrl
     val app = state.activeApps.find(_.id == appId)
       .getOrElse(state.completedApps.find(_.id == appId).orNull)
     if (app == null) {
@@ -81,6 +82,7 @@ private[ui] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app")
               }
             }
           </ul>
+          <p><a href={masterWebUiUrl}>Back to Master</a></p>
         </div>
       </div>
 

core/src/main/scala/org/apache/spark/scheduler/SchedulerBackend.scala

@@ -56,4 +56,6 @@ private[spark] trait SchedulerBackend {
    */
   def getDriverLogUrls: Option[Map[String, String]] = None
 
+  def getMasterWebUiUrl: Option[String] = None
+
 }

core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala

@@ -126,7 +126,8 @@ case class SparkListenerApplicationStart(
     time: Long,
     sparkUser: String,
     appAttemptId: Option[String],
-    driverLogs: Option[Map[String, String]] = None) extends SparkListenerEvent
+    driverLogs: Option[Map[String, String]] = None,
+    masterWebUiUrl: Option[String] = None) extends SparkListenerEvent
 
 @DeveloperApi
 case class SparkListenerApplicationEnd(time: Long) extends SparkListenerEvent

core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala

@@ -157,6 +157,16 @@ private[spark] class StandaloneSchedulerBackend(
     removeExecutor(fullId.split("/")(1), reason)
   }
 
+  var masterWebUiUrl: Option[String] = None
+
+  override def masterChanged(webUiUrl: String): Unit = {
+    masterWebUiUrl = Some(webUiUrl)
+  }
+
+  override def getMasterWebUiUrl(): Option[String] = {
+    masterWebUiUrl
+  }
+
   override def sufficientResourcesRegistered(): Boolean = {
     totalCoreCount.get() >= totalExpectedCores * minRegisteredRatio
   }

core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala

@@ -54,6 +54,10 @@ private[ui] class ExecutorsPage(
   private val GCTimePercent = 0.1
 
   def render(request: HttpServletRequest): Seq[Node] = {
+    val masterWebUiUrl = listener.masterWebUiUrl
+    val backLink = masterWebUiUrl.map(
+      link => <p><a href={link}>Back to Master</a></p>
+    ).getOrElse(Seq.empty)
     val content =
       <div>
         {
@@ -63,7 +67,7 @@ private[ui] class ExecutorsPage(
         }
       </div>;
 
-    UIUtils.headerSparkPage("Executors", content, parent, useDataTables = true)
+    UIUtils.headerSparkPage("Executors", backLink++content, parent, useDataTables = true)
   }
 }