case objects, mostly complete

core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.executor
 
 import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.scheduler.MemoryTypes
+import org.apache.spark.metrics.MetricGetter
 
 /**
  * :: DeveloperApi ::
@@ -32,5 +32,5 @@ import org.apache.spark.scheduler.MemoryTypes
  */
 @DeveloperApi
 class ExecutorMetrics private[spark] (val timestamp: Long) extends Serializable {
-  val metrics = new Array[Long](MemoryTypes.values().length)
+  val metrics = new Array[Long](MetricGetter.values.length)
 }

core/src/main/scala/org/apache/spark/metrics/MetricGetter.scala

@@ -23,6 +23,7 @@ import org.apache.spark.memory.MemoryManager
 
 sealed trait MetricGetter {
   def getMetricValue(memoryManager: MemoryManager): Long
+  val name = getClass().getName().stripSuffix("$")
 }
 
 abstract class MemoryManagerMetricGetter(f: MemoryManager => Long) extends MetricGetter {
@@ -32,9 +33,8 @@ abstract class MemoryManagerMetricGetter(f: MemoryManager => Long) extends Metri
 }
 
 abstract class MBeanMetricGetter(mBeanName: String) extends MetricGetter {
-  private val name = new ObjectName(mBeanName)
   val bean = ManagementFactory.newPlatformMXBeanProxy(ManagementFactory.getPlatformMBeanServer,
-    name.toString, classOf[BufferPoolMXBean])
+    new ObjectName(mBeanName).toString, classOf[BufferPoolMXBean])
 
   override def getMetricValue(memoryManager: MemoryManager): Long = {
     bean.getMemoryUsed

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

@@ -18,14 +18,15 @@
 package org.apache.spark.scheduler
 
 import org.apache.spark.executor.ExecutorMetrics
+import org.apache.spark.metrics.MetricGetter
 import org.apache.spark.status.api.v1.PeakMemoryMetrics
 
 /**
  * Records the peak values for executor level metrics. If jvmUsedHeapMemory is -1, then no
  * values have been recorded yet.
  */
 private[spark] class PeakExecutorMetrics {
-  val metrics = new Array[Long](MemoryTypes.values().length)
+  val metrics = new Array[Long](MetricGetter.values.length)
   metrics(0) = -1
 
   /**
@@ -38,7 +39,7 @@ private[spark] class PeakExecutorMetrics {
   def compareAndUpdate(executorMetrics: ExecutorMetrics): Boolean = {
     var updated: Boolean = false
 
-    (0 until MemoryTypes.values().length).foreach { metricIdx =>
+    (0 until MetricGetter.values.length).foreach { metricIdx =>
       val newVal = executorMetrics.metrics(metricIdx)
       if ( newVal > metrics(metricIdx)) {
         updated = true

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

@@ -25,7 +25,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 
 import org.apache.spark.JobExecutionStatus
-import org.apache.spark.scheduler.MemoryTypes
+import org.apache.spark.metrics.MetricGetter
 
 case class ApplicationInfo private[spark](
     id: String,
@@ -110,7 +110,7 @@ class MemoryMetrics private[spark](
 
 class PeakMemoryMetrics private[spark]() {
   // TODO special json-ification
-  val metrics = new Array[Long](MemoryTypes.values().length)
+  val metrics = new Array[Long](MetricGetter.values.length)
 }
 
 class JobData private[spark](

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -31,6 +31,7 @@ import org.json4s.jackson.JsonMethods._
 
 import org.apache.spark._
 import org.apache.spark.executor._
+import org.apache.spark.metrics.MetricGetter
 import org.apache.spark.rdd.RDDOperationScope
 import org.apache.spark.scheduler._
 import org.apache.spark.scheduler.cluster.ExecutorInfo
@@ -389,17 +390,12 @@ private[spark] object JsonProtocol {
    * @return the JSON representation
    */
   def executorMetricsToJson(executorMetrics: ExecutorMetrics): JValue = {
-    ("Timestamp" -> executorMetrics.timestamp) ~
-    ("JVM Used Heap Memory" -> executorMetrics.jvmUsedHeapMemory) ~
-    ("JVM Used Nonheap Memory" -> executorMetrics.jvmUsedNonHeapMemory) ~
-    ("Onheap Execution Memory" -> executorMetrics.onHeapExecutionMemory) ~
-    ("Offheap Execution Memory" -> executorMetrics.offHeapExecutionMemory) ~
-    ("Onheap Storage Memory" -> executorMetrics.onHeapStorageMemory) ~
-    ("Offheap Storage Memory" -> executorMetrics.offHeapStorageMemory) ~
-    ("Onheap Unified Memory" -> executorMetrics.onHeapUnifiedMemory) ~
-    ("Offheap Unified Memory" -> executorMetrics.offHeapUnifiedMemory) ~
-    ("Direct Memory" -> executorMetrics.directMemory) ~
-    ("Mapped Memory" -> executorMetrics.mappedMemory)
+    val metrics = MetricGetter.idxAndValues.map { case (idx, metric) =>
+      JField(metric.name, executorMetrics.metrics(idx))
+    }
+    JObject(
+      (Seq(JField("Timestamp", executorMetrics.timestamp)) ++ metrics): _*
+    )
   }
 
   def taskEndReasonToJson(taskEndReason: TaskEndReason): JValue = {
@@ -616,20 +612,12 @@ private[spark] object JsonProtocol {
    */
   def executorMetricsFromJson(json: JValue): ExecutorMetrics = {
     val timeStamp = (json \ "Timestamp").extract[Long]
-    val jvmUsedHeapMemory = (json \ "JVM Used Heap Memory").extract[Long]
-    val jvmUsedNonHeapMemory = (json \ "JVM Used Nonheap Memory").extract[Long]
-    val onHeapExecutionMemory = (json \ "Onheap Execution Memory").extract[Long]
-    val offHeapExecutionMemory = (json \ "Offheap Execution Memory").extract[Long]
-    val onHeapStorageMemory = (json \ "Onheap Storage Memory").extract[Long]
-    val offHeapStorageMemory = (json \ "Offheap Storage Memory").extract[Long]
-    val onHeapUnifiedMemory = (json \ "Onheap Unified Memory").extract[Long]
-    val offHeapUnifiedMemory = (json \ "Offheap Unified Memory").extract[Long]
-    val directMemory = (json \ "Direct Memory").extract[Long]
-    val mappedMemory = (json \ "Mapped Memory").extract[Long]
-    new ExecutorMetrics(timeStamp, jvmUsedHeapMemory, jvmUsedNonHeapMemory,
-      onHeapExecutionMemory, offHeapExecutionMemory, onHeapStorageMemory,
-      offHeapStorageMemory, onHeapUnifiedMemory, offHeapUnifiedMemory, directMemory,
-      mappedMemory)
+    val metrics = new ExecutorMetrics(timeStamp)
+    MetricGetter.idxAndValues.foreach { case (idx, metric) =>
+      val metricValue = (json \ metric.name).extract[Long]
+      metrics.metrics(idx) = metricValue
+    }
+    metrics
   }
 
   def taskEndFromJson(json: JValue): SparkListenerTaskEnd = {

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

@@ -34,7 +34,7 @@ import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.internal.Logging
 import org.apache.spark.io._
-import org.apache.spark.metrics.MetricsSystem
+import org.apache.spark.metrics.{MetricGetter, MetricsSystem}
 import org.apache.spark.scheduler.cluster.ExecutorInfo
 import org.apache.spark.util.{JsonProtocol, Utils}
 
@@ -412,16 +412,9 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
     (executorMetrics1, executorMetrics2) match {
       case (Some(e1), Some(e2)) =>
         assert(e1.timestamp === e2.timestamp)
-        assert(e1.jvmUsedHeapMemory === e2.jvmUsedHeapMemory)
-        assert(e1.jvmUsedNonHeapMemory === e2.jvmUsedNonHeapMemory)
-        assert(e1.onHeapExecutionMemory === e2.onHeapExecutionMemory)
-        assert(e1.offHeapExecutionMemory === e2.offHeapExecutionMemory)
-        assert(e1.onHeapStorageMemory === e2.onHeapStorageMemory)
-        assert(e1.offHeapStorageMemory === e2.offHeapStorageMemory)
-        assert(e1.onHeapUnifiedMemory === e2.onHeapUnifiedMemory)
-        assert(e1.offHeapUnifiedMemory === e2.offHeapUnifiedMemory)
-        assert(e1.directMemory === e2.directMemory)
-        assert(e1.mappedMemory === e2.mappedMemory)
+        (0 until MetricGetter.values.length).foreach { idx =>
+          assert(e1.metrics(idx) === e2.metrics(idx))
+        }
       case (None, None) =>
       case _ =>
         assert(false)

core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala

@@ -30,6 +30,7 @@ import org.scalatest.exceptions.TestFailedException
 
 import org.apache.spark._
 import org.apache.spark.executor._
+import org.apache.spark.metrics.MetricGetter
 import org.apache.spark.rdd.RDDOperationScope
 import org.apache.spark.scheduler._
 import org.apache.spark.scheduler.cluster.ExecutorInfo
@@ -686,12 +687,9 @@ private[spark] object JsonProtocolSuite extends Assertions {
     (metrics1, metrics2) match {
       case (Some(m1), Some(m2)) =>
         assert(m1.timestamp === m2.timestamp)
-        assert(m1.jvmUsedHeapMemory === m2.jvmUsedHeapMemory)
-        assert(m1.jvmUsedNonHeapMemory === m2.jvmUsedNonHeapMemory)
-        assert(m1.onHeapExecutionMemory === m2.onHeapExecutionMemory)
-        assert(m1.offHeapExecutionMemory === m2.offHeapExecutionMemory)
-        assert(m1.onHeapStorageMemory === m2.onHeapStorageMemory)
-        assert(m1.offHeapStorageMemory === m2.offHeapStorageMemory)
+        (0 until MetricGetter.values.length).foreach { idx =>
+          assert(m1.metrics(idx) === m2.metrics(idx))
+        }
       case (None, None) =>
       case _ =>
         assert(false)