Add jvm threads states metrics (#1075)

* add jvm threads states metrics * Update instrumentation/kamon-system-metrics/src/main/scala/kamon/instrumentation/system/jvm/JvmMetrics.scala Co-authored-by: Ivan Topolnjak <[email protected]> * rename timed_waited to timed-waited * remove trailing comma for scala 2.11 compatibility Co-authored-by: Ivan Topolnjak <[email protected]>
kamon-io · Nov 23, 2021 · 558e837 · 558e837
1 parent 49aef33
commit 558e837
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 3 deletions.
diff --git a/...ion/kamon-system-metrics/src/main/scala/kamon/instrumentation/system/jvm/JvmMetrics.scala b/...ion/kamon-system-metrics/src/main/scala/kamon/instrumentation/system/jvm/JvmMetrics.scala
@@ -17,10 +17,9 @@
 package kamon.instrumentation.system.jvm
 
 import java.lang.management.ManagementFactory
-
 import kamon.Kamon
 import kamon.instrumentation.system.jvm.JvmMetrics.MemoryUsageInstruments.{BufferPoolInstruments, MemoryRegionInstruments}
-import kamon.instrumentation.system.jvm.JvmMetricsCollector.{Collector, MemoryPool}
+import kamon.instrumentation.system.jvm.JvmMetricsCollector.{Collector, MemoryPool, ThreadState}
 import kamon.metric.{Gauge, Histogram, InstrumentGroup, MeasurementUnit}
 import kamon.tag.TagSet
 
@@ -108,6 +107,11 @@ object JvmMetrics {
     name = "jvm.threads.daemon",
     description = "Tracks the current number of daemon threads on the JVM"
   )
+
+  val ThreadsStates = Kamon.gauge(
+    name = "jvm.threads.states",
+    description = "Tracks the current number of threads on each possible state"
+  )
 
   val ClassesLoaded = Kamon.gauge(
     name = "jvm.class-loading.loaded",
@@ -208,9 +212,18 @@ object JvmMetrics {
   }
 
   class ThreadsInstruments extends InstrumentGroup(TagSet.Empty) {
+    private val _threadsStatesCache = mutable.Map.empty[ThreadState, Gauge]
     val total = register(ThreadsTotal)
     val peak = register(ThreadsPeak)
     val daemon = register(ThreadsDaemon)
+
+    def threadState(threadState: ThreadState): Gauge = {
+      _threadsStatesCache.getOrElseUpdate(threadState, {
+        val stateTag = TagSet.of("state", threadState.toString)
+
+        register(ThreadsStates, stateTag)
+      })
+    }
   }
 
   object MemoryUsageInstruments {

diff --git a/...-system-metrics/src/main/scala/kamon/instrumentation/system/jvm/JvmMetricsCollector.scala b/...-system-metrics/src/main/scala/kamon/instrumentation/system/jvm/JvmMetricsCollector.scala
@@ -27,7 +27,7 @@ import javax.management.{Notification, NotificationEmitter, NotificationListener
 import kamon.Kamon
 import kamon.instrumentation.system.jvm.JvmMetrics.{ClassLoadingInstruments, GarbageCollectionInstruments, MemoryUsageInstruments, ThreadsInstruments}
 import kamon.instrumentation.system.jvm.JvmMetricsCollector.MemoryPool.sanitize
-import kamon.instrumentation.system.jvm.JvmMetricsCollector.{Collector, MemoryPool}
+import kamon.instrumentation.system.jvm.JvmMetricsCollector.{Collector, MemoryPool, ThreadState}
 import kamon.module.{Module, ModuleFactory, ScheduledAction}
 import kamon.tag.TagSet
 
@@ -137,6 +137,15 @@ class JvmMetricsCollector(ec: ExecutionContext) extends ScheduledAction {
       threadsInstruments.total.update(threadsMxBen.getThreadCount())
       threadsInstruments.peak.update(threadsMxBen.getPeakThreadCount())
       threadsInstruments.daemon.update(threadsMxBen.getDaemonThreadCount())
+
+      threadsMxBen.getAllThreadIds.map(threadsMxBen.getThreadInfo(_, 0))
+        .groupBy(_.getThreadState)
+        .mapValues(_.length)
+        .foreach { 
+          case (state, count) =>  
+            val threadState = ThreadState.find(state.toString)
+            threadsInstruments.threadState(threadState).update(count) 
+        }
 
       val currentHeapUsage = ManagementFactory.getMemoryMXBean.getHeapMemoryUsage
       val freeHeap = Math.max(0L, currentHeapUsage.getMax -  currentHeapUsage.getUsed)
@@ -259,4 +268,27 @@ object JvmMetricsCollector {
     def sanitize(name: String): String =
       _invalidChars.replaceAllIn(name.toLowerCase, "-")
   }
+
+  sealed trait ThreadState
+  object ThreadState {
+    case object New extends ThreadState { override def toString: String = "new"}
+    case object Runnable extends ThreadState { override def toString: String = "runnable" }
+    case object Blocked extends ThreadState { override def toString: String = "blocked" }
+    case object Waiting extends ThreadState { override def toString: String = "waiting" }
+    case object TimedWaiting extends ThreadState { override def toString: String = "timed-waiting" }
+    case object Terminated extends ThreadState { override def toString: String = "terminated" }
+    case object Unknown extends ThreadState { override def toString: String = "unknown" }
+
+    def find(state: String): ThreadState =
+      _threadStateMapping.getOrElse(state, Unknown)
+
+    private val _threadStateMapping: Map[String, ThreadState] = Map(
+      "NEW"           -> New,
+      "RUNNABLE"      -> Runnable,
+      "BLOCKED"       -> Blocked,
+      "WAITING"       -> Waiting,
+      "TIMED_WAITING" -> TimedWaiting,
+      "TERMINATED"    -> Terminated
+    )
+  }
 }