code review comments: hid array implementation of executor metrics, a…

…nd add ExecutorMetrics, with getMetricValue()

method for accessing executor metric values. Rename MetricGetter to ExecutorMetricType.

Should ExecutorMetricType be moved to executor package, or ExecutorMetrics be moved to metrics package?
Should Json (de)serialization functions be moved from api.scala to ExecutorMetrics?

core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala

@@ -22,6 +22,7 @@ import java.util.concurrent.{ScheduledFuture, TimeUnit}
 import scala.collection.mutable
 import scala.concurrent.Future
 
+import org.apache.spark.executor.ExecutorMetrics
 import org.apache.spark.internal.Logging
 import org.apache.spark.rpc.{RpcCallContext, RpcEnv, ThreadSafeRpcEndpoint}
 import org.apache.spark.scheduler._
@@ -38,7 +39,7 @@ private[spark] case class Heartbeat(
     executorId: String,
     accumUpdates: Array[(Long, Seq[AccumulatorV2[_, _]])], // taskId -> accumulator updates
     blockManagerId: BlockManagerId,
-    executorUpdates: Array[Long]) // executor level updates
+    executorUpdates: ExecutorMetrics) // executor level updates
 
 /**
  * An event that SparkContext uses to notify HeartbeatReceiver that SparkContext.taskScheduler is

core/src/main/scala/org/apache/spark/Heartbeater.scala

@@ -19,9 +19,10 @@ package org.apache.spark
 
 import java.util.concurrent.TimeUnit
 
+import org.apache.spark.executor.ExecutorMetrics
 import org.apache.spark.internal.Logging
 import org.apache.spark.memory.MemoryManager
-import org.apache.spark.metrics.MetricGetter
+import org.apache.spark.metrics.ExecutorMetricType
 import org.apache.spark.util.{ThreadUtils, Utils}
 
 /**
@@ -62,8 +63,9 @@ private[spark] class Heartbeater(
    * Get the current executor level metrics. These are returned as an array, with the index
    * determined by MetricGetter.values
    */
-  def getCurrentMetrics(): Array[Long] = {
-      MetricGetter.values.map(_.getMetricValue(memoryManager)).toArray
+  def getCurrentMetrics(): ExecutorMetrics = {
+    val metrics = ExecutorMetricType.values.map(_.getMetricValue(memoryManager)).toArray
+    new ExecutorMetrics(metrics)
   }
 }
 

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

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.executor
+
+import org.apache.spark.annotation.DeveloperApi
+import org.apache.spark.metrics.ExecutorMetricType
+
+/**
+ * :: DeveloperApi ::
+ * Metrics tracked for executors and the driver.
+ *
+ * Executor-level metrics are sent from each executor to the driver as part of the Heartbeat.
+ */
+@DeveloperApi
+class ExecutorMetrics private[spark] extends Serializable {
+
+  // Metrics are indexed by MetricGetter.values
+  private val metrics = new Array[Long](ExecutorMetricType.values.length)
+
+  // the first element is initialized to -1, indicating that the values for the array
+  // haven't been set yet.
+  metrics(0) = -1
+
+  /** Returns the value for the specified metricType. */
+  def getMetricValue(metricType: ExecutorMetricType): Long = {
+    metrics(ExecutorMetricType.metricIdxMap(metricType))
+  }
+
+  /** Returns true if the values for the metrics have been set, false otherwise. */
+  def isSet(): Boolean = metrics(0) > -1
+
+  private[spark] def this(metrics: Array[Long]) {
+    this()
+    Array.copy(metrics, 0, this.metrics, 0, Math.min(metrics.size, this.metrics.size))
+  }
+
+  /**
+   * Constructor: create the ExecutorMetrics with the values specified.
+   *
+   * @param executorMetrics map of executor metric name to value
+   */
+  private[spark] def this(executorMetrics: Map[String, Long]) {
+    this()
+    (0 until ExecutorMetricType.values.length).foreach { idx =>
+      metrics(idx) = executorMetrics.getOrElse(ExecutorMetricType.values(idx).name, 0L)
+    }
+  }
+
+  /**
+   * Compare the specified executor metrics values with the current executor metric values,
+   * and update the value for any metrics where the new value for the metric is larger.
+   *
+   * @param executorMetrics the executor metrics to compare
+   * @return if there is a new peak value for any metric
+   */
+  private[spark] def compareAndUpdatePeakValues(executorMetrics: ExecutorMetrics): Boolean = {
+    var updated: Boolean = false
+
+    (0 until ExecutorMetricType.values.length).foreach { idx =>
+       if ( executorMetrics.metrics(idx) > metrics(idx)) {
+        updated = true
+        metrics(idx) = executorMetrics.metrics(idx)
+      }
+    }
+    updated
+  }
+}

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

@@ -21,64 +21,70 @@ import javax.management.ObjectName
 
 import org.apache.spark.memory.MemoryManager
 
-private[spark] sealed trait MetricGetter {
-  def getMetricValue(memoryManager: MemoryManager): Long
-  val name = getClass().getName().stripSuffix("$").split("""\.""").last
+/**
+ * Executor metric types for executor-level metrics stored in ExecutorMetrics.
+ */
+sealed trait ExecutorMetricType {
+  private[spark] def getMetricValue(memoryManager: MemoryManager): Long
+  private[spark] val name = getClass().getName().stripSuffix("$").split("""\.""").last
 }
 
-private[spark] abstract class MemoryManagerMetricGetter(
-    f: MemoryManager => Long) extends MetricGetter {
-  override def getMetricValue(memoryManager: MemoryManager): Long = {
+private[spark] abstract class MemoryManagerExecutorMetricType(
+    f: MemoryManager => Long) extends ExecutorMetricType {
+  override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
     f(memoryManager)
   }
 }
 
-private[spark]abstract class MBeanMetricGetter(mBeanName: String)
-  extends MetricGetter {
-  val bean = ManagementFactory.newPlatformMXBeanProxy(ManagementFactory.getPlatformMBeanServer,
+private[spark]abstract class MBeanExecutorMetricType(mBeanName: String)
+  extends ExecutorMetricType {
+  private val bean = ManagementFactory.newPlatformMXBeanProxy(
+    ManagementFactory.getPlatformMBeanServer,
     new ObjectName(mBeanName).toString, classOf[BufferPoolMXBean])
 
-  override def getMetricValue(memoryManager: MemoryManager): Long = {
+  override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
     bean.getMemoryUsed
   }
 }
 
-private[spark] case object JVMHeapMemory extends MetricGetter {
-  override def getMetricValue(memoryManager: MemoryManager): Long = {
+case object JVMHeapMemory extends ExecutorMetricType {
+  override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
     ManagementFactory.getMemoryMXBean.getHeapMemoryUsage().getUsed()
   }
 }
 
-private[spark] case object JVMOffHeapMemory extends MetricGetter {
-  override def getMetricValue(memoryManager: MemoryManager): Long = {
+case object JVMOffHeapMemory extends ExecutorMetricType {
+  override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
     ManagementFactory.getMemoryMXBean.getNonHeapMemoryUsage().getUsed()
   }
 }
 
-private[spark] case object OnHeapExecutionMemory extends MemoryManagerMetricGetter(
+case object OnHeapExecutionMemory extends MemoryManagerExecutorMetricType(
   _.onHeapExecutionMemoryUsed)
 
-private[spark] case object OffHeapExecutionMemory extends MemoryManagerMetricGetter(
+case object OffHeapExecutionMemory extends MemoryManagerExecutorMetricType(
   _.offHeapExecutionMemoryUsed)
 
-private[spark] case object OnHeapStorageMemory extends MemoryManagerMetricGetter(
+case object OnHeapStorageMemory extends MemoryManagerExecutorMetricType(
   _.onHeapStorageMemoryUsed)
 
-private[spark] case object OffHeapStorageMemory extends MemoryManagerMetricGetter(
+case object OffHeapStorageMemory extends MemoryManagerExecutorMetricType(
   _.offHeapStorageMemoryUsed)
 
-private[spark] case object OnHeapUnifiedMemory extends MemoryManagerMetricGetter(
+case object OnHeapUnifiedMemory extends MemoryManagerExecutorMetricType(
   (m => m.onHeapExecutionMemoryUsed + m.onHeapStorageMemoryUsed))
 
-private[spark] case object OffHeapUnifiedMemory extends MemoryManagerMetricGetter(
+case object OffHeapUnifiedMemory extends MemoryManagerExecutorMetricType(
   (m => m.offHeapExecutionMemoryUsed + m.offHeapStorageMemoryUsed))
 
-private[spark] case object DirectPoolMemory extends MBeanMetricGetter(
+case object DirectPoolMemory extends MBeanExecutorMetricType(
   "java.nio:type=BufferPool,name=direct")
-private[spark] case object MappedPoolMemory extends MBeanMetricGetter(
+
+case object MappedPoolMemory extends MBeanExecutorMetricType(
   "java.nio:type=BufferPool,name=mapped")
 
-private[spark] object MetricGetter {
+private[spark] object ExecutorMetricType {
+  // List of all executor metric types
   val values = IndexedSeq(
     JVMHeapMemory,
     JVMOffHeapMemory,
@@ -92,5 +98,7 @@ private[spark] object MetricGetter {
     MappedPoolMemory
   )
 
-  val idxAndValues = values.zipWithIndex.map(_.swap)
+  // Map of executor metric type to its index in values.
+  val metricIdxMap =
+    Map[ExecutorMetricType, Int](ExecutorMetricType.values.zipWithIndex: _*)
 }

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

@@ -34,7 +34,7 @@ import org.apache.commons.lang3.SerializationUtils
 
 import org.apache.spark._
 import org.apache.spark.broadcast.Broadcast
-import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config
 import org.apache.spark.network.util.JavaUtils
@@ -248,7 +248,7 @@ class DAGScheduler(
       accumUpdates: Array[(Long, Int, Int, Seq[AccumulableInfo])],
       blockManagerId: BlockManagerId,
       // executor metrics indexed by MetricGetter.values
-      executorUpdates: Array[Long]): Boolean = {
+      executorUpdates: ExecutorMetrics): Boolean = {
     listenerBus.post(SparkListenerExecutorMetricsUpdate(execId, accumUpdates,
       Some(executorUpdates)))
     blockManagerMaster.driverEndpoint.askSync[Boolean](

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

@@ -23,7 +23,6 @@ import java.nio.charset.StandardCharsets
 import java.util.EnumSet
 import java.util.Locale
 
-import scala.collection.mutable
 import scala.collection.mutable.{ArrayBuffer, HashMap}
 
 import org.apache.hadoop.conf.Configuration
@@ -36,6 +35,7 @@ import org.json4s.jackson.JsonMethods._
 
 import org.apache.spark.{SPARK_VERSION, SparkConf}
 import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.executor.ExecutorMetrics
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
 import org.apache.spark.io.CompressionCodec
@@ -96,7 +96,7 @@ private[spark] class EventLoggingListener(
   private[scheduler] val logPath = getLogPath(logBaseDir, appId, appAttemptId, compressionCodecName)
 
   // map of (stageId, stageAttempt), to peak executor metrics for the stage
-  private val liveStageExecutorMetrics = HashMap[(Int, Int), HashMap[String, PeakExecutorMetrics]]()
+  private val liveStageExecutorMetrics = HashMap[(Int, Int), HashMap[String, ExecutorMetrics]]()
 
   /**
    * Creates the log file in the configured log directory.
@@ -165,7 +165,7 @@ private[spark] class EventLoggingListener(
     if (shouldLogStageExecutorMetrics) {
       // record the peak metrics for the new stage
       liveStageExecutorMetrics.put((event.stageInfo.stageId, event.stageInfo.attemptNumber()),
-        new HashMap[String, PeakExecutorMetrics]())
+        new HashMap[String, ExecutorMetrics]())
     }
   }
 
@@ -195,7 +195,7 @@ private[spark] class EventLoggingListener(
       executorOpt.foreach { execMap =>
         execMap.foreach { case (executorId, peakExecutorMetrics) =>
             logEvent(new SparkListenerStageExecutorMetrics(executorId, event.stageInfo.stageId,
-              event.stageInfo.attemptNumber(), peakExecutorMetrics.metrics))
+              event.stageInfo.attemptNumber(), peakExecutorMetrics))
         }
       }
     }
@@ -272,8 +272,8 @@ private[spark] class EventLoggingListener(
       event.executorUpdates.foreach { executorUpdates =>
         liveStageExecutorMetrics.values.foreach { peakExecutorMetrics =>
           val peakMetrics = peakExecutorMetrics.getOrElseUpdate(
-            event.execId, new PeakExecutorMetrics())
-          peakMetrics.compareAndUpdate(executorUpdates)
+            event.execId, new ExecutorMetrics())
+          peakMetrics.compareAndUpdatePeakValues(executorUpdates)
         }
       }
     }

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

@@ -26,7 +26,7 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo
 
 import org.apache.spark.{SparkConf, TaskEndReason}
 import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.scheduler.cluster.ExecutorInfo
 import org.apache.spark.storage.{BlockManagerId, BlockUpdatedInfo}
 import org.apache.spark.ui.SparkUI
@@ -166,7 +166,7 @@ case class SparkListenerBlockUpdated(blockUpdatedInfo: BlockUpdatedInfo) extends
 case class SparkListenerExecutorMetricsUpdate(
     execId: String,
     accumUpdates: Seq[(Long, Int, Int, Seq[AccumulableInfo])],
-    executorUpdates: Option[Array[Long]] = None)
+    executorUpdates: Option[ExecutorMetrics] = None)
   extends SparkListenerEvent
 
 /**
@@ -182,7 +182,7 @@ case class SparkListenerStageExecutorMetrics(
     execId: String,
     stageId: Int,
     stageAttemptId: Int,
-    executorMetrics: Array[Long])
+    executorMetrics: ExecutorMetrics)
   extends SparkListenerEvent
 
 @DeveloperApi

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.scheduler
 
+import org.apache.spark.executor.ExecutorMetrics
 import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
 import org.apache.spark.storage.BlockManagerId
 import org.apache.spark.util.AccumulatorV2
@@ -76,7 +77,7 @@ private[spark] trait TaskScheduler {
       execId: String,
       accumUpdates: Array[(Long, Seq[AccumulatorV2[_, _]])],
       blockManagerId: BlockManagerId,
-      executorUpdates: Array[Long]): Boolean
+      executorUpdates: ExecutorMetrics): Boolean
 
   /**
    * Get an application ID associated with the job.