+ * @param procFileSource + * a BufferedSource instance, with access to the desired proc file
+ * @param originalMetricName + * the original metric name by which it is found in the proc file
+ * @return + * the metric value as Long + * @throws NoSuchElementException + * if a metric is not found with the provided original name + */ + override def getMetricValue( + procFileSource: BufferedSource, + originalMetricName: String + ): Double = { + val procFileData = (procFileCPUColumns + .zip( + procFileSource.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong) + ) + .toMap) + val metricValue = + procFileData(originalMetricName) / procFileData.foldLeft(0.0)(_ + _._2) + procFileSource.close() + metricValue + } + + /** Access and buffers a proc file located at the `procFilePath`.
+ * @return + * a `BufferedSource` instance of the file read + */ + override def getProcFileSource(): scala.io.BufferedSource = { + sourceMethod(procFilePath) + } +} From 155c00faaccb7dd0ba92c855ac1654aa7258651a Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 17 Apr 2024 20:51:30 -0300 Subject: [PATCH 02/12] chore: Added StatMetricCollector tests --- src/test/scala/StatMetricCollectorTest.scala | 117 +++++++++++++++++++ src/test/scala/proc/stat | 15 +++ 2 files changed, 132 insertions(+) create mode 100644 src/test/scala/StatMetricCollectorTest.scala create mode 100644 src/test/scala/proc/stat diff --git a/src/test/scala/StatMetricCollectorTest.scala b/src/test/scala/StatMetricCollectorTest.scala new file mode 100644 index 0000000..307d39c --- /dev/null +++ b/src/test/scala/StatMetricCollectorTest.scala @@ -0,0 +1,117 @@ +import io.github.dutrevis.StatMetricCollector +import org.scalatest.funsuite.AnyFunSuite +import org.scalamock.scalatest.MockFactory +import org.mockito.{MockitoSugar, ArgumentMatchersSugar} +import org.mockito.integrations.scalatest.ResetMocksAfterEachTest +import org.mockito.captor.ArgCaptor +import scala.io.{Source, BufferedSource} + +import java.io.ByteArrayInputStream +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Paths} + +class StatMetricCollectorTest + extends AnyFunSuite + with MockitoSugar + with ArgumentMatchersSugar + with ResetMocksAfterEachTest { + + // Arrange common mocks + val procFileSourceMock = mock[BufferedSource] + + // Arrange common values + val procFileDataTest = Map[String, Long]( + "cpu_user" -> 79242, + "cpu_nice" -> 0, + "cpu_system" -> 74306, + "cpu_idle" -> 842486413, + "cpu_iowait" -> 756859, + "cpu_irq" -> 6140, + "cpu_softirq" -> 67701 + ) + + val procFileContent: String = new String( + Files.readAllBytes( + Paths + .get(".") + .toAbsolutePath + .getParent() + .resolve("src/test/scala/proc/stat") + ), + StandardCharsets.UTF_8 + ) + + test("Method getMetricValue should return specific Double value") { + val metricCollector = new StatMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "cpu_user" + val expectedDoubleValue: Double = + procFileDataTest(originalMetricName) / procFileDataTest.foldLeft(0.0)( + _ + _._2 + ) + + assertResult(expectedDoubleValue) { + metricCollector.getMetricValue(procFileSourceTest, originalMetricName) + } + } + + test("Method getMetricValue should call BufferedSource.getLines") { + val metricCollector = new StatMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "cpu_user" + + when(procFileSourceMock.getLines()) + .thenReturn(procFileSourceTest.getLines()) + + metricCollector.getMetricValue(procFileSourceMock, originalMetricName) + + verify(procFileSourceMock, times(1)).getLines() + } + + test("Method getMetricValue should throw NoSuchElementException") { + val metricCollector = new StatMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "wrong_metric" + + when(procFileSourceMock.getLines()) + .thenReturn(procFileSourceTest.getLines()) + assertThrows[NoSuchElementException]( + metricCollector.getMetricValue(procFileSourceMock, originalMetricName) + ) + } +} + +class StatMetricCollectorSourceTest extends AnyFunSuite with MockFactory { + val procFileContent: String = new String( + Files.readAllBytes( + Paths + .get(".") + .toAbsolutePath + .getParent() + .resolve("src/test/scala/proc/stat") + ), + StandardCharsets.UTF_8 + ) + + test("Method getProcFileSource should return specific BufferedSource") { + val sourceMethod = mockFunction[String, BufferedSource] + val metricCollector = new StatMetricCollector(sourceMethod) + val sourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + sourceMethod + .expects(metricCollector.procFilePath) + .returns(sourceTest) + + assertResult(sourceTest) { + metricCollector.getProcFileSource() + } + } + +} diff --git a/src/test/scala/proc/stat b/src/test/scala/proc/stat new file mode 100644 index 0000000..44c38f2 --- /dev/null +++ b/src/test/scala/proc/stat @@ -0,0 +1,15 @@ +cpu 79242 0 74306 842486413 756859 6140 67701 0 +cpu0 49663 0 40234 104757317 542691 4420 39572 0 +cpu1 2724 0 2118 105420424 767 1719 6084 0 +cpu2 18578 0 18430 105191522 204592 0 714 0 +cpu3 513 0 979 105428698 739 0 2907 0 +cpu4 1623 0 2105 105426291 444 0 3373 0 +cpu5 3491 0 5326 105414798 7134 0 3087 0 +cpu6 1636 0 3081 105420689 201 0 8229 0 +cpu7 1011 0 2029 105426670 288 0 3731 0 +intr 1139300141 1054390414 3 0 5 255 0 0 0 3 0 0 0 4 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 2 3664020 44569070 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 2 3702799 4393655 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 2 3480582 527155 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 2 3445617 3870988 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 6 2 4877935 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 3 2 3445167 0 0 0 0 647 0 0 0 0 0 0 0 0 1 1 1 0 0 0 0 0 2 3 3481606 0 0 0 0 0 0 0 0 0 0 0 0 2004579 1 1 1 0 0 0 0 0 2 3 3445582 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 0 +ctxt 367249552 +btime 1310547399 +processes 107918 +procs_running 1 +procs_blocked 0 \ No newline at end of file From 51a61a1b96ee0e89ea2dfb42478d8484861ebbc6 Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Tue, 7 May 2024 13:53:50 -0300 Subject: [PATCH 03/12] feat: removed io.Source class usage --- .../github/dutrevis/StatMetricCollector.scala | 34 +++++++------------ 1 file changed, 13 insertions(+), 21 deletions(-) diff --git a/src/main/scala/io/github/dutrevis/StatMetricCollector.scala b/src/main/scala/io/github/dutrevis/StatMetricCollector.scala index a7c055f..8a7f20e 100644 --- a/src/main/scala/io/github/dutrevis/StatMetricCollector.scala +++ b/src/main/scala/io/github/dutrevis/StatMetricCollector.scala @@ -1,13 +1,13 @@ package io.github.dutrevis -import scala.io.{Source, BufferedSource} - -class StatMetricCollector( - sourceMethod: String => BufferedSource = Source.fromFile -) extends ProcFileMetricCollector { +class StatMetricCollector extends ProcFileMetricCollector { override val procFilePath = "/proc/stat" + /** A List with the names of the columns present in the `/proc/stat` file. + * They are used as keys of a Map that points to the values collected from + * the file.
+ */ val procFileCPUColumns = List( "cpu_user", "cpu_nice", @@ -18,25 +18,25 @@ class StatMetricCollector( "cpu_softirq" ) - /** Gets the value of a metric from a proc file located at the `procFilePath` - * value set. The metric is searched according to the original name provided - * in the parameter `originalMetricName`.
- * @param procFileSource - * a BufferedSource instance, with access to the desired proc file
+ /** Gets the value of a metric from the content of a proc file provided. The + * metric is searched according to the original name provided in the + * parameter `originalMetricName`.
+ * @param procFileContent + * a String with the content of the desired proc file
* @param originalMetricName * the original metric name by which it is found in the proc file
* @return - * the metric value as Long + * the metric value as Double * @throws NoSuchElementException * if a metric is not found with the provided original name */ override def getMetricValue( - procFileSource: BufferedSource, + procFileContent: String, originalMetricName: String ): Double = { val procFileData = (procFileCPUColumns .zip( - procFileSource.getLines + procFileContent.linesIterator .take(1) .mkString .split(" +") @@ -46,15 +46,7 @@ class StatMetricCollector( .toMap) val metricValue = procFileData(originalMetricName) / procFileData.foldLeft(0.0)(_ + _._2) - procFileSource.close() metricValue } - /** Access and buffers a proc file located at the `procFilePath`.
- * @return - * a `BufferedSource` instance of the file read - */ - override def getProcFileSource(): scala.io.BufferedSource = { - sourceMethod(procFilePath) - } } From e1f146ea0e736c5b256eb5795756c2f38e2d979c Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Tue, 7 May 2024 13:54:33 -0300 Subject: [PATCH 04/12] chore: changed StatMetricCollector tests --- src/test/scala/StatMetricCollectorTest.scala | 75 +++----------------- 1 file changed, 9 insertions(+), 66 deletions(-) diff --git a/src/test/scala/StatMetricCollectorTest.scala b/src/test/scala/StatMetricCollectorTest.scala index 307d39c..e2b521e 100644 --- a/src/test/scala/StatMetricCollectorTest.scala +++ b/src/test/scala/StatMetricCollectorTest.scala @@ -1,14 +1,11 @@ import io.github.dutrevis.StatMetricCollector import org.scalatest.funsuite.AnyFunSuite -import org.scalamock.scalatest.MockFactory import org.mockito.{MockitoSugar, ArgumentMatchersSugar} import org.mockito.integrations.scalatest.ResetMocksAfterEachTest import org.mockito.captor.ArgCaptor -import scala.io.{Source, BufferedSource} -import java.io.ByteArrayInputStream -import java.nio.charset.StandardCharsets -import java.nio.file.{Files, Paths} +import java.nio.charset.{StandardCharsets, Charset} +import java.nio.file.{Files, Path} class StatMetricCollectorTest extends AnyFunSuite @@ -16,9 +13,6 @@ class StatMetricCollectorTest with ArgumentMatchersSugar with ResetMocksAfterEachTest { - // Arrange common mocks - val procFileSourceMock = mock[BufferedSource] - // Arrange common values val procFileDataTest = Map[String, Long]( "cpu_user" -> 79242, @@ -30,10 +24,10 @@ class StatMetricCollectorTest "cpu_softirq" -> 67701 ) - val procFileContent: String = new String( + val procFileContentTest: String = new String( Files.readAllBytes( - Paths - .get(".") + Path + .of(".") .toAbsolutePath .getParent() .resolve("src/test/scala/proc/stat") @@ -42,10 +36,7 @@ class StatMetricCollectorTest ) test("Method getMetricValue should return specific Double value") { - val metricCollector = new StatMetricCollector() - val procFileSourceTest = (new BufferedSource( - new ByteArrayInputStream(procFileContent.getBytes) - )) + val metricCollector = new StatMetricCollector val originalMetricName: String = "cpu_user" val expectedDoubleValue: Double = procFileDataTest(originalMetricName) / procFileDataTest.foldLeft(0.0)( @@ -53,65 +44,17 @@ class StatMetricCollectorTest ) assertResult(expectedDoubleValue) { - metricCollector.getMetricValue(procFileSourceTest, originalMetricName) + metricCollector.getMetricValue(procFileContentTest, originalMetricName) } } - test("Method getMetricValue should call BufferedSource.getLines") { - val metricCollector = new StatMetricCollector() - val procFileSourceTest = (new BufferedSource( - new ByteArrayInputStream(procFileContent.getBytes) - )) - val originalMetricName: String = "cpu_user" - - when(procFileSourceMock.getLines()) - .thenReturn(procFileSourceTest.getLines()) - - metricCollector.getMetricValue(procFileSourceMock, originalMetricName) - - verify(procFileSourceMock, times(1)).getLines() - } - test("Method getMetricValue should throw NoSuchElementException") { - val metricCollector = new StatMetricCollector() - val procFileSourceTest = (new BufferedSource( - new ByteArrayInputStream(procFileContent.getBytes) - )) + val metricCollector = new StatMetricCollector val originalMetricName: String = "wrong_metric" - when(procFileSourceMock.getLines()) - .thenReturn(procFileSourceTest.getLines()) assertThrows[NoSuchElementException]( - metricCollector.getMetricValue(procFileSourceMock, originalMetricName) + metricCollector.getMetricValue(procFileContentTest, originalMetricName) ) } -} - -class StatMetricCollectorSourceTest extends AnyFunSuite with MockFactory { - val procFileContent: String = new String( - Files.readAllBytes( - Paths - .get(".") - .toAbsolutePath - .getParent() - .resolve("src/test/scala/proc/stat") - ), - StandardCharsets.UTF_8 - ) - - test("Method getProcFileSource should return specific BufferedSource") { - val sourceMethod = mockFunction[String, BufferedSource] - val metricCollector = new StatMetricCollector(sourceMethod) - val sourceTest = (new BufferedSource( - new ByteArrayInputStream(procFileContent.getBytes) - )) - sourceMethod - .expects(metricCollector.procFilePath) - .returns(sourceTest) - - assertResult(sourceTest) { - metricCollector.getProcFileSource() - } - } } From e8a75486d663ca046fe9f147705810b727311c2f Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 17 Apr 2024 15:20:48 -0300 Subject: [PATCH 05/12] feat: Created MeminfoMetricCollector class --- .../dutrevis/MeminfoMetricCollector.scala | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 src/main/scala/io/github/dutrevis/MeminfoMetricCollector.scala diff --git a/src/main/scala/io/github/dutrevis/MeminfoMetricCollector.scala b/src/main/scala/io/github/dutrevis/MeminfoMetricCollector.scala new file mode 100644 index 0000000..eba66ef --- /dev/null +++ b/src/main/scala/io/github/dutrevis/MeminfoMetricCollector.scala @@ -0,0 +1,44 @@ +package io.github.dutrevis + +import scala.io.{Source, BufferedSource} + +class MeminfoMetricCollector( + sourceMethod: String => BufferedSource = Source.fromFile +) extends ProcFileMetricCollector { + + override val procFilePath = "/proc/meminfo" + + /** Gets the value of a metric from a proc file located at the `procFilePath` + * value set. The metric is searched according to the original name provided + * in the parameter `originalMetricName`.
+ * @param procFileSource + * a BufferedSource instance, with access to the desired proc file
+ * @param originalMetricName + * the original metric name by which it is found in the proc file
+ * @return + * the metric value as Long + * @throws NoSuchElementException + * if a metric is not found with the provided original name + */ + override def getMetricValue( + procFileSource: BufferedSource, + originalMetricName: String + ): Long = { + val procFileData = procFileSource.getLines + .filter(metricLine => metricLine.contains(originalMetricName)) + .map { case s: String => s.split(":") } + .map { case Array(k, v) => k -> v.trim().split(" ").head.toLong } + .toMap + val metricValue = procFileData(originalMetricName) + procFileSource.close() + metricValue + } + + /** Access and buffers a proc file located at the `procFilePath`.
+ * @return + * a `BufferedSource` instance of the file read + */ + override def getProcFileSource(): scala.io.BufferedSource = { + sourceMethod(procFilePath) + } +} From ec2f1f08fa8e1c819a5726de5223657f1e943514 Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 17 Apr 2024 15:21:37 -0300 Subject: [PATCH 06/12] chore: Added MeminfoMetricCollector tests --- .../scala/MeminfoMetricCollectorTest.scala | 103 ++++++++++++++++++ src/test/scala/meminfo.test | 42 +++++++ 2 files changed, 145 insertions(+) create mode 100644 src/test/scala/MeminfoMetricCollectorTest.scala create mode 100644 src/test/scala/meminfo.test diff --git a/src/test/scala/MeminfoMetricCollectorTest.scala b/src/test/scala/MeminfoMetricCollectorTest.scala new file mode 100644 index 0000000..46a70a3 --- /dev/null +++ b/src/test/scala/MeminfoMetricCollectorTest.scala @@ -0,0 +1,103 @@ +import io.github.dutrevis.MeminfoMetricCollector +import org.scalatest.funsuite.AnyFunSuite +import org.scalamock.scalatest.MockFactory +import org.mockito.{MockitoSugar, ArgumentMatchersSugar} +import org.mockito.integrations.scalatest.ResetMocksAfterEachTest +import org.mockito.captor.ArgCaptor +import scala.io.{Source, BufferedSource} + +import java.io.ByteArrayInputStream +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Paths} + +class MeminfoMetricCollectorTest + extends AnyFunSuite + with MockitoSugar + with ArgumentMatchersSugar + with ResetMocksAfterEachTest { + + // Arrange common mocks + val procFileSourceMock = mock[BufferedSource] + + val procFileContent: String = new String( + Files.readAllBytes( + Paths + .get(".") + .toAbsolutePath + .getParent() + .resolve("src/test/scala/meminfo.test") + ), + StandardCharsets.UTF_8 + ) + + test("Method getMetricValue should return specific Long value") { + val metricCollector = new MeminfoMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "MemTotal" + val expectedLongValue: Long = 1921988 + + assertResult(expectedLongValue) { + metricCollector.getMetricValue(procFileSourceTest, originalMetricName) + } + } + + test("Method getMetricValue should call BufferedSource.getLines") { + val metricCollector = new MeminfoMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "MemTotal" + + when(procFileSourceMock.getLines()) + .thenReturn(procFileSourceTest.getLines()) + + metricCollector.getMetricValue(procFileSourceMock, originalMetricName) + + verify(procFileSourceMock, times(1)).getLines() + } + + test("Method getMetricValue should throw NoSuchElementException") { + val metricCollector = new MeminfoMetricCollector() + val procFileSourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + val originalMetricName: String = "WrongMetric" + + when(procFileSourceMock.getLines()) + .thenReturn(procFileSourceTest.getLines()) + assertThrows[NoSuchElementException]( + metricCollector.getMetricValue(procFileSourceMock, originalMetricName) + ) + } +} + +class MeminfoMetricCollectorSourceTest extends AnyFunSuite with MockFactory { + val procFileContent: String = new String( + Files.readAllBytes( + Paths + .get(".") + .toAbsolutePath + .getParent() + .resolve("src/test/scala/meminfo.test") + ), + StandardCharsets.UTF_8 + ) + + test("Method getProcFileSource should return specific BufferedSource") { + val sourceMethod = mockFunction[String, BufferedSource] + val metricCollector = new MeminfoMetricCollector(sourceMethod) + val sourceTest = (new BufferedSource( + new ByteArrayInputStream(procFileContent.getBytes) + )) + sourceMethod + .expects(metricCollector.procFilePath) + .returns(sourceTest) + + assertResult(sourceTest) { + metricCollector.getProcFileSource() + } + } + +} diff --git a/src/test/scala/meminfo.test b/src/test/scala/meminfo.test new file mode 100644 index 0000000..66305b9 --- /dev/null +++ b/src/test/scala/meminfo.test @@ -0,0 +1,42 @@ +MemTotal: 1921988 kB +MemFree: 1374408 kB +Buffers: 32688 kB +Cached: 370540 kB +SwapCached: 0 kB +Active: 344604 kB +Inactive: 80800 kB +Active(anon): 22364 kB +Inactive(anon): 4 kB +Active(file): 322240 kB +Inactive(file): 80796 kB +Unevictable: 0 kB +Mlocked: 0 kB +SwapTotal: 1048572 kB +SwapFree: 1048572 kB +Dirty: 48 kB +Writeback: 0 kB +AnonPages: 22260 kB +Mapped: 13628 kB +Shmem: 196 kB +Slab: 91648 kB +SReclaimable: 34024 kB +SUnreclaim: 57624 kB +KernelStack: 2880 kB +PageTables: 3620 kB +NFS_Unstable: 0 kB +Bounce: 0 kB +WritebackTmp: 0 kB +CommitLimit: 2009564 kB +Committed_AS: 134216 kB +VmallocTotal: 34359738367 kB +VmallocUsed: 12276 kB +VmallocChunk: 34359712840 kB +HardwareCorrupted: 0 kB +AnonHugePages: 0 kB +HugePages_Total: 0 +HugePages_Free: 0 +HugePages_Rsvd: 0 +HugePages_Surp: 0 +Hugepagesize: 2048 kB +DirectMap4k: 8064 kB +DirectMap2M: 2088960 kB From db08e265e371a6d0cde0a7f82580cc4bcef50946 Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Thu, 8 Feb 2024 16:37:54 -0300 Subject: [PATCH 07/12] feat: Added CPUMetrics class --- .../scala/io/github/dutrevis/CPUMetrics.scala | 173 ++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100644 src/main/scala/io/github/dutrevis/CPUMetrics.scala diff --git a/src/main/scala/io/github/dutrevis/CPUMetrics.scala b/src/main/scala/io/github/dutrevis/CPUMetrics.scala new file mode 100644 index 0000000..84eb85f --- /dev/null +++ b/src/main/scala/io/github/dutrevis/CPUMetrics.scala @@ -0,0 +1,173 @@ +package io.github.dutrevis + +import java.util.{Map => JMap} +import scala.collection.JavaConverters._ + +import com.codahale.metrics.{Gauge, MetricRegistry} +import org.apache.spark.SparkContext +import org.apache.spark.api.plugin.{ + DriverPlugin, + ExecutorPlugin, + PluginContext, + SparkPlugin +} + +import scala.io.Source + +// Collects CPU resource metrics from the unix-based operating system. +// Use when running Spark with fisical clusters. + +class CPUMetrics extends SparkPlugin { + + def registerUserCPU(metricRegistry: MetricRegistry): Unit = { + metricRegistry.register( + MetricRegistry.name("UserCPU"), + new Gauge[Double] { + override def getValue: Double = { + val procFile = Source.fromFile(CPUMetrics.procFileName) + val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong)).toMap + val metricValue = + procFileData("cpu_user") / procFileData.foldLeft(0.0)(_ + _._2) + procFile.close() + metricValue + } + } + ) + } + + def registerNiceCPU(metricRegistry: MetricRegistry): Unit = { + metricRegistry.register( + MetricRegistry.name("NiceCPU"), + new Gauge[Double] { + override def getValue: Double = { + val procFile = Source.fromFile(CPUMetrics.procFileName) + val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong)).toMap + val metricValue = + procFileData("cpu_nice") / procFileData.foldLeft(0.0)(_ + _._2) + procFile.close() + metricValue + } + } + ) + } + + def registerSystemCPU(metricRegistry: MetricRegistry): Unit = { + metricRegistry.register( + MetricRegistry.name("SystemCPU"), + new Gauge[Double] { + override def getValue: Double = { + val procFile = Source.fromFile(CPUMetrics.procFileName) + val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong)).toMap + val metricValue = + procFileData("cpu_system") / procFileData.foldLeft(0.0)(_ + _._2) + procFile.close() + metricValue + } + } + ) + } + + def registerIdleCPU(metricRegistry: MetricRegistry): Unit = { + metricRegistry.register( + MetricRegistry.name("IdleCPU"), + new Gauge[Double] { + override def getValue: Double = { + val procFile = Source.fromFile(CPUMetrics.procFileName) + val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong)).toMap + val metricValue = + procFileData("cpu_idle") / procFileData.foldLeft(0.0)(_ + _._2) + procFile.close() + metricValue + } + } + ) + } + + def registerWaitCPU(metricRegistry: MetricRegistry): Unit = { + metricRegistry.register( + MetricRegistry.name("WaitCPU"), + new Gauge[Double] { + override def getValue: Double = { + val procFile = Source.fromFile(CPUMetrics.procFileName) + val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines + .take(1) + .mkString + .split(" +") + .tail + .map(_.toLong)).toMap + val metricValue = + procFileData("cpu_iowait") / procFileData.foldLeft(0.0)(_ + _._2) + procFile.close() + metricValue + } + } + ) + } + + // Return the plugin's driver-side component. + override def driverPlugin(): DriverPlugin = { + new DriverPlugin() { + override def init( + sc: SparkContext, + myContext: PluginContext + ): JMap[String, String] = { + registerUserCPU(myContext.metricRegistry) + registerNiceCPU(myContext.metricRegistry) + registerSystemCPU(myContext.metricRegistry) + registerIdleCPU(myContext.metricRegistry) + registerWaitCPU(myContext.metricRegistry) + Map.empty[String, String].asJava + } + } + } + + // Return the plugin's executor-side component. + override def executorPlugin(): ExecutorPlugin = { + new ExecutorPlugin() { + override def init( + myContext: PluginContext, + extraConf: JMap[String, String] + ): Unit = { + registerUserCPU(myContext.metricRegistry) + registerNiceCPU(myContext.metricRegistry) + registerSystemCPU(myContext.metricRegistry) + registerIdleCPU(myContext.metricRegistry) + registerWaitCPU(myContext.metricRegistry) + } + } + } + +} + +object CPUMetrics { + private val procFileName = "/proc/stat" + private val cpuColumns = List( + "cpu_user", + "cpu_nice", + "cpu_system", + "cpu_idle", + "cpu_iowait", + "cpu_irq", + "cpu_softirq" + ) +} From 2faacca135aa43dda1c4c9a30f1ce4297269a2ee Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 21 Feb 2024 19:15:33 -0300 Subject: [PATCH 08/12] Added detailed docstrings --- .../scala/io/github/dutrevis/CPUMetrics.scala | 67 +++++++++++++++++-- 1 file changed, 62 insertions(+), 5 deletions(-) diff --git a/src/main/scala/io/github/dutrevis/CPUMetrics.scala b/src/main/scala/io/github/dutrevis/CPUMetrics.scala index 84eb85f..17d0199 100644 --- a/src/main/scala/io/github/dutrevis/CPUMetrics.scala +++ b/src/main/scala/io/github/dutrevis/CPUMetrics.scala @@ -2,6 +2,7 @@ package io.github.dutrevis import java.util.{Map => JMap} import scala.collection.JavaConverters._ +import scala.io.Source import com.codahale.metrics.{Gauge, MetricRegistry} import org.apache.spark.SparkContext @@ -12,13 +13,33 @@ import org.apache.spark.api.plugin.{ SparkPlugin } -import scala.io.Source - -// Collects CPU resource metrics from the unix-based operating system. -// Use when running Spark with fisical clusters. - +/** + * Collects CPU resource metrics from a unix-based operating system. + *
+ * Use when Spark is running in clusters with standalone, Mesos or YARN resource managers. + *
+ * CPU metrics are obtained from the numbers of the first line of the `/proc/stat` file, + * available at the proc pseudo-filesystem of unix-based operating systems. + * These numbers identify the amount of time the CPU has spent performing + * different kinds of work, arranged in columns at the following order: + * "cpu_user", "cpu_nice", "cpu_system", "cpu_idle", "cpu_iowait", "cpu_irq" and "cpu_softirq". + *
+ * @note All of the numbers retrieved are aggregates since the system first booted. + * @note Time units are in USER_HZ or Jiffies (typically hundredths of a second) + * @note Values for "cpu_steal", "cpu_guest" and "cpu_guest_nice", available at spectific + * Linux versions, are not parsed from the file. + */ class CPUMetrics extends SparkPlugin { + /** + * Collects the aggregated CPU usage value for normal processes executing in user mode, + * calculates its average out of the total of CPU usage time for all processes and + * registers the result into the provided MetricRegistry. + *
+ * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + *
+ * @note Registered value is of type LONG with precision of 2. + */ def registerUserCPU(metricRegistry: MetricRegistry): Unit = { metricRegistry.register( MetricRegistry.name("UserCPU"), @@ -40,6 +61,15 @@ class CPUMetrics extends SparkPlugin { ) } + /** + * Collects the aggregated CPU usage value for niced processes (there is, run with the + * nice command) executing in user mode, calculates its average out of the total of the + * CPU usage time for all processes and registers the result into the provided MetricRegistry. + *
+ * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + *
+ * @note Registered value is of type LONG with precision of 2. + */ def registerNiceCPU(metricRegistry: MetricRegistry): Unit = { metricRegistry.register( MetricRegistry.name("NiceCPU"), @@ -61,6 +91,15 @@ class CPUMetrics extends SparkPlugin { ) } + /** + * Collects the aggregated CPU usage value processes executing in kernel mode, + * calculates its average out of the total of the CPU usage time for all processes + * and registers the result into the provided MetricRegistry. + *
+ * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + *
+ * @note Registered value is of type LONG with precision of 2. + */ def registerSystemCPU(metricRegistry: MetricRegistry): Unit = { metricRegistry.register( MetricRegistry.name("SystemCPU"), @@ -82,6 +121,15 @@ class CPUMetrics extends SparkPlugin { ) } + /** + * Collects the aggregated CPU usage value for when no processes are running, + * calculates its average out of the total of the CPU usage time for all processes + * and registers the result into the provided MetricRegistry. + *
+ * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + *
+ * @note Registered value is of type LONG with precision of 2. + */ def registerIdleCPU(metricRegistry: MetricRegistry): Unit = { metricRegistry.register( MetricRegistry.name("IdleCPU"), @@ -103,6 +151,15 @@ class CPUMetrics extends SparkPlugin { ) } + /** + * Collects the aggregated CPU usage value for when it's waiting for I/O to complete, + * calculates its average out of the total of the CPU usage time for all processes + * and registers the result into the provided MetricRegistry. + *
+ * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + *
+ * @note Registered value is of type LONG with precision of 2. + */ def registerWaitCPU(metricRegistry: MetricRegistry): Unit = { metricRegistry.register( MetricRegistry.name("WaitCPU"), From 09c0874fb76edee4ba8705e91f847401fda19673 Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 17 Apr 2024 20:53:39 -0300 Subject: [PATCH 09/12] feat: Decoupled CPUMetrics methods --- .../scala/io/github/dutrevis/CPUMetrics.scala | 374 +++++++++--------- 1 file changed, 198 insertions(+), 176 deletions(-) diff --git a/src/main/scala/io/github/dutrevis/CPUMetrics.scala b/src/main/scala/io/github/dutrevis/CPUMetrics.scala index 17d0199..32a3040 100644 --- a/src/main/scala/io/github/dutrevis/CPUMetrics.scala +++ b/src/main/scala/io/github/dutrevis/CPUMetrics.scala @@ -4,7 +4,7 @@ import java.util.{Map => JMap} import scala.collection.JavaConverters._ import scala.io.Source -import com.codahale.metrics.{Gauge, MetricRegistry} +import com.codahale.metrics.{Gauge, Metric, MetricRegistry} import org.apache.spark.SparkContext import org.apache.spark.api.plugin.{ DriverPlugin, @@ -13,218 +13,240 @@ import org.apache.spark.api.plugin.{ SparkPlugin } -/** - * Collects CPU resource metrics from a unix-based operating system. - *
- * Use when Spark is running in clusters with standalone, Mesos or YARN resource managers. - *
- * CPU metrics are obtained from the numbers of the first line of the `/proc/stat` file, - * available at the proc pseudo-filesystem of unix-based operating systems. - * These numbers identify the amount of time the CPU has spent performing - * different kinds of work, arranged in columns at the following order: - * "cpu_user", "cpu_nice", "cpu_system", "cpu_idle", "cpu_iowait", "cpu_irq" and "cpu_softirq". - *
- * @note All of the numbers retrieved are aggregates since the system first booted. - * @note Time units are in USER_HZ or Jiffies (typically hundredths of a second) - * @note Values for "cpu_steal", "cpu_guest" and "cpu_guest_nice", available at spectific - * Linux versions, are not parsed from the file. +/** Collects CPU resource metrics from a unix-based operating system.
Use + * when Spark is running in clusters with standalone, Mesos or YARN resource + * managers.
CPU metrics are obtained from the numbers of the first line of + * the `/proc/stat` file, available at the proc pseudo-filesystem of unix-based + * operating systems. These numbers identify the amount of time the CPU has + * spent performing different kinds of work, arranged in columns at the + * following order: "cpu_user", "cpu_nice", "cpu_system", "cpu_idle", + * "cpu_iowait", "cpu_irq" and "cpu_softirq".
+ * @param metricCollector + * a StatMetricCollector instance
+ * @note + * All of the numbers retrieved are aggregates since the system first booted. + * @note + * Time units are in USER_HZ or Jiffies (typically hundredths of a second) + * @note + * Values for "cpu_steal", "cpu_guest" and "cpu_guest_nice", available at + * spectific Linux versions, are not parsed from the file. */ -class CPUMetrics extends SparkPlugin { +class CPUMetrics( + val metricCollector: StatMetricCollector = new StatMetricCollector() +) extends SparkPlugin { - /** - * Collects the aggregated CPU usage value for normal processes executing in user mode, - * calculates its average out of the total of CPU usage time for all processes and - * registers the result into the provided MetricRegistry. - *
- * @param metricRegistry a MetricRegistry instance from dropwizard.metrics - *
- * @note Registered value is of type LONG with precision of 2. + /** Maps the collector methods to their respective metric names, that will be + * displayed in the Dropwizard's metric system. */ - def registerUserCPU(metricRegistry: MetricRegistry): Unit = { - metricRegistry.register( - MetricRegistry.name("UserCPU"), - new Gauge[Double] { - override def getValue: Double = { - val procFile = Source.fromFile(CPUMetrics.procFileName) - val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines - .take(1) - .mkString - .split(" +") - .tail - .map(_.toLong)).toMap - val metricValue = - procFileData("cpu_user") / procFileData.foldLeft(0.0)(_ + _._2) - procFile.close() - metricValue - } - } - ) + val metricMapping = Map[String, (StatMetricCollector) => Double]( + "UserCPU" -> collectUserCPU, + "NiceCPU" -> collectNiceCPU, + "SystemCPU" -> collectSystemCPU, + "IdleCPU" -> collectIdleCPU, + "WaitCPU" -> collectWaitCPU + ) + + /** Registers a provided Dropwizard's Metric instance into a metric registry + * under a metric name.
+ * @param metricRegistry + * a MetricRegistry instance from dropwizard.metrics
+ * @param metricName + * a metric name as a String
+ * @param metricInstance + * an instance of a dropwizard's Metric class to be registered
+ * @throws IllegalArgumentException + * if the metric name is already registered + */ + def registerMetric( + metricRegistry: MetricRegistry, + metricName: String, + metricInstance: Metric + ): Unit = { + metricRegistry.register(metricName, metricInstance) + () } - /** - * Collects the aggregated CPU usage value for niced processes (there is, run with the - * nice command) executing in user mode, calculates its average out of the total of the - * CPU usage time for all processes and registers the result into the provided MetricRegistry. - *
- * @param metricRegistry a MetricRegistry instance from dropwizard.metrics - *
- * @note Registered value is of type LONG with precision of 2. + /** Creates a Dropwizard's Gauge metric type - an instantaneous reading of a + * particular value -, setting the provided collector method as the + * `getValue` method of the Gauge instance.
+ * @param metricCollector + * a StatMetricCollector instance
+ * @param collectorMethod + * a method that receives the metricCollector and returns a metric value as + * a Double
*/ - def registerNiceCPU(metricRegistry: MetricRegistry): Unit = { - metricRegistry.register( - MetricRegistry.name("NiceCPU"), - new Gauge[Double] { - override def getValue: Double = { - val procFile = Source.fromFile(CPUMetrics.procFileName) - val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines - .take(1) - .mkString - .split(" +") - .tail - .map(_.toLong)).toMap - val metricValue = - procFileData("cpu_nice") / procFileData.foldLeft(0.0)(_ + _._2) - procFile.close() - metricValue - } - } - ) + def createGaugeMetric( + metricCollector: StatMetricCollector, + collectorMethod: (StatMetricCollector) => Double + ): Gauge[Double] = { + new Gauge[Double] { + override def getValue: Double = { collectorMethod(metricCollector) } + } } - /** - * Collects the aggregated CPU usage value processes executing in kernel mode, - * calculates its average out of the total of the CPU usage time for all processes - * and registers the result into the provided MetricRegistry. - *
- * @param metricRegistry a MetricRegistry instance from dropwizard.metrics - *
- * @note Registered value is of type LONG with precision of 2. + /** Collects the aggregated CPU usage value for normal processes executing in + * user mode, as an average out of the total of CPU usage time for all + * processes.
+ * @param metricCollector + * a StatMetricCollector instance
+ * @note + * Collected value is of type Double with precision of 2. */ - def registerSystemCPU(metricRegistry: MetricRegistry): Unit = { - metricRegistry.register( - MetricRegistry.name("SystemCPU"), - new Gauge[Double] { - override def getValue: Double = { - val procFile = Source.fromFile(CPUMetrics.procFileName) - val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines - .take(1) - .mkString - .split(" +") - .tail - .map(_.toLong)).toMap - val metricValue = - procFileData("cpu_system") / procFileData.foldLeft(0.0)(_ + _._2) - procFile.close() - metricValue - } - } - ) + def collectUserCPU(metricCollector: StatMetricCollector): Double = { + val procFileSource = metricCollector.getProcFileSource() + val originalMetricName: String = "cpu_user" + metricCollector.getMetricValue(procFileSource, originalMetricName) } - /** - * Collects the aggregated CPU usage value for when no processes are running, - * calculates its average out of the total of the CPU usage time for all processes - * and registers the result into the provided MetricRegistry. - *
- * @param metricRegistry a MetricRegistry instance from dropwizard.metrics - *
- * @note Registered value is of type LONG with precision of 2. + /** Collects the aggregated CPU usage value for niced processes (there is, run + * with the nice command) executing in user mode and calculates its average + * out of the total of the CPU usage time for all processes.
+ * @param metricCollector + * a StatMetricCollector instance
+ * @note + * Collected value is of type Double with precision of 2. */ - def registerIdleCPU(metricRegistry: MetricRegistry): Unit = { - metricRegistry.register( - MetricRegistry.name("IdleCPU"), - new Gauge[Double] { - override def getValue: Double = { - val procFile = Source.fromFile(CPUMetrics.procFileName) - val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines - .take(1) - .mkString - .split(" +") - .tail - .map(_.toLong)).toMap - val metricValue = - procFileData("cpu_idle") / procFileData.foldLeft(0.0)(_ + _._2) - procFile.close() - metricValue - } - } - ) + def collectNiceCPU(metricCollector: StatMetricCollector): Double = { + val procFileSource = metricCollector.getProcFileSource() + val originalMetricName: String = "cpu_nice" + metricCollector.getMetricValue(procFileSource, originalMetricName) } - /** - * Collects the aggregated CPU usage value for when it's waiting for I/O to complete, - * calculates its average out of the total of the CPU usage time for all processes - * and registers the result into the provided MetricRegistry. + /** Collects the aggregated CPU usage value processes executing in kernel mode + * as an average out of the total of the CPU usage time for all processes. *
- * @param metricRegistry a MetricRegistry instance from dropwizard.metrics + * @param metricCollector + * a StatMetricCollector instance
+ * @note + * Collected value is of type Double with precision of 2. + */ + def collectSystemCPU(metricCollector: StatMetricCollector): Double = { + val procFileSource = metricCollector.getProcFileSource() + val originalMetricName: String = "cpu_system" + metricCollector.getMetricValue(procFileSource, originalMetricName) + } + + /** Collects the aggregated CPU usage value for when no processes are running, + * as an average out of the total of the CPU usage time for all processes. *
- * @note Registered value is of type LONG with precision of 2. + * @param metricCollector + * a StatMetricCollector instance
+ * @note + * Collected value is of type Double with precision of 2. */ - def registerWaitCPU(metricRegistry: MetricRegistry): Unit = { - metricRegistry.register( - MetricRegistry.name("WaitCPU"), - new Gauge[Double] { - override def getValue: Double = { - val procFile = Source.fromFile(CPUMetrics.procFileName) - val procFileData = (CPUMetrics.cpuColumns zip procFile.getLines - .take(1) - .mkString - .split(" +") - .tail - .map(_.toLong)).toMap - val metricValue = - procFileData("cpu_iowait") / procFileData.foldLeft(0.0)(_ + _._2) - procFile.close() - metricValue - } - } - ) + def collectIdleCPU(metricCollector: StatMetricCollector): Double = { + val procFileSource = metricCollector.getProcFileSource() + val originalMetricName: String = "cpu_idle" + metricCollector.getMetricValue(procFileSource, originalMetricName) + } + + /** Collects the aggregated CPU usage value for when it's waiting for I/O to + * complete, as an average out of the total of the CPU usage time for all + * processes.
+ * @param metricCollector + * a StatMetricCollector instance
+ * @note + * Collected value is of type Double with precision of 2. + */ + def collectWaitCPU(metricCollector: StatMetricCollector): Double = { + val procFileSource = metricCollector.getProcFileSource() + val originalMetricName: String = "cpu_iowait" + metricCollector.getMetricValue(procFileSource, originalMetricName) } - // Return the plugin's driver-side component. + /** Returns the plugin's driver-side component. The returned DriverPlugin + * instance is called once, early in the initialization of the Spark driver. + * The operation it performs consists in the sequential registration of each + * mapped metric as a Gauge metric into an existing metric Registry. A test + * call is executed once on each collector method before its registration, + * assuring that the metric is available to be read and collected from the + * local OS, thus preventing future errors when the Gauge is first executed + * by the metrics system. If a `NoSuchElementException` is thrown in this + * attempt, the method is not registered, enabling the registration of the + * subsequent mapped metrics by the plugin. The plugin component ends its + * execution once all metrics are registered, leaving to the Dropwizard's + * Metrics system the job of collecting and exporting the registered metrics + * in a pre or user-defined frequency.
+ * @note + * The driver's initialization is blocked during the operations inside + * `init`, so heavy performing operations must be avoided. + * @note + * The overriden `init` method must return a Map, that will be provided as + * `extraConf` to an `ExecutorPlugin` instance. + * @return + * An instance of the `DriverPlugin` + */ override def driverPlugin(): DriverPlugin = { new DriverPlugin() { override def init( sc: SparkContext, myContext: PluginContext ): JMap[String, String] = { - registerUserCPU(myContext.metricRegistry) - registerNiceCPU(myContext.metricRegistry) - registerSystemCPU(myContext.metricRegistry) - registerIdleCPU(myContext.metricRegistry) - registerWaitCPU(myContext.metricRegistry) + for ( + ( + metricName: String, + collectorMethod: ((StatMetricCollector) => Double) + ) <- + metricMapping + ) + try { + var testCall = collectorMethod(metricCollector) + registerMetric( + myContext.metricRegistry, + MetricRegistry.name(metricName), + createGaugeMetric(metricCollector, collectorMethod) + ) + } catch { + case e: NoSuchElementException => () + } Map.empty[String, String].asJava } } } - // Return the plugin's executor-side component. + /** Returns the plugin's executor-side component. The returned ExecutorPlugin + * instance is called once, early in the initialization of the executor + * process. The operation it performs consists in the sequential registration + * of each mapped metric as a Gauge metric into an existing metric Registry. + * A test call is executed once on each collector method before its + * registration, assuring that the metric is available to be read and + * collected from the local OS, thus preventing future errors when the Gauge + * is first executed by the metrics system. If a `NoSuchElementException` is + * thrown in this attempt, the method is not registered, enabling the + * registration of the subsequent mapped metrics by the plugin. The plugin + * component ends its execution once all metrics are registered, leaving to + * the Dropwizard's Metrics system the job of collecting and exporting the + * registered metrics in a pre or user-defined frequency.
+ * @note + * The executor's initialization is blocked during the operations inside + * `init`, so heavy performing operations must be avoided. + * @return + * An instance of the `ExecutorPlugin` Unit + */ override def executorPlugin(): ExecutorPlugin = { new ExecutorPlugin() { override def init( myContext: PluginContext, extraConf: JMap[String, String] ): Unit = { - registerUserCPU(myContext.metricRegistry) - registerNiceCPU(myContext.metricRegistry) - registerSystemCPU(myContext.metricRegistry) - registerIdleCPU(myContext.metricRegistry) - registerWaitCPU(myContext.metricRegistry) + for ( + ( + metricName: String, + collectorMethod: ((StatMetricCollector) => Double) + ) <- + metricMapping + ) + try { + var testCall = collectorMethod(metricCollector) + registerMetric( + myContext.metricRegistry, + MetricRegistry.name(metricName), + createGaugeMetric(metricCollector, collectorMethod) + ) + } catch { + case e: NoSuchElementException => () + } } } } - -} - -object CPUMetrics { - private val procFileName = "/proc/stat" - private val cpuColumns = List( - "cpu_user", - "cpu_nice", - "cpu_system", - "cpu_idle", - "cpu_iowait", - "cpu_irq", - "cpu_softirq" - ) } From c9339377abb8521f72d8b14ec26a18f806b05ecd Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Wed, 17 Apr 2024 21:27:02 -0300 Subject: [PATCH 10/12] chore: Added CPUMetrics tests --- src/test/scala/CPUMetricsTest.scala | 169 ++++++++++++++++++++++++++++ 1 file changed, 169 insertions(+) create mode 100644 src/test/scala/CPUMetricsTest.scala diff --git a/src/test/scala/CPUMetricsTest.scala b/src/test/scala/CPUMetricsTest.scala new file mode 100644 index 0000000..0bfeab8 --- /dev/null +++ b/src/test/scala/CPUMetricsTest.scala @@ -0,0 +1,169 @@ +import io.github.dutrevis.{CPUMetrics, StatMetricCollector} +import org.scalatest.funsuite.AnyFunSuite +import org.mockito.{MockitoSugar, ArgumentMatchersSugar} +import org.mockito.captor.ArgCaptor +import org.mockito.integrations.scalatest.ResetMocksAfterEachTest +import com.codahale.metrics.{Gauge, Metric, MetricRegistry} +import scala.io.BufferedSource + +class CPUMetricsTest + extends AnyFunSuite + with MockitoSugar + with ArgumentMatchersSugar + with ResetMocksAfterEachTest { + + // Arrange common mocks + val gaugeMock = mock[Gauge[Double]] + val metricMock = mock[Metric] + val metricRegistryMock = mock[MetricRegistry] + val procFileSourceMock = mock[BufferedSource] + val statMetricCollectorMock = mock[StatMetricCollector] + + test("Method createGaugeMetric should return Gauge[Double]") { + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val collectorMethod = (s: StatMetricCollector) => { 123456.toDouble } + + val returnedGaugeMetric = + cpuMetrics.createGaugeMetric(statMetricCollectorMock, collectorMethod) + assert(returnedGaugeMetric.isInstanceOf[Gauge[Double]]) + } + + test("Method registerMetric should call register") { + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val metricName: String = "any_metric" + + when(metricRegistryMock.register(any[String], any[Metric])) + .thenReturn(metricMock) + cpuMetrics.registerMetric(metricRegistryMock, metricName, gaugeMock) + + verify(metricRegistryMock, times(1)).register(metricName, gaugeMock) + } + + test("Method collectUserCPU should call getMetricValue") { + val originalMetricName: String = "cpu_user" + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + cpuMetrics.collectUserCPU(statMetricCollectorMock) + verify(statMetricCollectorMock, times(1)) + .getMetricValue(procFileSourceMock, originalMetricName) + } + + test("Method collectUserCPU should call getMetricValue with args") { + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val originalMetricName: String = "cpu_user" + val procFileSourceCaptor = ArgCaptor[BufferedSource] + val originalMetricNameCaptor = ArgCaptor[String] + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + cpuMetrics.collectUserCPU(statMetricCollectorMock) + verify(statMetricCollectorMock).getMetricValue( + procFileSourceCaptor, + originalMetricNameCaptor + ) + procFileSourceCaptor hasCaptured procFileSourceMock + originalMetricNameCaptor hasCaptured originalMetricName + } + + test("Method collectUserCPU should return Double") { + val originalMetricName: String = "cpu_user" + val expectedDoubleValue: Double = 79242 + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + when( + statMetricCollectorMock.getMetricValue( + procFileSourceMock, + originalMetricName + ) + ) + .thenReturn(expectedDoubleValue) + + assertResult(expectedDoubleValue) { + cpuMetrics.collectUserCPU(statMetricCollectorMock) + } + } + + test("Method collectNiceCPU should return Double") { + val originalMetricName: String = "cpu_nice" + val expectedDoubleValue: Double = 0 + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + when( + statMetricCollectorMock.getMetricValue( + procFileSourceMock, + originalMetricName + ) + ) + .thenReturn(expectedDoubleValue) + + assertResult(expectedDoubleValue) { + cpuMetrics.collectNiceCPU(statMetricCollectorMock) + } + } + + test("Method collectSystemCPU should return Double") { + val originalMetricName: String = "cpu_system" + val expectedDoubleValue: Double = 74306 + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + when( + statMetricCollectorMock.getMetricValue( + procFileSourceMock, + originalMetricName + ) + ) + .thenReturn(expectedDoubleValue) + + assertResult(expectedDoubleValue) { + cpuMetrics.collectSystemCPU(statMetricCollectorMock) + } + } + + test("Method collectIdleCPU should return Double") { + val originalMetricName: String = "cpu_idle" + val expectedDoubleValue: Double = 842486413 + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + when( + statMetricCollectorMock.getMetricValue( + procFileSourceMock, + originalMetricName + ) + ) + .thenReturn(expectedDoubleValue) + + assertResult(expectedDoubleValue) { + cpuMetrics.collectIdleCPU(statMetricCollectorMock) + } + } + + test("Method collectWaitCPU should return Double") { + val originalMetricName: String = "cpu_iowait" + val expectedDoubleValue: Double = 756859 + val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + + when(statMetricCollectorMock.getProcFileSource()) + .thenReturn(procFileSourceMock) + when( + statMetricCollectorMock.getMetricValue( + procFileSourceMock, + originalMetricName + ) + ) + .thenReturn(expectedDoubleValue) + + assertResult(expectedDoubleValue) { + cpuMetrics.collectWaitCPU(statMetricCollectorMock) + } + } +} From b1a821fabb71f4fca736b1196cf8a741c859ed7b Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Tue, 7 May 2024 14:10:28 -0300 Subject: [PATCH 11/12] feat: removed io.Source usage --- .../scala/io/github/dutrevis/CPUMetrics.scala | 29 +++++++++---------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/src/main/scala/io/github/dutrevis/CPUMetrics.scala b/src/main/scala/io/github/dutrevis/CPUMetrics.scala index 32a3040..98df0c3 100644 --- a/src/main/scala/io/github/dutrevis/CPUMetrics.scala +++ b/src/main/scala/io/github/dutrevis/CPUMetrics.scala @@ -2,7 +2,6 @@ package io.github.dutrevis import java.util.{Map => JMap} import scala.collection.JavaConverters._ -import scala.io.Source import com.codahale.metrics.{Gauge, Metric, MetricRegistry} import org.apache.spark.SparkContext @@ -21,8 +20,6 @@ import org.apache.spark.api.plugin.{ * spent performing different kinds of work, arranged in columns at the * following order: "cpu_user", "cpu_nice", "cpu_system", "cpu_idle", * "cpu_iowait", "cpu_irq" and "cpu_softirq".
- * @param metricCollector - * a StatMetricCollector instance
* @note * All of the numbers retrieved are aggregates since the system first booted. * @note @@ -31,9 +28,7 @@ import org.apache.spark.api.plugin.{ * Values for "cpu_steal", "cpu_guest" and "cpu_guest_nice", available at * spectific Linux versions, are not parsed from the file. */ -class CPUMetrics( - val metricCollector: StatMetricCollector = new StatMetricCollector() -) extends SparkPlugin { +class CPUMetrics extends SparkPlugin { /** Maps the collector methods to their respective metric names, that will be * displayed in the Dropwizard's metric system. @@ -93,9 +88,9 @@ class CPUMetrics( * Collected value is of type Double with precision of 2. */ def collectUserCPU(metricCollector: StatMetricCollector): Double = { - val procFileSource = metricCollector.getProcFileSource() + val procFileContent = metricCollector.getProcFileContent() val originalMetricName: String = "cpu_user" - metricCollector.getMetricValue(procFileSource, originalMetricName) + metricCollector.getMetricValue(procFileContent, originalMetricName) } /** Collects the aggregated CPU usage value for niced processes (there is, run @@ -107,9 +102,9 @@ class CPUMetrics( * Collected value is of type Double with precision of 2. */ def collectNiceCPU(metricCollector: StatMetricCollector): Double = { - val procFileSource = metricCollector.getProcFileSource() + val procFileContent = metricCollector.getProcFileContent() val originalMetricName: String = "cpu_nice" - metricCollector.getMetricValue(procFileSource, originalMetricName) + metricCollector.getMetricValue(procFileContent, originalMetricName) } /** Collects the aggregated CPU usage value processes executing in kernel mode @@ -121,9 +116,9 @@ class CPUMetrics( * Collected value is of type Double with precision of 2. */ def collectSystemCPU(metricCollector: StatMetricCollector): Double = { - val procFileSource = metricCollector.getProcFileSource() + val procFileContent = metricCollector.getProcFileContent() val originalMetricName: String = "cpu_system" - metricCollector.getMetricValue(procFileSource, originalMetricName) + metricCollector.getMetricValue(procFileContent, originalMetricName) } /** Collects the aggregated CPU usage value for when no processes are running, @@ -135,9 +130,9 @@ class CPUMetrics( * Collected value is of type Double with precision of 2. */ def collectIdleCPU(metricCollector: StatMetricCollector): Double = { - val procFileSource = metricCollector.getProcFileSource() + val procFileContent = metricCollector.getProcFileContent() val originalMetricName: String = "cpu_idle" - metricCollector.getMetricValue(procFileSource, originalMetricName) + metricCollector.getMetricValue(procFileContent, originalMetricName) } /** Collects the aggregated CPU usage value for when it's waiting for I/O to @@ -149,9 +144,9 @@ class CPUMetrics( * Collected value is of type Double with precision of 2. */ def collectWaitCPU(metricCollector: StatMetricCollector): Double = { - val procFileSource = metricCollector.getProcFileSource() + val procFileContent = metricCollector.getProcFileContent() val originalMetricName: String = "cpu_iowait" - metricCollector.getMetricValue(procFileSource, originalMetricName) + metricCollector.getMetricValue(procFileContent, originalMetricName) } /** Returns the plugin's driver-side component. The returned DriverPlugin @@ -182,6 +177,7 @@ class CPUMetrics( sc: SparkContext, myContext: PluginContext ): JMap[String, String] = { + val metricCollector = new StatMetricCollector for ( ( metricName: String, @@ -229,6 +225,7 @@ class CPUMetrics( myContext: PluginContext, extraConf: JMap[String, String] ): Unit = { + val metricCollector = new StatMetricCollector for ( ( metricName: String, From fd8daf8b19de4cfdf9a4ffa7f88a97d4b61aa13b Mon Sep 17 00:00:00 2001 From: Eduardo Trevisani <11429718+dutrevis@users.noreply.github.com> Date: Tue, 7 May 2024 14:10:46 -0300 Subject: [PATCH 12/12] chore: changed CPUMetrics tests --- src/test/scala/CPUMetricsTest.scala | 77 +++++++++++++++-------------- 1 file changed, 40 insertions(+), 37 deletions(-) diff --git a/src/test/scala/CPUMetricsTest.scala b/src/test/scala/CPUMetricsTest.scala index 0bfeab8..aabab6b 100644 --- a/src/test/scala/CPUMetricsTest.scala +++ b/src/test/scala/CPUMetricsTest.scala @@ -1,10 +1,11 @@ import io.github.dutrevis.{CPUMetrics, StatMetricCollector} -import org.scalatest.funsuite.AnyFunSuite -import org.mockito.{MockitoSugar, ArgumentMatchersSugar} + +import com.codahale.metrics.{Gauge, Metric, MetricRegistry} + import org.mockito.captor.ArgCaptor import org.mockito.integrations.scalatest.ResetMocksAfterEachTest -import com.codahale.metrics.{Gauge, Metric, MetricRegistry} -import scala.io.BufferedSource +import org.mockito.{MockitoSugar, ArgumentMatchersSugar} +import org.scalatest.funsuite.AnyFunSuite class CPUMetricsTest extends AnyFunSuite @@ -16,11 +17,13 @@ class CPUMetricsTest val gaugeMock = mock[Gauge[Double]] val metricMock = mock[Metric] val metricRegistryMock = mock[MetricRegistry] - val procFileSourceMock = mock[BufferedSource] val statMetricCollectorMock = mock[StatMetricCollector] + // Arrange common values + val procFileContentTest = new String + test("Method createGaugeMetric should return Gauge[Double]") { - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics val collectorMethod = (s: StatMetricCollector) => { 123456.toDouble } val returnedGaugeMetric = @@ -29,7 +32,7 @@ class CPUMetricsTest } test("Method registerMetric should call register") { - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics val metricName: String = "any_metric" when(metricRegistryMock.register(any[String], any[Metric])) @@ -41,42 +44,42 @@ class CPUMetricsTest test("Method collectUserCPU should call getMetricValue") { val originalMetricName: String = "cpu_user" - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) cpuMetrics.collectUserCPU(statMetricCollectorMock) verify(statMetricCollectorMock, times(1)) - .getMetricValue(procFileSourceMock, originalMetricName) + .getMetricValue(procFileContentTest, originalMetricName) } test("Method collectUserCPU should call getMetricValue with args") { - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics val originalMetricName: String = "cpu_user" - val procFileSourceCaptor = ArgCaptor[BufferedSource] + val procFileContentCaptor = ArgCaptor[String] val originalMetricNameCaptor = ArgCaptor[String] - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) cpuMetrics.collectUserCPU(statMetricCollectorMock) verify(statMetricCollectorMock).getMetricValue( - procFileSourceCaptor, + procFileContentCaptor, originalMetricNameCaptor ) - procFileSourceCaptor hasCaptured procFileSourceMock + procFileContentCaptor hasCaptured procFileContentTest originalMetricNameCaptor hasCaptured originalMetricName } test("Method collectUserCPU should return Double") { val originalMetricName: String = "cpu_user" val expectedDoubleValue: Double = 79242 - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) when( statMetricCollectorMock.getMetricValue( - procFileSourceMock, + procFileContentTest, originalMetricName ) ) @@ -90,13 +93,13 @@ class CPUMetricsTest test("Method collectNiceCPU should return Double") { val originalMetricName: String = "cpu_nice" val expectedDoubleValue: Double = 0 - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) when( statMetricCollectorMock.getMetricValue( - procFileSourceMock, + procFileContentTest, originalMetricName ) ) @@ -110,13 +113,13 @@ class CPUMetricsTest test("Method collectSystemCPU should return Double") { val originalMetricName: String = "cpu_system" val expectedDoubleValue: Double = 74306 - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) when( statMetricCollectorMock.getMetricValue( - procFileSourceMock, + procFileContentTest, originalMetricName ) ) @@ -130,13 +133,13 @@ class CPUMetricsTest test("Method collectIdleCPU should return Double") { val originalMetricName: String = "cpu_idle" val expectedDoubleValue: Double = 842486413 - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) when( statMetricCollectorMock.getMetricValue( - procFileSourceMock, + procFileContentTest, originalMetricName ) ) @@ -150,13 +153,13 @@ class CPUMetricsTest test("Method collectWaitCPU should return Double") { val originalMetricName: String = "cpu_iowait" val expectedDoubleValue: Double = 756859 - val cpuMetrics = new CPUMetrics(statMetricCollectorMock) + val cpuMetrics = new CPUMetrics - when(statMetricCollectorMock.getProcFileSource()) - .thenReturn(procFileSourceMock) + when(statMetricCollectorMock.getProcFileContent()) + .thenReturn(procFileContentTest) when( statMetricCollectorMock.getMetricValue( - procFileSourceMock, + procFileContentTest, originalMetricName ) )