[SPARK-24958] Add executors' process tree total memory information to heartbeat signals.

Spark executors' process tree total memory information can be really useful. Currently such information are not available. The goal of this PR is to compute such information for each executor, add these information to the heartbeat signals, and compute the peaks at the driver.

This PR is tested by running the current unit tests and also some added ones. I have also tested this on our internal cluster and have verified the results.

Author: Reza Safi

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

@@ -25,28 +25,38 @@ import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.Queue
 
-import org.apache.spark.internal.Logging
+import org.apache.spark.SparkEnv
+import org.apache.spark.internal.{config, Logging}
 
+private[spark] case class ProcfsBasedSystemsMetrics(jvmVmemTotal: Long,
+                                             jvmRSSTotal: Long,
+                                             pythonVmemTotal: Long,
+                                             pythonRSSTotal: Long,
+                                             otherVmemTotal: Long,
+                                             otherRSSTotal: Long)
 
 // Some of the ideas here are taken from the ProcfsBasedProcessTree class in hadoop
 // project.
-class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
-  val procfsDir = "/proc/"
+private[spark] class ProcfsBasedSystems extends Logging {
+  var procfsDir = "/proc/"
+  val procfsStatFile = "stat"
+  var pageSize = 0
   var isAvailable: Boolean = isItProcfsBased
-  val pid: Int = computePid()
-  val ptree: scala.collection.mutable.Map[ Int, Set[Int]] =
+  private val pid: Int = computePid()
+  private val ptree: scala.collection.mutable.Map[ Int, Set[Int]] =
     scala.collection.mutable.Map[ Int, Set[Int]]()
-  val PROCFS_STAT_FILE = "stat"
-  var latestJVMVmemTotal: Long = 0
-  var latestJVMRSSTotal: Long = 0
-  var latestPythonVmemTotal: Long = 0
-  var latestPythonRSSTotal: Long = 0
-  var latestOtherVmemTotal: Long = 0
-  var latestOtherRSSTotal: Long = 0
 
-  createProcessTree
+  var allMetrics: ProcfsBasedSystemsMetrics = ProcfsBasedSystemsMetrics(0, 0, 0, 0, 0, 0)
+  private var latestJVMVmemTotal: Long = 0
+  private var latestJVMRSSTotal: Long = 0
+  private var latestPythonVmemTotal: Long = 0
+  private var latestPythonRSSTotal: Long = 0
+  private var latestOtherVmemTotal: Long = 0
+  private var latestOtherRSSTotal: Long = 0
 
-  def isItProcfsBased: Boolean = {
+  computeProcessTree()
+
+  private def isItProcfsBased: Boolean = {
     val testing = sys.env.contains("SPARK_TESTING") || sys.props.contains("spark.testing")
     if (testing) {
       return true
@@ -59,14 +69,14 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
     catch {
       case f: FileNotFoundException => return false
     }
-
-    val shouldLogStageExecutorProcessTreeMetrics = org.apache.spark.SparkEnv.get.conf.
-      getBoolean("spark.eventLog.logStageExecutorProcessTreeMetrics.enabled", true)
-    true && shouldLogStageExecutorProcessTreeMetrics
+    val shouldLogStageExecutorMetrics =
+      SparkEnv.get.conf.get(config.EVENT_LOG_STAGE_EXECUTOR_METRICS)
+    val shouldLogStageExecutorProcessTreeMetrics =
+      SparkEnv.get.conf.get(config.EVENT_LOG_PROCESS_TREE_METRICS)
+    shouldLogStageExecutorProcessTreeMetrics && shouldLogStageExecutorMetrics
   }
 
-
-  def computePid(): Int = {
+  private def computePid(): Int = {
     if (!isAvailable) {
       return -1;
     }
@@ -75,7 +85,7 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
       // https://docs.oracle.com/javase/9/docs/api/java/lang/ProcessHandle.html
       val cmd = Array("bash", "-c", "echo $PPID")
       val length = 10
-      var out: Array[Byte] = Array.fill[Byte](length)(0)
+      val out: Array[Byte] = Array.fill[Byte](length)(0)
       Runtime.getRuntime.exec(cmd).getInputStream.read(out)
       val pid = Integer.parseInt(new String(out, "UTF-8").trim)
       return pid;
@@ -92,11 +102,18 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
     }
   }
 
+  private def computePageSize(): Unit = {
+    val cmd = Array("getconf", "PAGESIZE")
+    val out: Array[Byte] = Array.fill[Byte](10)(0)
+    Runtime.getRuntime.exec(cmd).getInputStream.read(out)
+    pageSize = Integer.parseInt(new String(out, "UTF-8").trim)
+  }
 
-  def createProcessTree(): Unit = {
+  private def computeProcessTree(): Unit = {
     if (!isAvailable) {
       return
     }
+    computePageSize
     val queue: Queue[Int] = new Queue[Int]()
     queue += pid
     while( !queue.isEmpty ) {
@@ -112,35 +129,37 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
     }
   }
 
-
-  def updateProcessTree(): Unit = {
-    if (!isAvailable) {
-      return
-    }
-    val queue: Queue[Int] = new Queue[Int]()
-    queue += pid
-    while( !queue.isEmpty ) {
-      val p = queue.dequeue()
-      val c = getChildPIds(p)
-      if(!c.isEmpty) {
-        queue ++= c
-        val preChildren = ptree.get(p)
-        preChildren match {
-          case Some(children) => if (!c.toSet.equals(children)) {
-            val diff: Set[Int] = children -- c.toSet
-            ptree.update(p, c.toSet )
-            diff.foreach(ptree.remove(_))
-          }
-          case None => ptree.update(p, c.toSet )
-        }
+  private def getChildPIds(pid: Int): ArrayBuffer[Int] = {
+    try {
+      val cmd = Array("pgrep", "-P", pid.toString)
+      val input = Runtime.getRuntime.exec(cmd).getInputStream
+      val childPidsInByte: mutable.ArrayBuffer[Byte] = new mutable.ArrayBuffer()
+      var d = input.read()
+      while (d != -1) {
+        childPidsInByte.append(d.asInstanceOf[Byte])
+        d = input.read()
       }
-      else {
-        ptree.update(p, Set[Int]())
+      input.close()
+      val childPids = new String(childPidsInByte.toArray, "UTF-8").split("\n")
+      val childPidsInInt: ArrayBuffer[Int] = new ArrayBuffer[Int]()
+      for (p <- childPids) {
+        if (p != "") {
+          childPidsInInt += Integer.parseInt(p)
+        }
       }
+      childPidsInInt
+    } catch {
+      case e: IOException => logDebug("IO Exception when trying to compute process tree." +
+        " As a result reporting of ProcessTree metrics is stopped")
+        isAvailable = false
+        return new mutable.ArrayBuffer()
+      case _ => logDebug("Some exception occurred when trying to compute process tree." +
+        " As a result reporting of ProcessTree metrics is stopped")
+        isAvailable = false
+        return new mutable.ArrayBuffer()
     }
   }
 
-
   /**
    * Hadoop ProcfsBasedProcessTree class used regex and pattern matching to retrive the memory
    * info. I tried that but found it not correct during tests, so I used normal string analysis
@@ -152,7 +171,7 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
       val pidDir: File = new File(procfsDir, pid.toString)
       val fReader = new InputStreamReader(
         new FileInputStream(
-          new File(pidDir, PROCFS_STAT_FILE)), Charset.forName("UTF-8"))
+          new File(pidDir, procfsStatFile)), Charset.forName("UTF-8"))
       val in: BufferedReader = new BufferedReader(fReader)
       val procInfo = in.readLine
       in.close
@@ -172,16 +191,19 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
         latestOtherRSSTotal += procInfoSplit(23).toLong }
       }
     } catch {
-      case f: FileNotFoundException => return null
+      case f: FileNotFoundException =>
     }
   }
 
+  def updateAllMetrics(): Unit = {
+    allMetrics = computeAllMetrics
+  }
 
-  def getOtherRSSInfo(): Long = {
+  private def computeAllMetrics(): ProcfsBasedSystemsMetrics = {
     if (!isAvailable) {
-      return -1
+      return ProcfsBasedSystemsMetrics(-1, -1, -1, -1, -1, -1)
     }
-    updateProcessTree
+    computeProcessTree
     val pids = ptree.keySet
     latestJVMRSSTotal = 0
     latestJVMVmemTotal = 0
@@ -190,82 +212,57 @@ class ProcfsBasedSystems  extends ProcessTreeMetrics with Logging {
     latestOtherRSSTotal = 0
     latestOtherVmemTotal = 0
     for (p <- pids) {
-       getProcessInfo(p)
+      getProcessInfo(p)
     }
-    latestOtherRSSTotal
+    ProcfsBasedSystemsMetrics(
+      getJVMVirtualMemInfo,
+      getJVMRSSInfo,
+      getPythonVirtualMemInfo,
+      getPythonRSSInfo,
+      getOtherVirtualMemInfo,
+      getOtherRSSInfo)
+
   }
 
+  def getOtherRSSInfo(): Long = {
+    if (!isAvailable) {
+      return -1
+    }
+    latestOtherRSSTotal*pageSize
+  }
 
   def getOtherVirtualMemInfo(): Long = {
     if (!isAvailable) {
       return -1
     }
-    // We won't call updateProcessTree and also compute total virtual memory here
-    // since we already did all of this when we computed RSS info
     latestOtherVmemTotal
   }
 
-
   def getJVMRSSInfo(): Long = {
     if (!isAvailable) {
       return -1
     }
-    latestJVMRSSTotal
+    latestJVMRSSTotal*pageSize
   }
 
-
   def getJVMVirtualMemInfo(): Long = {
     if (!isAvailable) {
       return -1
     }
     latestJVMVmemTotal
   }
 
-
   def getPythonRSSInfo(): Long = {
     if (!isAvailable) {
       return -1
     }
-    latestPythonRSSTotal
+    latestPythonRSSTotal*pageSize
   }
 
-
   def getPythonVirtualMemInfo(): Long = {
     if (!isAvailable) {
       return -1
     }
     latestPythonVmemTotal
   }
-
-
-  def getChildPIds(pid: Int): ArrayBuffer[Int] = {
-    try {
-      val cmd = Array("pgrep", "-P", pid.toString)
-      val input = Runtime.getRuntime.exec(cmd).getInputStream
-      val childPidsInByte: mutable.ArrayBuffer[Byte] = new mutable.ArrayBuffer()
-      var d = input.read()
-      while (d != -1) {
-        childPidsInByte.append(d.asInstanceOf[Byte])
-        d = input.read()
-      }
-      input.close()
-      val childPids = new String(childPidsInByte.toArray, "UTF-8").split("\n")
-      val childPidsInInt: ArrayBuffer[Int] = new ArrayBuffer[Int]()
-      for (p <- childPids) {
-        if (p != "") {
-          childPidsInInt += Integer.parseInt(p)
-        }
-      }
-      childPidsInInt
-    } catch {
-    case e: IOException => logDebug("IO Exception when trying to compute process tree." +
-      " As a result reporting of ProcessTree metrics is stopped")
-      isAvailable = false
-      return new mutable.ArrayBuffer()
-    case _ => logDebug("Some exception occurred when trying to compute process tree. As a result" +
-      "  reporting of ProcessTree metrics is stopped")
-      isAvailable = false
-      return new mutable.ArrayBuffer()
-    }
-  }
 }

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -74,6 +74,11 @@ package object config {
       .booleanConf
       .createWithDefault(false)
 
+  private[spark] val EVENT_LOG_PROCESS_TREE_METRICS =
+    ConfigBuilder("spark.eventLog.logStageExecutorProcessTreeMetrics.enabled")
+      .booleanConf
+      .createWithDefault(false)
+
   private[spark] val EVENT_LOG_OVERWRITE =
     ConfigBuilder("spark.eventLog.overwrite").booleanConf.createWithDefault(false)
 

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

@@ -19,7 +19,7 @@ package org.apache.spark.metrics
 import java.lang.management.{BufferPoolMXBean, ManagementFactory}
 import javax.management.ObjectName
 
-import org.apache.spark.executor.{ProcessTreeMetrics, ProcfsBasedSystems}
+import org.apache.spark.executor.ProcfsBasedSystems
 import org.apache.spark.memory.MemoryManager
 
 /**
@@ -62,37 +62,38 @@ case object JVMOffHeapMemory extends ExecutorMetricType {
 
 case object ProcessTreeJVMRSSMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getJVMRSSInfo()
+    ExecutorMetricType.pTreeInfo.updateAllMetrics()
+    ExecutorMetricType.pTreeInfo.allMetrics.jvmRSSTotal
   }
 }
 
 case object ProcessTreeJVMVMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getJVMVirtualMemInfo()
+    ExecutorMetricType.pTreeInfo.allMetrics.jvmVmemTotal
   }
 }
 
 case object ProcessTreePythonRSSMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getPythonRSSInfo()
+    ExecutorMetricType.pTreeInfo.allMetrics.pythonRSSTotal
   }
 }
 
 case object ProcessTreePythonVMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getPythonVirtualMemInfo()
+    ExecutorMetricType.pTreeInfo.allMetrics.pythonVmemTotal
   }
 }
 
 case object ProcessTreeOtherRSSMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getOtherRSSInfo()
+    ExecutorMetricType.pTreeInfo.allMetrics.otherRSSTotal
   }
 }
 
 case object ProcessTreeOtherVMemory extends ExecutorMetricType {
   override private[spark] def getMetricValue(memoryManager: MemoryManager): Long = {
-    ExecutorMetricType.pTreeInfo.getOtherVirtualMemInfo()
+    ExecutorMetricType.pTreeInfo.allMetrics.otherVmemTotal
   }
 }
 
@@ -121,7 +122,7 @@ case object MappedPoolMemory extends MBeanExecutorMetricType(
   "java.nio:type=BufferPool,name=mapped")
 
 private[spark] object ExecutorMetricType {
-  final val pTreeInfo: ProcessTreeMetrics = new ProcfsBasedSystems
+  final val pTreeInfo = new ProcfsBasedSystems
 
   // List of all executor metric types
   val values = IndexedSeq(

core/src/test/resources/HistoryServerExpectations/application_list_json_expectation.json

@@ -1,4 +1,19 @@
 [ {
+  "id" : "application_1538416563558_0014",
+  "name" : "PythonBisectingKMeansExample",
+  "attempts" : [ {
+    "startTime" : "2018-10-02T00:42:39.580GMT",
+    "endTime" : "2018-10-02T00:44:02.338GMT",
+    "lastUpdated" : "",
+    "duration" : 82758,
+    "sparkUser" : "root",
+    "completed" : true,
+    "appSparkVersion" : "2.5.0-SNAPSHOT",
+    "lastUpdatedEpoch" : 0,
+    "startTimeEpoch" : 1538440959580,
+    "endTimeEpoch" : 1538441042338
+  } ]
+}, {
   "id" : "application_1506645932520_24630151",
   "name" : "Spark shell",
   "attempts" : [ {