[samza] branch master updated: SAMZA-2618: Add per-container metric for memory-utilization (#1459)

[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 f305149  SAMZA-2618: Add per-container metric for memory-utilization (#1459)
f305149 is described below

commit f305149d224f46300bf796770ccab798ca8d845d
Author: Ziting <90...@users.noreply.github.com>
AuthorDate: Wed Feb 3 15:55:47 2021 -0800

    SAMZA-2618: Add per-container metric for memory-utilization (#1459)
    
    Feature: Add a metric to measure the fraction of physical memory used out of the total memory of the container. During autosizing, the container-count and size can change as traffic changes, so only monitoring the phsical-memory-mb does not provide enough information for alerting. We need a fraction or % metric to allow for easy monitoring and alerting
    
    Changes: Add a metric to measure the fraction of physical memory used out of the total memory of the container.
---
 docs/learn/documentation/versioned/operations/monitoring.md        | 1 +
 .../src/main/scala/org/apache/samza/container/SamzaContainer.scala | 7 ++++++-
 .../scala/org/apache/samza/container/SamzaContainerMetrics.scala   | 3 ++-
 3 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/docs/learn/documentation/versioned/operations/monitoring.md b/docs/learn/documentation/versioned/operations/monitoring.md
index c11fbe0..de2497c 100644
--- a/docs/learn/documentation/versioned/operations/monitoring.md
+++ b/docs/learn/documentation/versioned/operations/monitoring.md
@@ -393,6 +393,7 @@ All \<system\>, \<stream\>, \<partition\>, \<store-name\>, \<topic\>, are popula
 | | disk-quota-bytes | Disk memory usage quota for key-value stores (in bytes). |
 | | 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. |
 | | <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 28c3135..3a4ba17 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
@@ -617,13 +617,18 @@ object SamzaContainer extends Logging {
       taskConfig,
       clock)
 
+    val containerMemoryMb : Int = new ClusterManagerConfig(config).getContainerMemoryMb
+
     val memoryStatisticsMonitor : SystemStatisticsMonitor = new StatisticsMonitorImpl()
     memoryStatisticsMonitor.registerListener(new SystemStatisticsMonitor.Listener {
       override def onUpdate(sample: SystemMemoryStatistics): Unit = {
         val physicalMemoryBytes : Long = sample.getPhysicalMemoryBytes
-        val physicalMemoryMb : Double = physicalMemoryBytes / (1024.0 * 1024.0)
+        val physicalMemoryMb : Float = physicalMemoryBytes / (1024.0F * 1024.0F)
+        val memoryUtilization : Float = physicalMemoryMb.toFloat / containerMemoryMb
         logger.debug("Container physical memory utilization (mb): " + physicalMemoryMb)
+        logger.debug("Container physical memory utilization: " + memoryUtilization)
         samzaContainerMetrics.physicalMemoryMb.set(physicalMemoryMb)
+        samzaContainerMetrics.physicalMemoryUtilization.set(memoryUtilization);
       }
     })
 
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 73dcd20..46de166 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
@@ -46,7 +46,8 @@ class SamzaContainerMetrics(
   val diskUsageBytes = newGauge("disk-usage-bytes", 0L)
   val diskQuotaBytes = newGauge("disk-quota-bytes", Long.MaxValue)
   val executorWorkFactor = newGauge("executor-work-factor", 1.0)
-  val physicalMemoryMb = newGauge[Double]("physical-memory-mb", 0.0F)
+  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 taskStoreRestorationMetrics: util.Map[TaskName, Gauge[Long]] = new util.HashMap[TaskName, Gauge[Long]]()