SHS-NG M4.2: Port executors page to new 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 some extra code to save the
executor-related events in the UI store. This just implements the
existing functionality, without making any changes related to efficiency
or scalability of that graph.

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.

The static files are still kept in the core/ resources directory since
Jetty's default servlet does not handle fetching static files from multiple
jars.

TODO: add unit tests for the new ExecutorSummary fields being populated.

Author: Marcelo Vanzin

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

@@ -17,6 +17,10 @@
 
 package org.apache.spark.scheduler.cluster
 
+import scala.annotation.meta.getter
+
+import com.fasterxml.jackson.annotation.JsonIgnore
+
 import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef}
 
 /**
@@ -29,9 +33,12 @@ import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef}
  * @param totalCores The total number of cores available to the executor
  */
 private[cluster] class ExecutorData(
+   @(JsonIgnore @getter)
    val executorEndpoint: RpcEndpointRef,
+   @(JsonIgnore @getter)
    val executorAddress: RpcAddress,
    override val executorHost: String,
+   @(JsonIgnore @getter)
    var freeCores: Int,
    override val totalCores: Int,
    override val logUrlMap: Map[String, String]

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

@@ -54,6 +54,9 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
 
   private var appId: String = null
   private var activeJobs: Set[Int] = Set()
+  private var executorEventId: Long = 0L
+
+  private var coresPerTask: Int = 1
 
   override def onApplicationStart(event: SparkListenerApplicationStart): Unit = {
     assert(event.appId.isDefined, "Application without IDs are not supported.")
@@ -96,6 +99,8 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
       details("System Properties"),
       details("Classpath Entries"))
 
+    coresPerTask = envInfo.sparkProperties.toMap.get("spark.task.cpus").map(_.toInt)
+      .getOrElse(coresPerTask)
     kvstore.write(new ApplicationEnvironmentInfoWrapper(envInfo))
   }
 
@@ -129,9 +134,12 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
         hostPort = event.executorInfo.executorHost,
         isActive = true,
         totalCores = event.executorInfo.totalCores,
+        maxTasks = event.executorInfo.totalCores / coresPerTask,
         executorLogs = event.executorInfo.logUrlMap)
       new ExecutorSummaryWrapper(newInfo)
     }
+
+    writeExecutorEvent(event)
   }
 
   override def onExecutorRemoved(event: SparkListenerExecutorRemoved): Unit = {
@@ -140,12 +148,19 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
         isActive = false)
       new ExecutorSummaryWrapper(newInfo)
     }
+
+    writeExecutorEvent(event)
   }
 
   override def onExecutorBlacklisted(event: SparkListenerExecutorBlacklisted): Unit = {
     updateBlackListStatus(event.executorId, true)
   }
 
+  private def writeExecutorEvent(event: SparkListenerEvent): Unit = {
+    executorEventId += 1
+    kvstore.write(new ExecutorEventData(executorEventId, event))
+  }
+
   override def onExecutorUnblacklisted(event: SparkListenerExecutorUnblacklisted): Unit = {
     updateBlackListStatus(event.executorId, false)
   }
@@ -342,6 +357,14 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
         numActiveTasks = stage.info.numActiveTasks + 1)
       newStageDataWrapper(stage, newInfo)
     }
+
+    updateExecutorSummary(event.taskInfo.executorId) { uiexec =>
+      val updated = newExecutorSummary(
+        uiexec.info,
+        activeTasks = uiexec.info.activeTasks + 1,
+        totalTasks = uiexec.info.totalTasks + 1)
+      new ExecutorSummaryWrapper(updated)
+    }
   }
 
   override def onTaskGettingResult(event: SparkListenerTaskGettingResult): Unit = {
@@ -440,6 +463,7 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
       newExecutorStageSummary(
         exec,
         failedTasks = exec.failedTasks + failedDelta,
+        succeededTasks = exec.succeededTasks + completedDelta,
         inputBytes = exec.inputBytes + metricsDelta.inputMetrics.bytesRead,
         outputBytes = exec.outputBytes + metricsDelta.outputMetrics.bytesWritten,
         shuffleRead = exec.shuffleRead + metricsDelta.shuffleReadMetrics.localBytesRead +
@@ -448,6 +472,30 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
         memoryBytesSpilled = exec.memoryBytesSpilled + metricsDelta.memoryBytesSpilled,
         diskBytesSpilled = exec.diskBytesSpilled + metricsDelta.diskBytesSpilled)
     }
+
+    updateExecutorSummary(event.taskInfo.executorId) { uiexec =>
+      val (gcTime, inputBytes, shuffleRead, shuffleWrite) = if (event.taskMetrics != null) {
+        val readMetrics = event.taskMetrics.shuffleReadMetrics
+        (event.taskMetrics.jvmGCTime,
+          event.taskMetrics.inputMetrics.bytesRead,
+          readMetrics.localBytesRead + readMetrics.remoteBytesRead,
+          event.taskMetrics.shuffleWriteMetrics.bytesWritten)
+      } else {
+        (0L, 0L, 0L, 0L)
+      }
+
+      val updated = newExecutorSummary(
+        uiexec.info,
+        activeTasks = uiexec.info.activeTasks - 1,
+        completedTasks = uiexec.info.completedTasks + completedDelta,
+        failedTasks = uiexec.info.failedTasks + failedDelta,
+        totalDuration = uiexec.info.totalDuration + event.taskInfo.duration,
+        totalGCTime = uiexec.info.totalGCTime + gcTime,
+        totalInputBytes = uiexec.info.totalInputBytes + inputBytes,
+        totalShuffleRead = uiexec.info.totalShuffleRead + shuffleRead,
+        totalShuffleWrite = uiexec.info.totalShuffleWrite + shuffleWrite)
+      new ExecutorSummaryWrapper(updated)
+    }
   }
 
   override def onStageCompleted(event: SparkListenerStageCompleted): Unit = {
@@ -486,10 +534,24 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
   }
 
   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.
     updateExecutorSummary(event.blockManagerId.executorId) { exec =>
+      // Only create memory metrics if the event has the info; this avoid setting bogus values
+      // when replaying old application logs.
+      val memMetrics = event.maxOnHeapMem.map { _ =>
+        newMemoryMetrics(
+          exec.info.memoryMetrics,
+          totalOnHeapStorageMemory = event.maxOnHeapMem,
+          totalOffHeapStorageMemory = event.maxOffHeapMem)
+      }
+
       val updated = newExecutorSummary(
         exec.info,
-        maxMemory = event.maxMem)
+        hostPort = event.blockManagerId.hostPort,
+        isActive = true,
+        maxMemory = event.maxMem,
+        memoryMetrics = memMetrics)
       new ExecutorSummaryWrapper(updated)
     }
   }
@@ -566,6 +628,12 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
     // Function to apply a delta to a value, but ensure that it doesn't go negative.
     def newValue(old: Long, delta: Long): Long = math.max(0, old + delta)
 
+    // Function to calculate the remaining memory after applying a delta. Assumes that if
+    // one value is provided, all the values are.
+    def remainingMemory(max: Option[Long], oldv: Option[Long], newv: Option[Long]): Option[Long] = {
+      max.map { m => newValue(m, oldv.get - newv.get) }
+    }
+
     // If the storage level is NONE, then don't update the storage level of existing information.
     val updatedStorageLevel = if (storageLevel.useMemory || storageLevel.useDisk) {
       Some(storageLevel.description)
@@ -579,6 +647,7 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
       new ExecutorSummaryWrapper(newExecutorSummary(None, id = executorId)))
 
     var rddBlocksDelta = 0
+    val memMetrics = executorInfo.info.memoryMetrics
 
     // Update the block entry in the RDD info, keeping track of the deltas above so that we
     // can update the executor information too.
@@ -634,12 +703,31 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
       val newDistMem = newValue(oldDist.memoryUsed, event.blockUpdatedInfo.memSize * memoryMult)
       val newDistDisk = newValue(oldDist.diskUsed, event.blockUpdatedInfo.diskSize * diskMult)
       val newDists = if (newDistMem > 0 || newDistDisk > 0) {
+        val newOffHeap = if (storageLevel.useOffHeap) Some(newDistMem) else None
+        val newOnHeap = if (!storageLevel.useOffHeap) Some(newDistMem) else None
+        val remainingOffHeap = if (storageLevel.useOffHeap) {
+          remainingMemory(memMetrics.map(_.totalOffHeapStorageMemory), oldDist.offHeapMemoryUsed,
+            newOffHeap)
+        } else {
+          None
+        }
+        val remainingOnHeap = if (!storageLevel.useOffHeap) {
+          remainingMemory(memMetrics.map(_.totalOnHeapStorageMemory), oldDist.onHeapMemoryUsed,
+            newOnHeap)
+        } else {
+          None
+        }
+
         val newDist = newRDDDataDistribution(
           oldDist,
           memoryUsed = newDistMem,
           memoryRemaining = newValue(oldDist.memoryRemaining,
             event.blockUpdatedInfo.memSize * memoryMult * -1),
-          diskUsed = newDistDisk)
+          diskUsed = newDistDisk,
+          onHeapMemoryUsed = newOnHeap,
+          offHeapMemoryUsed = newOffHeap,
+          onHeapMemoryRemaining = remainingOnHeap,
+          offHeapMemoryRemaining = remainingOffHeap)
         Seq(newDist)
       } else {
         Nil
@@ -656,13 +744,30 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
     }
 
     // Update the ExecutorSummary for the block's manager.
+    val updatedMemMetrics = memMetrics.map { m =>
+      val onHeapUsed = if (!storageLevel.useOffHeap) {
+        newValue(m.usedOnHeapStorageMemory, event.blockUpdatedInfo.memSize * memoryMult)
+      } else {
+        m.usedOnHeapStorageMemory
+      }
+      val offHeapUsed = if (storageLevel.useOffHeap) {
+        newValue(m.usedOffHeapStorageMemory, event.blockUpdatedInfo.memSize * memoryMult)
+      } else {
+        m.usedOffHeapStorageMemory
+      }
+      newMemoryMetrics(m,
+        usedOnHeapStorageMemory = onHeapUsed,
+        usedOffHeapStorageMemory = offHeapUsed)
+    }
+
     val newExecSummary = newExecutorSummary(
       executorInfo.info,
       rddBlocks = newValue(executorInfo.info.rddBlocks, rddBlocksDelta).toInt,
       memoryUsed = newValue(executorInfo.info.memoryUsed,
         event.blockUpdatedInfo.memSize * memoryMult),
       diskUsed = newValue(executorInfo.info.diskUsed,
-        event.blockUpdatedInfo.diskSize * diskMult))
+        event.blockUpdatedInfo.diskSize * diskMult),
+      memoryMetrics = updatedMemMetrics)
     kvstore.write(new ExecutorSummaryWrapper(newExecSummary))
   }
 
@@ -819,6 +924,19 @@ private class AppStateListener(override protected val kvstore: KVStore) extends
       option(memoryMetrics, old.map(_.memoryMetrics)))
   }
 
+  private def newMemoryMetrics(
+      old: Option[v1.MemoryMetrics],
+      usedOnHeapStorageMemory: Option[Long] = None,
+      usedOffHeapStorageMemory: Option[Long] = None,
+      totalOnHeapStorageMemory: Option[Long] = None,
+      totalOffHeapStorageMemory: Option[Long] = None): v1.MemoryMetrics = {
+    new v1.MemoryMetrics(
+      value(usedOnHeapStorageMemory, old.map(_.usedOnHeapStorageMemory), 0L),
+      value(usedOffHeapStorageMemory, old.map(_.usedOffHeapStorageMemory), 0L),
+      value(totalOnHeapStorageMemory, old.map(_.totalOnHeapStorageMemory), 0L),
+      value(totalOffHeapStorageMemory, old.map(_.totalOffHeapStorageMemory), 0L))
+  }
+
   private def newJobData(
       old: Option[v1.JobData],
       jobId: Option[Int] = None,

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

@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.{JobExecutionStatus, SparkConf}
 import org.apache.spark.kvstore.KVStore
-import org.apache.spark.scheduler.SparkListenerBus
+import org.apache.spark.scheduler.{SparkListenerBus, SparkListenerEvent}
 import org.apache.spark.status.api.v1
 import org.apache.spark.util.{Distribution, Utils}
 
@@ -56,6 +56,15 @@ private[spark] class AppStateStore private (store: KVStore, tempStorePath: Optio
       .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] = {
     val it = store.view(classOf[StageDataWrapper]).asScala.map(_.info)
     if (!statuses.isEmpty) {
@@ -198,6 +207,10 @@ private[spark] class AppStateStore private (store: KVStore, tempStorePath: Optio
     store.read(classOf[RDDStorageInfoWrapper], rddId).info
   }
 
+  def executorEvents(): Seq[SparkListenerEvent] = {
+    store.view(classOf[ExecutorEventData]).asScala.map(_.event).toSeq
+  }
+
   def close(): Unit = {
     store.close()
     tempStorePath.foreach(Utils.deleteRecursively)

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/storeTypes.scala

@@ -22,6 +22,7 @@ import java.lang.{Integer => JInteger, Long => JLong}
 import com.fasterxml.jackson.annotation.JsonIgnore
 
 import org.apache.spark.kvstore.KVIndex
+import org.apache.spark.scheduler.SparkListenerEvent
 import org.apache.spark.status.api.v1._
 import org.apache.spark.status.KVUtils._
 
@@ -133,3 +134,11 @@ private[spark] class ExecutorStageSummaryWrapper(
   def stage: Array[Int] = Array(stageId, stageAttemptId)
 
 }
+
+/**
+ * Store raw executor events so that the executor timeline can be drawn. The event is wrapped
+ * in a container so that a monotonically increasing ID can be added to it.
+ */
+private[spark] class ExecutorEventData(
+  @KVIndexParam val id: Long,
+  val event: SparkListenerEvent)