[SPARK-24958][CORE] Add memory from procfs to executor metrics. #22612
…urning set of metrics.
- Loading branch information
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -45,7 +45,7 @@ private[spark] class ProcfsBasedSystems(val procfsDir: String = "/proc/") extend | |
| var pageSize = computePageSize() | ||
| var isAvailable: Boolean = isProcfsAvailable | ||
| private val pid = computePid() | ||
|
||
| private var ptree = mutable.Map[ Int, Set[Int]]() | ||
|
|
||
| var allMetrics: ProcfsBasedSystemsMetrics = ProcfsBasedSystemsMetrics(0, 0, 0, 0, 0, 0) | ||
|
|
||
|
|
@@ -84,7 +84,7 @@ private[spark] class ProcfsBasedSystems(val procfsDir: String = "/proc/") extend | |
| return pid; | ||
| } | ||
| catch { | ||
| case e: SparkException => logWarning("Exception when trying to compute process tree." + | ||
| " As a result reporting of ProcessTree metrics is stopped", e) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| isAvailable = false | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| return -1 | ||
|
|
@@ -95,15 +95,23 @@ private[spark] class ProcfsBasedSystems(val procfsDir: String = "/proc/") extend | |
| if (testing) { | ||
| return 0; | ||
| } | ||
| try { | ||
| val cmd = Array("getconf", "PAGESIZE") | ||
| val out2 = Utils.executeAndGetOutput(cmd) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| return Integer.parseInt(out2.split("\n")(0)) | ||
| } catch { | ||
| case e: Exception => logWarning("Exception when trying to compute pagesize, as a" + | ||
| " result reporting of ProcessTree metrics is stopped") | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| isAvailable = false | ||
| return 0 | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| } | ||
| } | ||
|
|
||
| private def computeProcessTree(): Unit = { | ||
| if (!isAvailable || testing) { | ||
| return | ||
| } | ||
| ptree = mutable.Map[ Int, Set[Int]]() | ||
| val queue = mutable.Queue.empty[Int] | ||
| queue += pid | ||
| while( !queue.isEmpty ) { | ||
|
|
@@ -121,34 +129,34 @@ private[spark] class ProcfsBasedSystems(val procfsDir: String = "/proc/") extend | |
|
|
||
| private def getChildPids(pid: Int): ArrayBuffer[Int] = { | ||
| try { | ||
| // val cmd = Array("pgrep", "-P", pid.toString) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| val builder = new ProcessBuilder("pgrep", "-P", pid.toString) | ||
| val process = builder.start() | ||
| // val output = new StringBuilder() | ||
| val threadName = "read stdout for " + "pgrep" | ||
| val childPidsInInt = mutable.ArrayBuffer.empty[Int] | ||
| def appendChildPid(s: String): Unit = { | ||
| if (s != "") { | ||
| logDebug("Found a child pid:" + s) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| childPidsInInt += Integer.parseInt(s) | ||
| } | ||
| } | ||
| val stdoutThread = Utils.processStreamByLine(threadName, | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| process.getInputStream, appendChildPid) | ||
| val exitCode = process.waitFor() | ||
| stdoutThread.join() | ||
| // pgrep will have exit code of 1 if there are more than one child process | ||
| // and it will have a exit code of 2 if there is no child process | ||
| if (exitCode != 0 && exitCode > 2) { | ||
| val cmd = builder.command().toArray.mkString(" ") | ||
| logWarning(s"Process $cmd" + | ||
| s" exited with code $exitCode, with stderr:" + s"${process.getErrorStream} ") | ||
|
||
| throw new SparkException(s"Process $cmd exited with code $exitCode") | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| } | ||
| childPidsInInt | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| } catch { | ||
| case e: Exception => logWarning("Exception when trying to compute process tree." + | ||
| " As a result reporting of ProcessTree metrics is stopped.", e) | ||
| isAvailable = false | ||
| return mutable.ArrayBuffer.empty[Int] | ||
| } | ||
|
|
@@ -173,54 +181,42 @@ private[spark] class ProcfsBasedSystems(val procfsDir: String = "/proc/") extend | |
| val vmem = procInfoSplit(22).toLong | ||
| val rssPages = procInfoSplit(23).toLong | ||
| if (procInfoSplit(1).toLowerCase(Locale.US).contains("java")) { | ||
|
||
| allMetrics = allMetrics.copy( | ||
| jvmVmemTotal = allMetrics.jvmVmemTotal + vmem, | ||
| jvmRSSTotal = allMetrics.jvmRSSTotal + (rssPages*pageSize) | ||
| ) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| } | ||
| else if (procInfoSplit(1).toLowerCase(Locale.US).contains("python")) { | ||
| allMetrics = allMetrics.copy( | ||
| pythonVmemTotal = allMetrics.pythonVmemTotal + vmem, | ||
| pythonRSSTotal = allMetrics.pythonRSSTotal + (rssPages*pageSize) | ||
| ) | ||
| } | ||
| else { | ||
| allMetrics = allMetrics.copy( | ||
| otherVmemTotal = allMetrics.otherVmemTotal + vmem, | ||
| otherRSSTotal = allMetrics.otherRSSTotal + (rssPages*pageSize) | ||
| ) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } catch { | ||
| case f: FileNotFoundException => logWarning("There was a problem with reading" + | ||
| " the stat file of the process. ", f) | ||
rezasafi marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| } | ||
| } | ||
|
|
||
| private[spark] def computeAllMetrics(): ProcfsBasedSystemsMetrics = { | ||
| if (!isAvailable) { | ||
| return ProcfsBasedSystemsMetrics(0, 0, 0, 0, 0, 0) | ||
| } | ||
| computeProcessTree | ||
| val pids = ptree.keySet | ||
| allMetrics = ProcfsBasedSystemsMetrics(0, 0, 0, 0, 0, 0) | ||
| for (p <- pids) { | ||
| computeProcessInfo(p) | ||
|
||
| } | ||
| return allMetrics | ||
| } | ||
| } | ||
Uh oh!
There was an error while loading. Please reload this page.