[SPARK-20646][core] Port executors page to new UI backend.

The executors page is built on top of the REST API, so the page itself
was easy to hook up to the new code.

Some other pages depend on the `ExecutorListener` class that is being
removed, though, so they needed to be modified to use data from the
new store. Fortunately, all they seemed to need is the map of executor
logs, so that was somewhat easy too.

The executor timeline graph required adding some properties to the
ExecutorSummary API type. Instead of following the previous code,
which stored all the listener events in memory, the timeline is
now created based on the data available from the API.

I had to change some of the test golden files because the old code would
return executors in "random" order (since it used a mutable Map instead
of something that returns a sorted list), and the new code returns executors
in id order.

Tested with existing unit tests.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

@@ -88,6 +88,15 @@ private[spark] class AppStatusListener(
     kvstore.write(new ApplicationInfoWrapper(appInfo))
   }
 
+  override def onEnvironmentUpdate(event: SparkListenerEnvironmentUpdate): Unit = {
+    val details = event.environmentDetails
+
+    coresPerTask = details.getOrElse("Spark Properties", Nil).toMap
+      .get("spark.task.cpus")
+      .map(_.toInt)
+      .getOrElse(coresPerTask)
+  }
+
   override def onApplicationEnd(event: SparkListenerApplicationEnd): Unit = {
     val old = appInfo.attempts.head
     val attempt = new v1.ApplicationAttemptInfo(
@@ -114,7 +123,7 @@ private[spark] class AppStatusListener(
   override def onExecutorAdded(event: SparkListenerExecutorAdded): Unit = {
     // This needs to be an update in case an executor re-registers after the driver has
     // marked it as "dead".
-    val exec = getOrCreateExecutor(event.executorId)
+    val exec = getOrCreateExecutor(event.executorId, event.time)
     exec.host = event.executorInfo.executorHost
     exec.isActive = true
     exec.totalCores = event.executorInfo.totalCores
@@ -126,6 +135,8 @@ private[spark] class AppStatusListener(
   override def onExecutorRemoved(event: SparkListenerExecutorRemoved): Unit = {
     liveExecutors.remove(event.executorId).foreach { exec =>
       exec.isActive = false
+      exec.removeTime = new Date(event.time)
+      exec.removeReason = event.reason
       update(exec, System.nanoTime())
     }
   }
@@ -339,18 +350,25 @@ private[spark] class AppStatusListener(
     }
 
     liveExecutors.get(event.taskInfo.executorId).foreach { exec =>
-      if (event.taskMetrics != null) {
-        val readMetrics = event.taskMetrics.shuffleReadMetrics
-        exec.totalGcTime += event.taskMetrics.jvmGCTime
-        exec.totalInputBytes += event.taskMetrics.inputMetrics.bytesRead
-        exec.totalShuffleRead += readMetrics.localBytesRead + readMetrics.remoteBytesRead
-        exec.totalShuffleWrite += event.taskMetrics.shuffleWriteMetrics.bytesWritten
-      }
-
       exec.activeTasks -= 1
       exec.completedTasks += completedDelta
       exec.failedTasks += failedDelta
       exec.totalDuration += event.taskInfo.duration
+
+      // Note: For resubmitted tasks, we continue to use the metrics that belong to the
+      // first attempt of this task. This may not be 100% accurate because the first attempt
+      // could have failed half-way through. The correct fix would be to keep track of the
+      // metrics added by each attempt, but this is much more complicated.
+      if (event.reason != Resubmitted) {
+        if (event.taskMetrics != null) {
+          val readMetrics = event.taskMetrics.shuffleReadMetrics
+          exec.totalGcTime += event.taskMetrics.jvmGCTime
+          exec.totalInputBytes += event.taskMetrics.inputMetrics.bytesRead
+          exec.totalShuffleRead += readMetrics.localBytesRead + readMetrics.remoteBytesRead
+          exec.totalShuffleWrite += event.taskMetrics.shuffleWriteMetrics.bytesWritten
+        }
+      }
+
       maybeUpdate(exec, now)
     }
   }
@@ -391,7 +409,7 @@ private[spark] class AppStatusListener(
   override def onBlockManagerAdded(event: SparkListenerBlockManagerAdded): Unit = {
     // This needs to set fields that are already set by onExecutorAdded because the driver is
     // considered an "executor" in the UI, but does not have a SparkListenerExecutorAdded event.
-    val exec = getOrCreateExecutor(event.blockManagerId.executorId)
+    val exec = getOrCreateExecutor(event.blockManagerId.executorId, event.time)
     exec.hostPort = event.blockManagerId.hostPort
     event.maxOnHeapMem.foreach { _ =>
       exec.totalOnHeap = event.maxOnHeapMem.get
@@ -543,8 +561,8 @@ private[spark] class AppStatusListener(
     }
   }
 
-  private def getOrCreateExecutor(executorId: String): LiveExecutor = {
-    liveExecutors.getOrElseUpdate(executorId, new LiveExecutor(executorId))
+  private def getOrCreateExecutor(executorId: String, addTime: Long): LiveExecutor = {
+    liveExecutors.getOrElseUpdate(executorId, new LiveExecutor(executorId, addTime))
   }
 
   private def getOrCreateStage(info: StageInfo): LiveStage = {

core/src/main/scala/org/apache/spark/status/AppStatusStore.scala

@@ -47,8 +47,22 @@ private[spark] class AppStatusStore(store: KVStore) {
   }
 
   def executorList(activeOnly: Boolean): Seq[v1.ExecutorSummary] = {
-    store.view(classOf[ExecutorSummaryWrapper]).index("active").reverse().first(true)
-      .last(true).asScala.map(_.info).toSeq
+    val base = store.view(classOf[ExecutorSummaryWrapper])
+    val filtered = if (activeOnly) {
+      base.index("active").reverse().first(true).last(true)
+    } else {
+      base
+    }
+    filtered.asScala.map(_.info).toSeq
+  }
+
+  def executorSummary(executorId: String): Option[v1.ExecutorSummary] = {
+    try {
+      Some(store.read(classOf[ExecutorSummaryWrapper], executorId).info)
+    } catch {
+      case _: NoSuchElementException =>
+        None
+    }
   }
 
   def stageList(statuses: JList[v1.StageStatus]): Seq[v1.StageData] = {

core/src/main/scala/org/apache/spark/status/LiveEntity.scala

@@ -212,13 +212,17 @@ private class LiveTask(
 
 }
 
-private class LiveExecutor(val executorId: String) extends LiveEntity {
+private class LiveExecutor(val executorId: String, _addTime: Long) extends LiveEntity {
 
   var hostPort: String = null
   var host: String = null
   var isActive = true
   var totalCores = 0
 
+  val addTime = new Date(_addTime)
+  var removeTime: Date = null
+  var removeReason: String = null
+
   var rddBlocks = 0
   var memoryUsed = 0L
   var diskUsed = 0L
@@ -276,6 +280,9 @@ private class LiveExecutor(val executorId: String) extends LiveEntity {
       totalShuffleWrite,
       isBlacklisted,
       maxMemory,
+      addTime,
+      Option(removeTime),
+      Option(removeReason),
       executorLogs,
       memoryMetrics)
     new ExecutorSummaryWrapper(info)

core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala

@@ -20,22 +20,11 @@ import javax.ws.rs.{GET, Produces}
 import javax.ws.rs.core.MediaType
 
 import org.apache.spark.ui.SparkUI
-import org.apache.spark.ui.exec.ExecutorsPage
 
 @Produces(Array(MediaType.APPLICATION_JSON))
 private[v1] class AllExecutorListResource(ui: SparkUI) {
 
   @GET
-  def executorList(): Seq[ExecutorSummary] = {
-    val listener = ui.executorsListener
-    listener.synchronized {
-      // The follow codes should be protected by `listener` to make sure no executors will be
-      // removed before we query their status. See SPARK-12784.
-      (0 until listener.activeStorageStatusList.size).map { statusId =>
-        ExecutorsPage.getExecInfo(listener, statusId, isActive = true)
-      } ++ (0 until listener.deadStorageStatusList.size).map { statusId =>
-        ExecutorsPage.getExecInfo(listener, statusId, isActive = false)
-      }
-    }
-  }
+  def executorList(): Seq[ExecutorSummary] = ui.store.executorList(false)
+
 }

core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala

@@ -20,21 +20,11 @@ import javax.ws.rs.{GET, Produces}
 import javax.ws.rs.core.MediaType
 
 import org.apache.spark.ui.SparkUI
-import org.apache.spark.ui.exec.ExecutorsPage
 
 @Produces(Array(MediaType.APPLICATION_JSON))
 private[v1] class ExecutorListResource(ui: SparkUI) {
 
   @GET
-  def executorList(): Seq[ExecutorSummary] = {
-    val listener = ui.executorsListener
-    listener.synchronized {
-      // The follow codes should be protected by `listener` to make sure no executors will be
-      // removed before we query their status. See SPARK-12784.
-      val storageStatusList = listener.activeStorageStatusList
-      (0 until storageStatusList.size).map { statusId =>
-        ExecutorsPage.getExecInfo(listener, statusId, isActive = true)
-      }
-    }
-  }
+  def executorList(): Seq[ExecutorSummary] = ui.store.executorList(true)
+
 }

core/src/main/scala/org/apache/spark/status/api/v1/api.scala

@@ -85,6 +85,9 @@ class ExecutorSummary private[spark](
     val totalShuffleWrite: Long,
     val isBlacklisted: Boolean,
     val maxMemory: Long,
+    val addTime: Date,
+    val removeTime: Option[Date],
+    val removeReason: Option[String],
     val executorLogs: Map[String, String],
     val memoryMetrics: Option[MemoryMetrics])
 

core/src/main/scala/org/apache/spark/ui/SparkUI.scala

@@ -30,7 +30,7 @@ import org.apache.spark.status.api.v1.{ApiRootResource, ApplicationAttemptInfo,
 import org.apache.spark.storage.StorageStatusListener
 import org.apache.spark.ui.JettyUtils._
 import org.apache.spark.ui.env.{EnvironmentListener, EnvironmentTab}
-import org.apache.spark.ui.exec.{ExecutorsListener, ExecutorsTab}
+import org.apache.spark.ui.exec.ExecutorsTab
 import org.apache.spark.ui.jobs.{JobProgressListener, JobsTab, StagesTab}
 import org.apache.spark.ui.scope.RDDOperationGraphListener
 import org.apache.spark.ui.storage.{StorageListener, StorageTab}
@@ -46,7 +46,6 @@ private[spark] class SparkUI private (
     securityManager: SecurityManager,
     val environmentListener: EnvironmentListener,
     val storageStatusListener: StorageStatusListener,
-    val executorsListener: ExecutorsListener,
     val jobProgressListener: JobProgressListener,
     val storageListener: StorageListener,
     val operationGraphListener: RDDOperationGraphListener,
@@ -70,7 +69,7 @@ private[spark] class SparkUI private (
   def initialize() {
     val jobsTab = new JobsTab(this)
     attachTab(jobsTab)
-    val stagesTab = new StagesTab(this)
+    val stagesTab = new StagesTab(this, store)
     attachTab(stagesTab)
     attachTab(new StorageTab(this))
     attachTab(new EnvironmentTab(this))
@@ -186,19 +185,17 @@ private[spark] object SparkUI {
     }
     val environmentListener = new EnvironmentListener
     val storageStatusListener = new StorageStatusListener(conf)
-    val executorsListener = new ExecutorsListener(storageStatusListener, conf)
     val storageListener = new StorageListener(storageStatusListener)
     val operationGraphListener = new RDDOperationGraphListener(conf)
 
     addListenerFn(environmentListener)
     addListenerFn(storageStatusListener)
-    addListenerFn(executorsListener)
     addListenerFn(storageListener)
     addListenerFn(operationGraphListener)
 
     new SparkUI(store, sc, conf, securityManager, environmentListener, storageStatusListener,
-      executorsListener, jobProgressListener, storageListener, operationGraphListener,
-      appName, basePath, lastUpdateTime, startTime, appSparkVersion)
+      jobProgressListener, storageListener, operationGraphListener, appName, basePath,
+      lastUpdateTime, startTime, appSparkVersion)
   }
 
 }

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

@@ -22,11 +22,12 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.{Node, Text}
 
-import org.apache.spark.ui.{UIUtils, WebUIPage}
+import org.apache.spark.SparkContext
+import org.apache.spark.ui.{SparkUITab, UIUtils, WebUIPage}
 
-private[ui] class ExecutorThreadDumpPage(parent: ExecutorsTab) extends WebUIPage("threadDump") {
-
-  private val sc = parent.sc
+private[ui] class ExecutorThreadDumpPage(
+    parent: SparkUITab,
+    sc: Option[SparkContext]) extends WebUIPage("threadDump") {
 
   // stripXSS is called first to remove suspicious characters used in XSS attacks
   def render(request: HttpServletRequest): Seq[Node] = {