+ system-metrics: introduce Context Switches(only for Linux) metrics and closes #66

5 files changed, 192 insertions, 3 deletions

diff --git a/kamon-system-metrics/src/main/resources/reference.conf b/kamon-system-metrics/src/main/resources/reference.conf
index e5315223..fbdb3b89 100644
--- a/kamon-system-metrics/src/main/resources/reference.conf
+++ b/kamon-system-metrics/src/main/resources/reference.conf
@@ -51,6 +51,12 @@ kamon {
           swap-used = ${kamon.metrics.precision.default-histogram-precision}
           swap-free = ${kamon.metrics.precision.default-histogram-precision}
         }
+
+        context-switches {
+          per-process-voluntary = ${kamon.metrics.precision.default-histogram-precision}
+          per-process-non-voluntary = ${kamon.metrics.precision.default-histogram-precision}
+          global = ${kamon.metrics.precision.default-histogram-precision}
+        }
       }
 
       jvm {
diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/ContextSwitchesMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/ContextSwitchesMetrics.scala
new file mode 100644
index 00000000..a38b114c
--- /dev/null
+++ b/kamon-system-metrics/src/main/scala/kamon/metrics/ContextSwitchesMetrics.scala
@@ -0,0 +1,78 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2014 the kamon project <http://kamon.io/>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ * =========================================================================================
+ */
+
+package kamon.metrics
+
+import akka.actor.ActorSystem
+import com.typesafe.config.Config
+import kamon.metric._
+import kamon.metric.instrument.Histogram
+
+case class ContextSwitchesMetrics(name: String) extends MetricGroupIdentity {
+  val category = ContextSwitchesMetrics
+}
+
+object ContextSwitchesMetrics extends MetricGroupCategory {
+  val name = "context-switches"
+
+  case object PerProcessVoluntary extends MetricIdentity { val name = "per-process-voluntary" }
+  case object PerProcessNonVoluntary extends MetricIdentity { val name = "per-process-non-voluntary" }
+  case object Global extends MetricIdentity { val name = "global" }
+
+  case class ContextSwitchesMetricsRecorder(perProcessVoluntary: Histogram, perProcessNonVoluntary: Histogram, global: Histogram)
+      extends MetricGroupRecorder {
+
+    def collect(context: CollectionContext): MetricGroupSnapshot = {
+      ContextSwitchesMetricsSnapshot(perProcessVoluntary.collect(context), perProcessNonVoluntary.collect(context), global.collect(context))
+    }
+
+    def cleanup: Unit = {}
+  }
+
+  case class ContextSwitchesMetricsSnapshot(perProcessVoluntary: Histogram.Snapshot, perProcessNonVoluntary: Histogram.Snapshot, global: Histogram.Snapshot)
+      extends MetricGroupSnapshot {
+
+    type GroupSnapshotType = ContextSwitchesMetricsSnapshot
+
+    def merge(that: ContextSwitchesMetricsSnapshot, context: CollectionContext): GroupSnapshotType = {
+      ContextSwitchesMetricsSnapshot(perProcessVoluntary.merge(that.perProcessVoluntary, context), perProcessVoluntary.merge(that.perProcessVoluntary, context), global.merge(that.global, context))
+    }
+
+    lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map(
+      PerProcessVoluntary -> perProcessVoluntary,
+      PerProcessNonVoluntary -> perProcessNonVoluntary,
+      Global -> global)
+  }
+
+  val Factory = new MetricGroupFactory {
+
+    type GroupRecorder = ContextSwitchesMetricsRecorder
+
+    def create(config: Config, system: ActorSystem): GroupRecorder = {
+      val settings = config.getConfig("precision.system.context-switches")
+
+      val perProcessVoluntary = settings.getConfig("per-process-voluntary")
+      val perProcessNonVoluntary = settings.getConfig("per-process-non-voluntary")
+      val global = settings.getConfig("global")
+
+      new ContextSwitchesMetricsRecorder(
+        Histogram.fromConfig(perProcessVoluntary),
+        Histogram.fromConfig(perProcessNonVoluntary),
+        Histogram.fromConfig(global))
+    }
+  }
+}
+
diff --git a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala
index 29048915..62ffdb33 100644
--- a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala
+++ b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala
@@ -54,6 +54,7 @@ object SystemMetricsExtension {
   val Network = "network"
   val Memory = "memory"
   val Heap = "heap"
+  val ContextSwitches = "context-switches"
 
   def toKB(value: Long): Long = (value / 1024)
   def toMB(value: Long): Long = (value / 1024 / 1024)
diff --git a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala
index 725f634d..f41a76d5 100644
--- a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala
+++ b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala
@@ -15,20 +15,24 @@
  */
 package kamon.system
 
-import akka.actor.{ Actor, Props }
+import java.io.IOException
+
+import akka.actor.{ ActorLogging, Actor, Props }
 import kamon.Kamon
 import kamon.metric.Metrics
 import kamon.metrics.CPUMetrics.CPUMetricRecorder
+import kamon.metrics.ContextSwitchesMetrics.ContextSwitchesMetricsRecorder
 import kamon.metrics.MemoryMetrics.MemoryMetricRecorder
 import kamon.metrics.NetworkMetrics.NetworkMetricRecorder
 import kamon.metrics.ProcessCPUMetrics.ProcessCPUMetricsRecorder
-import kamon.metrics.{ CPUMetrics, MemoryMetrics, NetworkMetrics, ProcessCPUMetrics }
+import kamon.metrics._
 import kamon.system.sigar.SigarHolder
 import org.hyperic.sigar.{ Mem, NetInterfaceStat, SigarProxy }
 
 import scala.concurrent.duration.FiniteDuration
+import scala.io.Source
 
-class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with SigarExtensionProvider {
+class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with ActorLogging with SigarExtensionProvider {
   import kamon.system.SystemMetricsCollector._
   import kamon.system.SystemMetricsExtension._
 
@@ -40,6 +44,7 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with
   val processCpuRecorder = systemMetricsExtension.register(ProcessCPUMetrics(ProcessCPU), ProcessCPUMetrics.Factory)
   val memoryRecorder = systemMetricsExtension.register(MemoryMetrics(Memory), MemoryMetrics.Factory)
   val networkRecorder = systemMetricsExtension.register(NetworkMetrics(Network), NetworkMetrics.Factory)
+  val contextSwitchesRecorder = systemMetricsExtension.register(ContextSwitchesMetrics(ContextSwitches), ContextSwitchesMetrics.Factory)
 
   def receive: Receive = {
     case Collect ⇒ collectMetrics()
@@ -52,6 +57,9 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with
     processCpuRecorder.map(recordProcessCpu)
     memoryRecorder.map(recordMemory)
     networkRecorder.map(recordNetwork)
+
+    if (OsUtils.isLinux)
+      contextSwitchesRecorder.map(recordContextSwitches)
   }
 
   private def recordCpu(cpur: CPUMetricRecorder) = {
@@ -100,11 +108,62 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with
       }
     }
   }
+
+  private def recordContextSwitches(ctxt: ContextSwitchesMetricsRecorder) = {
+    def contextSwitchesByProcess(pid: Long): (Long, Long) = {
+      val filename = s"/proc/$pid/status"
+      var voluntaryContextSwitches = 0L
+      var nonVoluntaryContextSwitches = 0L
+
+      try {
+        for (line ← Source.fromFile(filename).getLines()) {
+          if (line.startsWith("voluntary_ctxt_switches")) {
+            voluntaryContextSwitches = line.substring(line.indexOf(":") + 1).trim.toLong
+          }
+          if (line.startsWith("nonvoluntary_ctxt_switches")) {
+            nonVoluntaryContextSwitches = line.substring(line.indexOf(":") + 1).trim.toLong
+          }
+        }
+      } catch {
+        case ex: IOException ⇒ {
+          log.error("Error trying to read [{}]", filename)
+        }
+      }
+      (voluntaryContextSwitches, nonVoluntaryContextSwitches)
+    }
+
+    def contextSwitches: Long = {
+      val filename = "/proc/stat"
+      var contextSwitches = 0L
+
+      try {
+        for (line ← Source.fromFile(filename).getLines()) {
+          if (line.startsWith("ctxt")) {
+            contextSwitches = line.substring(line.indexOf(" ") + 1).toLong
+          }
+        }
+      } catch {
+        case ex: IOException ⇒ {
+          log.error("Error trying to read [{}]", filename)
+        }
+      }
+      contextSwitches
+    }
+
+    val (perProcessVoluntary, perProcessNonVoluntary) = contextSwitchesByProcess(pid)
+    ctxt.perProcessVoluntary.record(perProcessVoluntary)
+    ctxt.perProcessNonVoluntary.record(perProcessNonVoluntary)
+    ctxt.global.record(contextSwitches)
+  }
 }
 
 object SystemMetricsCollector {
   case object Collect
 
+  object OsUtils {
+    def isLinux: Boolean = System.getProperty("os.name").indexOf("Linux") != -1;
+  }
+
   def props(collectInterval: FiniteDuration): Props = Props[SystemMetricsCollector](new SystemMetricsCollector(collectInterval))
 }
 
diff --git a/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala b/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala
index 8c340b1c..b475a416 100644
--- a/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala
+++ b/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala
@@ -21,6 +21,7 @@ import com.typesafe.config.ConfigFactory
 import kamon.Kamon
 import kamon.metric.Subscriptions.TickMetricSnapshot
 import kamon.metrics.CPUMetrics.CPUMetricSnapshot
+import kamon.metrics.ContextSwitchesMetrics.ContextSwitchesMetricsSnapshot
 import kamon.metrics.GCMetrics.GCMetricSnapshot
 import kamon.metrics.HeapMetrics.HeapMetricSnapshot
 import kamon.metrics.MemoryMetrics.MemoryMetricSnapshot
@@ -100,6 +101,20 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       |          significant-value-digits = 2
       |        }
       |     }
+      |     context-switches {
+      |        per-process-voluntary {
+      |          highest-trackable-value = 3600000000000
+      |          significant-value-digits = 2
+      |        }
+      |        per-process-non-voluntary {
+      |          highest-trackable-value = 3600000000000
+      |          significant-value-digits = 2
+      |        }
+      |        global {
+      |          highest-trackable-value = 3600000000000
+      |          significant-value-digits = 2
+      |        }
+      |     }
       |     network {
       |        rx-bytes {
       |            highest-trackable-value = 3600000000000
@@ -214,6 +229,17 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
     }
   }
 
+  "the Kamon ContextSwitches Metrics" should {
+    "record Context Switches Global, Voluntary and Non Voluntary metrics" in new ContextSwitchesMetricsListenerFixture {
+      val metricsListener = subscribeToMetrics()
+
+      val ContextSwitchesMetrics = expectContextSwitchesMetrics(metricsListener, 3 seconds)
+      ContextSwitchesMetrics.perProcessVoluntary.max should be > 0L
+      ContextSwitchesMetrics.perProcessNonVoluntary.max should be > 0L
+      ContextSwitchesMetrics.global.max should be > 0L
+    }
+  }
+
   def expectCPUMetrics(listener: TestProbe, waitTime: FiniteDuration): CPUMetricSnapshot = {
     val tickSnapshot = within(waitTime) {
       listener.expectMsgType[TickMetricSnapshot]
@@ -328,4 +354,23 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       metricsListener
     }
   }
+
+  def expectContextSwitchesMetrics(listener: TestProbe, waitTime: FiniteDuration): ContextSwitchesMetricsSnapshot = {
+    val tickSnapshot = within(waitTime) {
+      listener.expectMsgType[TickMetricSnapshot]
+    }
+    val contextSwitchesMetricsOption = tickSnapshot.metrics.get(ContextSwitchesMetrics(SystemMetricsExtension.ContextSwitches))
+    contextSwitchesMetricsOption should not be empty
+    contextSwitchesMetricsOption.get.asInstanceOf[ContextSwitchesMetricsSnapshot]
+  }
+
+  trait ContextSwitchesMetricsListenerFixture {
+    def subscribeToMetrics(): TestProbe = {
+      val metricsListener = TestProbe()
+      Kamon(Metrics).subscribe(ContextSwitchesMetrics, "*", metricsListener.ref, permanently = true)
+      // Wait for one empty snapshot before proceeding to the test.
+      metricsListener.expectMsgType[TickMetricSnapshot]
+      metricsListener
+    }
+  }
 }