Merge branch 'wip/improve-metric-recorders-infrastructure'

Conflicts:
	kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala

1 files changed, 77 insertions, 118 deletions

diff --git a/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala b/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala
index 534edd57..e150d967 100644
--- a/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala
+++ b/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala
@@ -17,40 +17,33 @@ package kamon.jdbc.instrumentation
 
 import java.sql.{ DriverManager, SQLException }
 
-import akka.actor.ActorSystem
-import akka.testkit.{ TestKitBase, TestProbe }
 import com.typesafe.config.ConfigFactory
-import kamon.Kamon
 import kamon.jdbc.{ Jdbc, JdbcNameGenerator, SqlErrorProcessor, SlowQueryProcessor }
-import kamon.jdbc.metric.StatementsMetrics
-import kamon.jdbc.metric.StatementsMetrics.StatementsMetricsSnapshot
-import kamon.metric.{ TraceMetrics, Metrics }
-import kamon.metric.Subscriptions.TickMetricSnapshot
-import kamon.metric.TraceMetrics.TraceMetricsSnapshot
-import kamon.trace.{ SegmentCategory, SegmentMetricIdentity, TraceRecorder }
-import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike }
-
-import scala.concurrent.duration._
-
-class StatementInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with BeforeAndAfterAll {
-
-  implicit lazy val system: ActorSystem = ActorSystem("jdbc-spec", ConfigFactory.parseString(
-    """
-      |kamon {
-      |   jdbc {
-      |     slow-query-threshold = 100 milliseconds
-      |
-      |     # Fully qualified name of the implementation of kamon.jdbc.SlowQueryProcessor.
-      |     slow-query-processor = kamon.jdbc.instrumentation.NoOpSlowQueryProcessor
-      |
-      |     # Fully qualified name of the implementation of kamon.jdbc.SqlErrorProcessor.
-      |     sql-error-processor = kamon.jdbc.instrumentation.NoOpSqlErrorProcessor
-      |
-      |     # Fully qualified name of the implementation of kamon.jdbc.JdbcNameGenerator
-      |     name-generator = kamon.jdbc.instrumentation.NoOpJdbcNameGenerator
-      |   }
-      |}
-    """.stripMargin))
+import kamon.metric.TraceMetricsSpec
+import kamon.testkit.BaseKamonSpec
+import kamon.trace.{ SegmentCategory, TraceContext }
+
+class StatementInstrumentationSpec extends BaseKamonSpec("jdbc-spec") {
+  import TraceMetricsSpec.SegmentSyntax
+
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |kamon {
+        |   jdbc {
+        |     slow-query-threshold = 100 milliseconds
+        |
+        |     # Fully qualified name of the implementation of kamon.jdbc.SlowQueryProcessor.
+        |     slow-query-processor = kamon.jdbc.instrumentation.NoOpSlowQueryProcessor
+        |
+        |     # Fully qualified name of the implementation of kamon.jdbc.SqlErrorProcessor.
+        |     sql-error-processor = kamon.jdbc.instrumentation.NoOpSqlErrorProcessor
+        |
+        |     # Fully qualified name of the implementation of kamon.jdbc.JdbcNameGenerator
+        |     name-generator = kamon.jdbc.instrumentation.NoOpJdbcNameGenerator
+        |   }
+        |}
+      """.stripMargin)
 
   val connection = DriverManager.getConnection("jdbc:h2:mem:jdbc-spec", "SA", "")
 
@@ -67,117 +60,105 @@ class StatementInstrumentationSpec extends TestKitBase with WordSpecLike with Ma
   }
 
   "the StatementInstrumentation" should {
-    "record the execution time of INSERT operation" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-insert") {
-
-        val metricsListener = subscribeToMetrics()
-
+    "record the execution time of INSERT operation" in {
+      TraceContext.withContext(newContext("jdbc-trace-insert")) {
         for (id ← 1 to 100) {
           val insert = s"INSERT INTO Address (Nr, Name) VALUES($id, 'foo')"
           val insertStatement = connection.prepareStatement(insert)
           insertStatement.execute()
         }
 
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.writes.numberOfMeasurements should be(100)
-
-        TraceRecorder.finish()
+        TraceContext.currentContext.finish()
       }
 
-      val snapshot = takeSnapshotOf("jdbc-trace-insert")
-      snapshot.elapsedTime.numberOfMeasurements should be(1)
-      snapshot.segments.size should be(1)
-      snapshot.segments(SegmentMetricIdentity("Jdbc[Insert]", SegmentCategory.Database, Jdbc.SegmentLibraryName)).numberOfMeasurements should be(100)
-    }
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.histogram("writes").get.numberOfMeasurements should be(100)
 
-    "record the execution time of SELECT operation" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-select") {
-
-        val metricsListener = subscribeToMetrics()
+      val traceSnapshot = takeSnapshotOf("jdbc-trace-insert", "trace")
+      traceSnapshot.histogram("elapsed-time").get.numberOfMeasurements should be(1)
+      traceSnapshot.segments.size should be(1)
+      traceSnapshot.segment("Jdbc[Insert]", SegmentCategory.Database, Jdbc.SegmentLibraryName).numberOfMeasurements should be(100)
+    }
 
+    "record the execution time of SELECT operation" in {
+      TraceContext.withContext(newContext("jdbc-trace-select")) {
         for (id ← 1 to 100) {
           val select = s"SELECT * FROM Address where Nr = $id"
           val selectStatement = connection.createStatement()
           selectStatement.execute(select)
         }
 
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.reads.numberOfMeasurements should be(100)
-
-        TraceRecorder.finish()
+        TraceContext.currentContext.finish()
       }
 
-      val snapshot = takeSnapshotOf("jdbc-trace-select")
-      snapshot.elapsedTime.numberOfMeasurements should be(1)
-      snapshot.segments.size should be(1)
-      snapshot.segments(SegmentMetricIdentity("Jdbc[Select]", SegmentCategory.Database, Jdbc.SegmentLibraryName)).numberOfMeasurements should be(100)
-    }
-
-    "record the execution time of UPDATE operation" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-update") {
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.histogram("reads").get.numberOfMeasurements should be(100)
 
-        val metricsListener = subscribeToMetrics()
+      val traceSnapshot = takeSnapshotOf("jdbc-trace-select", "trace")
+      traceSnapshot.histogram("elapsed-time").get.numberOfMeasurements should be(1)
+      traceSnapshot.segments.size should be(1)
+      traceSnapshot.segment("Jdbc[Select]", SegmentCategory.Database, Jdbc.SegmentLibraryName).numberOfMeasurements should be(100)
+    }
 
+    "record the execution time of UPDATE operation" in {
+      TraceContext.withContext(newContext("jdbc-trace-update")) {
         for (id ← 1 to 100) {
           val update = s"UPDATE Address SET Name = 'bar$id' where Nr = $id"
           val updateStatement = connection.prepareStatement(update)
           updateStatement.execute()
         }
 
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.writes.numberOfMeasurements should be(100)
-        TraceRecorder.finish()
+        TraceContext.currentContext.finish()
       }
 
-      val snapshot = takeSnapshotOf("jdbc-trace-update")
-      snapshot.elapsedTime.numberOfMeasurements should be(1)
-      snapshot.segments.size should be(1)
-      snapshot.segments(SegmentMetricIdentity("Jdbc[Update]", SegmentCategory.Database, Jdbc.SegmentLibraryName)).numberOfMeasurements should be(100)
-    }
-
-    "record the execution time of DELETE operation" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-delete") {
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.histogram("writes").get.numberOfMeasurements should be(100)
 
-        val metricsListener = subscribeToMetrics()
+      val traceSnapshot = takeSnapshotOf("jdbc-trace-update", "trace")
+      traceSnapshot.histogram("elapsed-time").get.numberOfMeasurements should be(1)
+      traceSnapshot.segments.size should be(1)
+      traceSnapshot.segment("Jdbc[Update]", SegmentCategory.Database, Jdbc.SegmentLibraryName).numberOfMeasurements should be(100)
+    }
 
+    "record the execution time of DELETE operation" in {
+      TraceContext.withContext(newContext("jdbc-trace-delete")) {
         for (id ← 1 to 100) {
           val delete = s"DELETE FROM Address where Nr = $id"
           val deleteStatement = connection.createStatement()
           deleteStatement.execute(delete)
         }
 
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.writes.numberOfMeasurements should be(100)
-        TraceRecorder.finish()
+        TraceContext.currentContext.finish()
       }
 
-      val snapshot = takeSnapshotOf("jdbc-trace-delete")
-      snapshot.elapsedTime.numberOfMeasurements should be(1)
-      snapshot.segments.size should be(1)
-      snapshot.segments(SegmentMetricIdentity("Jdbc[Delete]", SegmentCategory.Database, Jdbc.SegmentLibraryName)).numberOfMeasurements should be(100)
-    }
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.histogram("writes").get.numberOfMeasurements should be(100)
 
-    "record the execution time of SLOW QUERIES based on the kamon.jdbc.slow-query-threshold" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-slow") {
+      val traceSnapshot = takeSnapshotOf("jdbc-trace-delete", "trace")
+      traceSnapshot.histogram("elapsed-time").get.numberOfMeasurements should be(1)
+      traceSnapshot.segments.size should be(1)
+      traceSnapshot.segment("Jdbc[Delete]", SegmentCategory.Database, Jdbc.SegmentLibraryName).numberOfMeasurements should be(100)
 
-        val metricsListener = subscribeToMetrics()
+    }
 
+    "record the execution time of SLOW QUERIES based on the kamon.jdbc.slow-query-threshold" in {
+      TraceContext.withContext(newContext("jdbc-trace-slow")) {
         for (id ← 1 to 2) {
           val select = s"SELECT * FROM Address; CALL SLEEP(100)"
           val selectStatement = connection.createStatement()
           selectStatement.execute(select)
         }
 
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.slows.count should be(2)
+        TraceContext.currentContext.finish()
       }
-    }
 
-    "count all SQL ERRORS" in new StatementsMetricsListenerFixture {
-      TraceRecorder.withNewTraceContext("jdbc-trace-errors") {
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.counter("slows").get.count should be(2)
 
-        val metricsListener = subscribeToMetrics()
+    }
 
+    "count all SQL ERRORS" in {
+      TraceContext.withContext(newContext("jdbc-trace-errors")) {
         for (_ ← 1 to 10) {
           intercept[SQLException] {
             val error = "SELECT * FROM NO_EXISTENT_TABLE"
@@ -185,35 +166,13 @@ class StatementInstrumentationSpec extends TestKitBase with WordSpecLike with Ma
             errorStatement.execute(error)
           }
         }
-        val StatementMetrics = expectStatementsMetrics(metricsListener, 3 seconds)
-        StatementMetrics.errors.count should be(10)
-      }
-    }
-  }
 
-  trait StatementsMetricsListenerFixture {
-    def subscribeToMetrics(): TestProbe = {
-      val metricsListener = TestProbe()
-      Kamon(Metrics).subscribe(StatementsMetrics, "*", metricsListener.ref, permanently = true)
-      // Wait for one empty snapshot before proceeding to the test.
-      metricsListener.expectMsgType[TickMetricSnapshot]
-      metricsListener
-    }
-  }
+        TraceContext.currentContext.finish()
+      }
 
-  def expectStatementsMetrics(listener: TestProbe, waitTime: FiniteDuration): StatementsMetricsSnapshot = {
-    val tickSnapshot = within(waitTime) {
-      listener.expectMsgType[TickMetricSnapshot]
+      val jdbcSnapshot = takeSnapshotOf("jdbc-statements", "jdbc-statements")
+      jdbcSnapshot.counter("errors").get.count should be(10)
     }
-    val statementsMetricsOption = tickSnapshot.metrics.get(StatementsMetrics(StatementInstrumentation.Statements))
-    statementsMetricsOption should not be empty
-    statementsMetricsOption.get.asInstanceOf[StatementsMetricsSnapshot]
-  }
-
-  def takeSnapshotOf(traceName: String): TraceMetricsSnapshot = {
-    val recorder = Kamon(Metrics)(system).register(TraceMetrics(traceName), TraceMetrics.Factory)
-    val collectionContext = Kamon(Metrics)(system).buildDefaultCollectionContext
-    recorder.get.collect(collectionContext)
   }
 }