[SPARK-27489][WEBUI] UI updates to show executor resource information

### What changes were proposed in this pull request?
We are adding other resource type support to the executors and Spark. We should show the resource information for each executor on the UI Executors page.
This also adds a toggle button to show the resources column.  It is off by default.

![executorui1](https://user-images.githubusercontent.com/4563792/63891432-c815b580-c9aa-11e9-9f41-62975649efbc.png)

![Screenshot from 2019-08-28 14-56-26](https://user-images.githubusercontent.com/4563792/63891516-fd220800-c9aa-11e9-9fe4-89fcdca37306.png)

### Why are the changes needed?
to show user what resources the executors have. Like Gpus, fpgas, etc

### Does this PR introduce any user-facing change?
Yes introduces UI and rest api changes to show the resources

### How was this patch tested?
Unit tests and manual UI tests on yarn and standalone modes.

Closes #25613 from tgravescs/SPARK-27489-gpu-ui-latest.

Authored-by: Thomas Graves <tgraves@nvidia.com>
Signed-off-by: Gengliang Wang <gengliang.wang@databricks.com>

Author: tgravescs

core/src/main/resources/org/apache/spark/ui/static/executorspage-template.html

@@ -92,6 +92,7 @@ <h4 class="title-table">Executors</h4>
               Off Heap Storage Memory</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Disk Used">Disk Used</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Cores">Cores</span></th>
+          <th><span data-toggle="tooltip" data-placement="top" title="Resources">Resources</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Active Tasks">Active Tasks</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Failed Tasks">Failed Tasks</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Complete Tasks">Complete Tasks</span></th>

core/src/main/resources/org/apache/spark/ui/static/executorspage.js

@@ -39,6 +39,19 @@ function formatStatus(status, type, row) {
     return "Dead"
 }
 
+function formatResourceCells(resources) {
+    var result = ""
+    var count = 0
+    $.each(resources, function (name, resInfo) {
+        if (count > 0) {
+            result += ", "
+        }
+        result += name + ': [' + resInfo.addresses.join(", ") + ']'
+        count += 1
+    });
+    return result
+}
+
 jQuery.extend(jQuery.fn.dataTableExt.oSort, {
     "title-numeric-pre": function (a) {
         var x = a.match(/title="*(-?[0-9\.]+)/)[1];
@@ -106,7 +119,7 @@ function totalDurationColor(totalGCTime, totalDuration) {
 }
 
 var sumOptionalColumns = [3, 4];
-var execOptionalColumns = [5, 6];
+var execOptionalColumns = [5, 6, 9];
 var execDataTable;
 var sumDataTable;
 
@@ -401,6 +414,7 @@ $(document).ready(function () {
                         },
                         {data: 'diskUsed', render: formatBytes},
                         {data: 'totalCores'},
+                        {name: 'resourcesCol', data: 'resources', render: formatResourceCells, orderable: false},
                         {
                             data: 'activeTasks',
                             "fnCreatedCell": function (nTd, sData, oData, iRow, iCol) {
@@ -446,7 +460,8 @@ $(document).ready(function () {
                     "order": [[0, "asc"]],
                     "columnDefs": [
                         {"visible": false, "targets": 5},
-                        {"visible": false, "targets": 6}
+                        {"visible": false, "targets": 6},
+                        {"visible": false, "targets": 9}
                     ]
                 };
 
@@ -553,6 +568,7 @@ $(document).ready(function () {
                     "<div><input type='checkbox' class='toggle-vis' id='select-all-box'>Select All</div>" +
                     "<div id='on_heap_memory' class='on-heap-memory-checkbox-div'><input type='checkbox' class='toggle-vis' data-sum-col-idx='3' data-exec-col-idx='5'>On Heap Memory</div>" +
                     "<div id='off_heap_memory' class='off-heap-memory-checkbox-div'><input type='checkbox' class='toggle-vis' data-sum-col-idx='4' data-exec-col-idx='6'>Off Heap Memory</div>" +
+                    "<div id='extra_resources' class='resources-checkbox-div'><input type='checkbox' class='toggle-vis' data-sum-col-idx='' data-exec-col-idx='9'>Resources</div>" +
                     "</div>");
 
                 reselectCheckboxesBasedOnTaskTableState();
@@ -584,8 +600,10 @@ $(document).ready(function () {
                         var execCol = execDataTable.column(execColIdx);
                         execCol.visible(!execCol.visible());
                         var sumColIdx = thisBox.attr("data-sum-col-idx");
-                        var sumCol = sumDataTable.column(sumColIdx);
-                        sumCol.visible(!sumCol.visible());
+                        if (sumColIdx) {
+                            var sumCol = sumDataTable.column(sumColIdx);
+                            sumCol.visible(!sumCol.visible());
+                        }
                     }
                 });
 

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

@@ -72,7 +72,7 @@ private[spark] class HistoryAppStatusStore(
       source.totalGCTime, source.totalInputBytes, source.totalShuffleRead,
       source.totalShuffleWrite, source.isBlacklisted, source.maxMemory, source.addTime,
       source.removeTime, source.removeReason, newExecutorLogs, source.memoryMetrics,
-      source.blacklistedInStages, source.peakMemoryMetrics, source.attributes)
+      source.blacklistedInStages, source.peakMemoryMetrics, source.attributes, source.resources)
   }
 
 }

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

@@ -26,12 +26,15 @@ import org.apache.spark.rpc.RpcEndpointRef
 import org.apache.spark.util.Utils
 
 private[spark] case class WorkerResourceInfo(name: String, addresses: Seq[String])
-  extends ResourceAllocator(name, addresses) {
+  extends ResourceAllocator {
+
+  override protected def resourceName = this.name
+  override protected def resourceAddresses = this.addresses
 
   def acquire(amount: Int): ResourceInformation = {
     val allocated = availableAddrs.take(amount)
     acquire(allocated)
-    new ResourceInformation(name, allocated.toArray)
+    new ResourceInformation(resourceName, allocated.toArray)
   }
 }
 

core/src/main/scala/org/apache/spark/resource/ResourceAllocator.scala

@@ -23,18 +23,20 @@ import org.apache.spark.SparkException
 import org.apache.spark.util.collection.OpenHashMap
 
 /**
- * Class used to help executor/worker allocate resources
- * Please note that this class is intended to be used in a single thread.
- * @param name Resource name, e.g. gpu/fpga
- * @param addresses Resource addresses provided by the executor/worker
+ * Trait used to help executor/worker allocate resources.
+ * Please note that this is intended to be used in a single thread.
  */
-class ResourceAllocator(name: String, addresses: Seq[String]) extends Serializable {
+trait ResourceAllocator {
+
+  protected def resourceName: String
+  protected def resourceAddresses: Seq[String]
+
   /**
    * Map from an address to its availability, the value `true` means the address is available,
    * while value `false` means the address is assigned.
    * TODO Use [[OpenHashMap]] instead to gain better performance.
    */
-  private val addressAvailabilityMap = mutable.HashMap(addresses.map(_ -> true): _*)
+  private lazy val addressAvailabilityMap = mutable.HashMap(resourceAddresses.map(_ -> true): _*)
 
   /**
    * Sequence of currently available resource addresses.
@@ -59,15 +61,15 @@ class ResourceAllocator(name: String, addresses: Seq[String]) extends Serializab
   def acquire(addrs: Seq[String]): Unit = {
     addrs.foreach { address =>
       if (!addressAvailabilityMap.contains(address)) {
-        throw new SparkException(s"Try to acquire an address that doesn't exist. $name address " +
-          s"$address doesn't exist.")
+        throw new SparkException(s"Try to acquire an address that doesn't exist. $resourceName " +
+          s"address $address doesn't exist.")
       }
       val isAvailable = addressAvailabilityMap(address)
       if (isAvailable) {
         addressAvailabilityMap(address) = false
       } else {
-        throw new SparkException(s"Try to acquire an address that is not available. $name " +
-          s"address $address is not available.")
+        throw new SparkException("Try to acquire an address that is not available. " +
+          s"$resourceName address $address is not available.")
       }
     }
   }
@@ -80,14 +82,14 @@ class ResourceAllocator(name: String, addresses: Seq[String]) extends Serializab
   def release(addrs: Seq[String]): Unit = {
     addrs.foreach { address =>
       if (!addressAvailabilityMap.contains(address)) {
-        throw new SparkException(s"Try to release an address that doesn't exist. $name address " +
-          s"$address doesn't exist.")
+        throw new SparkException(s"Try to release an address that doesn't exist. $resourceName " +
+          s"address $address doesn't exist.")
       }
       val isAvailable = addressAvailabilityMap(address)
       if (!isAvailable) {
         addressAvailabilityMap(address) = true
       } else {
-        throw new SparkException(s"Try to release an address that is not assigned. $name " +
+        throw new SparkException(s"Try to release an address that is not assigned. $resourceName " +
           s"address $address is not assigned.")
       }
     }

core/src/main/scala/org/apache/spark/resource/ResourceInformation.scala

@@ -74,6 +74,16 @@ private[spark] object ResourceInformation {
           s"Here is a correct example: $exampleJson.", e)
     }
   }
+
+  def parseJson(json: JValue): ResourceInformation = {
+    implicit val formats = DefaultFormats
+    try {
+      json.extract[ResourceInformationJson].toResourceInformation
+    } catch {
+      case NonFatal(e) =>
+        throw new SparkException(s"Error parsing JSON into ResourceInformation:\n$json\n", e)
+    }
+  }
 }
 
 /** A case class to simplify JSON serialization of [[ResourceInformation]]. */

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.scheduler
 
-import org.apache.spark.resource.ResourceAllocator
+import org.apache.spark.resource.{ResourceAllocator, ResourceInformation}
 
 /**
  * Class to hold information about a type of Resource on an Executor. This information is managed
@@ -27,4 +27,8 @@ import org.apache.spark.resource.ResourceAllocator
  * @param addresses Resource addresses provided by the executor
  */
 private[spark] class ExecutorResourceInfo(name: String, addresses: Seq[String])
-  extends ResourceAllocator(name, addresses)
+  extends ResourceInformation(name, addresses.toArray) with ResourceAllocator {
+
+  override protected def resourceName = this.name
+  override protected def resourceAddresses = this.addresses
+}

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

@@ -31,12 +31,12 @@ import org.apache.spark.scheduler.ExecutorResourceInfo
  * @param resourcesInfo The information of the currently available resources on the executor
  */
 private[cluster] class ExecutorData(
-   val executorEndpoint: RpcEndpointRef,
-   val executorAddress: RpcAddress,
-   override val executorHost: String,
-   var freeCores: Int,
-   override val totalCores: Int,
-   override val logUrlMap: Map[String, String],
-   override val attributes: Map[String, String],
-   val resourcesInfo: Map[String, ExecutorResourceInfo]
-) extends ExecutorInfo(executorHost, totalCores, logUrlMap, attributes)
+    val executorEndpoint: RpcEndpointRef,
+    val executorAddress: RpcAddress,
+    override val executorHost: String,
+    var freeCores: Int,
+    override val totalCores: Int,
+    override val logUrlMap: Map[String, String],
+    override val attributes: Map[String, String],
+    override val resourcesInfo: Map[String, ExecutorResourceInfo]
+) extends ExecutorInfo(executorHost, totalCores, logUrlMap, attributes, resourcesInfo)

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

@@ -17,6 +17,7 @@
 package org.apache.spark.scheduler.cluster
 
 import org.apache.spark.annotation.DeveloperApi
+import org.apache.spark.resource.ResourceInformation
 
 /**
  * :: DeveloperApi ::
@@ -27,10 +28,19 @@ class ExecutorInfo(
    val executorHost: String,
    val totalCores: Int,
    val logUrlMap: Map[String, String],
-   val attributes: Map[String, String]) {
+   val attributes: Map[String, String],
+   val resourcesInfo: Map[String, ResourceInformation]) {
 
   def this(executorHost: String, totalCores: Int, logUrlMap: Map[String, String]) = {
-    this(executorHost, totalCores, logUrlMap, Map.empty)
+    this(executorHost, totalCores, logUrlMap, Map.empty, Map.empty)
+  }
+
+  def this(
+      executorHost: String,
+      totalCores: Int,
+      logUrlMap: Map[String, String],
+      attributes: Map[String, String]) = {
+    this(executorHost, totalCores, logUrlMap, attributes, Map.empty)
   }
 
   def canEqual(other: Any): Boolean = other.isInstanceOf[ExecutorInfo]
@@ -41,12 +51,13 @@ class ExecutorInfo(
         executorHost == that.executorHost &&
         totalCores == that.totalCores &&
         logUrlMap == that.logUrlMap &&
-        attributes == that.attributes
+        attributes == that.attributes &&
+        resourcesInfo == that.resourcesInfo
     case _ => false
   }
 
   override def hashCode(): Int = {
-    val state = Seq(executorHost, totalCores, logUrlMap, attributes)
+    val state = Seq(executorHost, totalCores, logUrlMap, attributes, resourcesInfo)
     state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
   }
 }

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

@@ -199,6 +199,7 @@ private[spark] class AppStatusListener(
     exec.totalCores = event.executorInfo.totalCores
     exec.maxTasks = event.executorInfo.totalCores / coresPerTask
     exec.executorLogs = event.executorInfo.logUrlMap
+    exec.resources = event.executorInfo.resourcesInfo
     exec.attributes = event.executorInfo.attributes
     liveUpdate(exec, System.nanoTime())
   }