[SPARK-20645][CORE] Port environment page to new UI backend.

This change modifies the status listener to collect the information
needed to render the envionment page, and populates that page and the
API with information collected by the listener.

Tested with existing and added unit tests.

Author: Marcelo Vanzin <[email protected]>

Closes apache#19677 from vanzin/SPARK-20645.

Author: Marcelo Vanzin

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

@@ -88,6 +88,24 @@ private[spark] class AppStatusListener(
     kvstore.write(new ApplicationInfoWrapper(appInfo))
   }
 
+  override def onEnvironmentUpdate(event: SparkListenerEnvironmentUpdate): Unit = {
+    val details = event.environmentDetails
+
+    val jvmInfo = Map(details("JVM Information"): _*)
+    val runtime = new v1.RuntimeInfo(
+      jvmInfo.get("Java Version").orNull,
+      jvmInfo.get("Java Home").orNull,
+      jvmInfo.get("Scala Version").orNull)
+
+    val envInfo = new v1.ApplicationEnvironmentInfo(
+      runtime,
+      details.getOrElse("Spark Properties", Nil),
+      details.getOrElse("System Properties", Nil),
+      details.getOrElse("Classpath Entries", Nil))
+
+    kvstore.write(new ApplicationEnvironmentInfoWrapper(envInfo))
+  }
+
   override def onApplicationEnd(event: SparkListenerApplicationEnd): Unit = {
     val old = appInfo.attempts.head
     val attempt = new v1.ApplicationAttemptInfo(

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

@@ -33,6 +33,15 @@ import org.apache.spark.util.kvstore.{InMemoryStore, KVStore}
  */
 private[spark] class AppStatusStore(store: KVStore) {
 
+  def applicationInfo(): v1.ApplicationInfo = {
+    store.view(classOf[ApplicationInfoWrapper]).max(1).iterator().next().info
+  }
+
+  def environmentInfo(): v1.ApplicationEnvironmentInfo = {
+    val klass = classOf[ApplicationEnvironmentInfoWrapper]
+    store.read(klass, klass.getName()).info
+  }
+
   def jobsList(statuses: JList[JobExecutionStatus]): Seq[v1.JobData] = {
     val it = store.view(classOf[JobDataWrapper]).asScala.map(_.info)
     if (!statuses.isEmpty()) {

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

@@ -26,20 +26,7 @@ private[v1] class ApplicationEnvironmentResource(ui: SparkUI) {
 
   @GET
   def getEnvironmentInfo(): ApplicationEnvironmentInfo = {
-    val listener = ui.environmentListener
-    listener.synchronized {
-      val jvmInfo = Map(listener.jvmInformation: _*)
-      val runtime = new RuntimeInfo(
-        jvmInfo("Java Version"),
-        jvmInfo("Java Home"),
-        jvmInfo("Scala Version"))
-
-      new ApplicationEnvironmentInfo(
-        runtime,
-        listener.sparkProperties,
-        listener.systemProperties,
-        listener.classpathEntries)
-    }
+    ui.store.environmentInfo()
   }
 
 }

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

@@ -34,6 +34,17 @@ private[spark] class ApplicationInfoWrapper(val info: ApplicationInfo) {
 
 }
 
+private[spark] class ApplicationEnvironmentInfoWrapper(val info: ApplicationEnvironmentInfo) {
+
+  /**
+   * There's always a single ApplicationEnvironmentInfo object per application, so this
+   * ID doesn't need to be dynamic. But the KVStore API requires an ID.
+   */
+  @JsonIgnore @KVIndex
+  def id: String = classOf[ApplicationEnvironmentInfoWrapper].getName()
+
+}
+
 private[spark] class ExecutorSummaryWrapper(val info: ExecutorSummary) {
 
   @JsonIgnore @KVIndex

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

@@ -25,11 +25,10 @@ import org.apache.spark.{SecurityManager, SparkConf, SparkContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler._
 import org.apache.spark.status.AppStatusStore
-import org.apache.spark.status.api.v1.{ApiRootResource, ApplicationAttemptInfo, ApplicationInfo,
-  UIRoot}
+import org.apache.spark.status.api.v1._
 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.env.EnvironmentTab
 import org.apache.spark.ui.exec.{ExecutorsListener, ExecutorsTab}
 import org.apache.spark.ui.jobs.{JobProgressListener, JobsTab, StagesTab}
 import org.apache.spark.ui.scope.RDDOperationGraphListener
@@ -44,7 +43,6 @@ private[spark] class SparkUI private (
     val sc: Option[SparkContext],
     val conf: SparkConf,
     securityManager: SecurityManager,
-    val environmentListener: EnvironmentListener,
     val storageStatusListener: StorageStatusListener,
     val executorsListener: ExecutorsListener,
     val jobProgressListener: JobProgressListener,
@@ -73,7 +71,7 @@ private[spark] class SparkUI private (
     val stagesTab = new StagesTab(this)
     attachTab(stagesTab)
     attachTab(new StorageTab(this))
-    attachTab(new EnvironmentTab(this))
+    attachTab(new EnvironmentTab(this, store))
     attachTab(new ExecutorsTab(this))
     attachHandler(createStaticHandler(SparkUI.STATIC_RESOURCE_DIR, "/static"))
     attachHandler(createRedirectHandler("/", "/jobs/", basePath = basePath))
@@ -88,9 +86,13 @@ private[spark] class SparkUI private (
   initialize()
 
   def getSparkUser: String = {
-    environmentListener.sparkUser
-      .orElse(environmentListener.systemProperties.toMap.get("user.name"))
-      .getOrElse("<unknown>")
+    try {
+      Option(store.applicationInfo().attempts.head.sparkUser)
+        .orElse(store.environmentInfo().systemProperties.toMap.get("user.name"))
+        .getOrElse("<unknown>")
+    } catch {
+      case _: NoSuchElementException => "<unknown>"
+    }
   }
 
   def getAppName: String = appName
@@ -143,6 +145,7 @@ private[spark] class SparkUI private (
   def setStreamingJobProgressListener(sparkListener: SparkListener): Unit = {
     streamingJobProgressListener = Option(sparkListener)
   }
+
 }
 
 private[spark] abstract class SparkUITab(parent: SparkUI, prefix: String)
@@ -184,21 +187,20 @@ private[spark] object SparkUI {
       addListenerFn(listener)
       listener
     }
-    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)
+    new SparkUI(store, sc, conf, securityManager, storageStatusListener, executorsListener,
+      jobProgressListener, storageListener, operationGraphListener, appName, basePath,
+      lastUpdateTime, startTime, appSparkVersion)
   }
 
 }

core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala

@@ -21,22 +21,31 @@ import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
 
-import org.apache.spark.ui.{UIUtils, WebUIPage}
+import org.apache.spark.SparkConf
+import org.apache.spark.status.AppStatusStore
+import org.apache.spark.ui._
 import org.apache.spark.util.Utils
 
-private[ui] class EnvironmentPage(parent: EnvironmentTab) extends WebUIPage("") {
-  private val listener = parent.listener
+private[ui] class EnvironmentPage(
+    parent: EnvironmentTab,
+    conf: SparkConf,
+    store: AppStatusStore) extends WebUIPage("") {
 
   def render(request: HttpServletRequest): Seq[Node] = {
+    val appEnv = store.environmentInfo()
+    val jvmInformation = Map(
+      "Java Version" -> appEnv.runtime.javaVersion,
+      "Java Home" -> appEnv.runtime.javaHome,
+      "Scala Version" -> appEnv.runtime.scalaVersion)
+
     val runtimeInformationTable = UIUtils.listingTable(
-      propertyHeader, jvmRow, listener.jvmInformation, fixedWidth = true)
+      propertyHeader, jvmRow, jvmInformation, fixedWidth = true)
     val sparkPropertiesTable = UIUtils.listingTable(propertyHeader, propertyRow,
-      Utils.redact(parent.conf, listener.sparkProperties), fixedWidth = true)
-
+      Utils.redact(conf, appEnv.sparkProperties.toSeq), fixedWidth = true)
     val systemPropertiesTable = UIUtils.listingTable(
-      propertyHeader, propertyRow, listener.systemProperties, fixedWidth = true)
+      propertyHeader, propertyRow, appEnv.systemProperties, fixedWidth = true)
     val classpathEntriesTable = UIUtils.listingTable(
-      classPathHeaders, classPathRow, listener.classpathEntries, fixedWidth = true)
+      classPathHeaders, classPathRow, appEnv.classpathEntries, fixedWidth = true)
     val content =
       <span>
         <h4>Runtime Information</h4> {runtimeInformationTable}
@@ -54,3 +63,9 @@ private[ui] class EnvironmentPage(parent: EnvironmentTab) extends WebUIPage("")
   private def propertyRow(kv: (String, String)) = <tr><td>{kv._1}</td><td>{kv._2}</td></tr>
   private def classPathRow(data: (String, String)) = <tr><td>{data._1}</td><td>{data._2}</td></tr>
 }
+
+private[ui] class EnvironmentTab(
+    parent: SparkUI,
+    store: AppStatusStore) extends SparkUITab(parent, "environment") {
+  attachPage(new EnvironmentPage(this, parent.conf, store))
+}