[samza] branch master updated: Add a metric to track the number of active threads of a container's thread pool (#1476)

[bh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bharathkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git


The following commit(s) were added to refs/heads/master by this push:
     new acd4493  Add a metric to track the number of active threads of a container's thread pool (#1476)
acd4493 is described below

commit acd4493a62051b020f928bd288a3623a8c979f92
Author: Ziting <90...@users.noreply.github.com>
AuthorDate: Fri Mar 19 16:19:19 2021 -0700

    Add a metric to track the number of active threads of a container's thread pool (#1476)
    
    Feature: Add a metric to track the number of active threads of a container's thread pool. This will be used to calculate the thread pool utilization of a job, which will be used as a signal for the autosizing controller to scale up or down the job thread pool size and number of containers.
    
    Changes:
     - Add a metric to track the number of active threads of a container's thread pool.
     - Update the existing container thread pool size metric to reflect the pool size from the thread pool instead of emitting the static job.container.thread.pool.size
---
 .../documentation/versioned/operations/monitoring.md      |  2 ++
 .../scala/org/apache/samza/container/SamzaContainer.scala | 15 +++++++++------
 .../apache/samza/container/SamzaContainerMetrics.scala    |  1 +
 3 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/docs/learn/documentation/versioned/operations/monitoring.md b/docs/learn/documentation/versioned/operations/monitoring.md
index de2497c..0671047 100644
--- a/docs/learn/documentation/versioned/operations/monitoring.md
+++ b/docs/learn/documentation/versioned/operations/monitoring.md
@@ -394,6 +394,8 @@ All \<system\>, \<stream\>, \<partition\>, \<store-name\>, \<topic\>, are popula
 | | executor-work-factor | The work factor of the run loop. A work factor of 1 indicates full throughput, while a work factor of less than 1 will introduce delays into the execution to approximate the requested work factor. The work factor is set by the disk space monitor in accordance with the disk quota policy. Given the latest percentage of available disk quota, this policy returns the work factor that should be applied. |
 | | physical-memory-mb | The physical memory used by the Samza container process (native + on heap) (in MBs). |
 | | physical-memory-utilization | The ratio between the physical memory used by the Samza container process (native + on heap) and the total physical memory of the Samza container. |
+| | container-thread-pool-size | The current size of a Samza container's thread pool. It may or may not be the same as job.container.thread.pool.size, depending on the implementation. |
+| | container-active-threads | The approximate actively used threads in a Samza container's thread pool. |
 | | <TaskName\>-<StoreName\>-restore-time | Time taken to restore task stores (per task store). |
 
 
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 3a4ba17..2e87f48 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -26,7 +26,7 @@ import java.nio.file.Path
 import java.time.Duration
 import java.util
 import java.util.{Base64, Optional}
-import java.util.concurrent.{CountDownLatch, ExecutorService, Executors, ScheduledExecutorService, TimeUnit}
+import java.util.concurrent.{CountDownLatch, ExecutorService, Executors, ScheduledExecutorService, ThreadPoolExecutor, TimeUnit}
 
 import com.google.common.annotations.VisibleForTesting
 import com.google.common.util.concurrent.ThreadFactoryBuilder
@@ -476,7 +476,6 @@ object SamzaContainer extends Logging {
 
     val threadPoolSize = jobConfig.getThreadPoolSize
     info("Got thread pool size: " + threadPoolSize)
-    samzaContainerMetrics.containerThreadPoolSize.set(threadPoolSize)
 
     val taskThreadPool = if (threadPoolSize > 0) {
       Executors.newFixedThreadPool(threadPoolSize,
@@ -619,16 +618,20 @@ object SamzaContainer extends Logging {
 
     val containerMemoryMb : Int = new ClusterManagerConfig(config).getContainerMemoryMb
 
-    val memoryStatisticsMonitor : SystemStatisticsMonitor = new StatisticsMonitorImpl()
-    memoryStatisticsMonitor.registerListener(new SystemStatisticsMonitor.Listener {
+    val hostStatisticsMonitor : SystemStatisticsMonitor = new StatisticsMonitorImpl()
+    hostStatisticsMonitor.registerListener(new SystemStatisticsMonitor.Listener {
       override def onUpdate(sample: SystemMemoryStatistics): Unit = {
         val physicalMemoryBytes : Long = sample.getPhysicalMemoryBytes
         val physicalMemoryMb : Float = physicalMemoryBytes / (1024.0F * 1024.0F)
         val memoryUtilization : Float = physicalMemoryMb.toFloat / containerMemoryMb
+        val containerThreadPoolSize : Long = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getPoolSize
+        val containerActiveThreads : Long = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getActiveCount
         logger.debug("Container physical memory utilization (mb): " + physicalMemoryMb)
         logger.debug("Container physical memory utilization: " + memoryUtilization)
         samzaContainerMetrics.physicalMemoryMb.set(physicalMemoryMb)
-        samzaContainerMetrics.physicalMemoryUtilization.set(memoryUtilization);
+        samzaContainerMetrics.physicalMemoryUtilization.set(memoryUtilization)
+        samzaContainerMetrics.containerThreadPoolSize.set(containerThreadPoolSize)
+        samzaContainerMetrics.containerActiveThreads.set(containerActiveThreads)
       }
     })
 
@@ -674,7 +677,7 @@ object SamzaContainer extends Logging {
       reporters = reporters,
       jvm = jvm,
       diskSpaceMonitor = diskSpaceMonitor,
-      hostStatisticsMonitor = memoryStatisticsMonitor,
+      hostStatisticsMonitor = hostStatisticsMonitor,
       taskThreadPool = taskThreadPool,
       timerExecutor = timerExecutor,
       containerContext = containerContext,
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
index 46de166..33d288f 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
@@ -49,6 +49,7 @@ class SamzaContainerMetrics(
   val physicalMemoryMb = newGauge("physical-memory-mb", 0.0F)
   val physicalMemoryUtilization = newGauge("physical-memory-utilization", 0.0F)
   val containerThreadPoolSize = newGauge("container-thread-pool-size", 0L)
+  val containerActiveThreads = newGauge("container-active-threads", 0L)
 
   val taskStoreRestorationMetrics: util.Map[TaskName, Gauge[Long]] = new util.HashMap[TaskName, Gauge[Long]]()