1 files changed, 153 insertions, 36 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 ec8ca4a1..ee68234d 100644
--- a/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala
+++ b/kamon-jdbc/src/test/scala/kamon/jdbc/instrumentation/StatementInstrumentationSpec.scala
@@ -15,53 +15,170 @@
 
 package kamon.jdbc.instrumentation
 
-import java.sql.{SQLException, DriverManager}
+import java.sql.{DriverManager, SQLException}
 
 import akka.actor.ActorSystem
-import akka.testkit.TestKit
+import akka.testkit.{TestKitBase, TestProbe}
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.jdbc.SlowQueryProcessor
+import kamon.jdbc.metric.StatementsMetrics
+import kamon.jdbc.metric.StatementsMetrics.StatementsMetricsSnapshot
+import kamon.metric.Metrics
+import kamon.metric.Subscriptions.TickMetricSnapshot
 import kamon.trace.TraceRecorder
-import org.scalatest.{Matchers, WordSpecLike}
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+import scala.concurrent.duration._
 
-class StatementInstrumentationSpec extends TestKit(ActorSystem("jdbc-spec")) with WordSpecLike with Matchers {
+class StatementInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with BeforeAndAfterAll {
 
-  val connection = DriverManager.getConnection("jdbc:h2:mem:test","SA", "")
+  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.NOPSlowQueryProcessor
+      |   }
+      |}
+    """.stripMargin))
+
+  val connection = DriverManager.getConnection("jdbc:h2:mem:jdbc-spec","SA", "")
+
+  override protected def beforeAll(): Unit = {
+    connection should not be null
+
+    val create = "CREATE TABLE Address (Nr INTEGER, Name VARCHAR(128));"
+    val createStatement = connection.createStatement()
+    createStatement.executeUpdate(create)
+
+    val sleep = "CREATE ALIAS SLEEP FOR \"java.lang.Thread.sleep(long)\""
+    val sleepStatement = connection.createStatement()
+    sleepStatement.executeUpdate(sleep)
+  }
 
   "the StatementInstrumentation" should {
-    "bblabals" in {
-      TraceRecorder.withNewTraceContext("jdbc-trace") {
-        connection should not be null
-
-        val create = "CREATE TABLE Address (Nr INTEGER, Name VARCHAR(128));"
-        val createStatement = connection.createStatement()
-        createStatement.executeUpdate(create)
-
-        val insert  = "INSERT INTO Address (Nr, Name) VALUES(1, 'foo')"
-        val insertStatement = connection.prepareStatement(insert)
-        insertStatement.execute()
-
-        val select  =
-          """
-            |/*this is a comment*/
-            |SELECT * FROM Address""".stripMargin
-        val selectStatement = connection.prepareCall(select)
-        selectStatement.execute()
-
-        val update  = "UPDATE Address SET Name = 'bar' where Nr = 1"
-        val updateStatement = connection.createStatement()
-        updateStatement.execute(update)
-
-        val delete  = "DELETE FROM Address where Nr = 1"
-        val deleteStatement = connection.createStatement()
-        deleteStatement.execute(delete)
-
-        intercept[SQLException] {
-          val error = "SELECT * FROM NON_EXIST_TABLE"
-          val errorStatement = connection.createStatement()
-          errorStatement.execute(error)
+    "record the execution time of the INSERT operation" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-insert") {
+
+        val metricsListener = subscribeToMetrics()
+
+        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)
+      }
+    }
+
+    "record the execution time of SELECT operation" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-select") {
+
+        val metricsListener = subscribeToMetrics()
+
+        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)
+      }
+    }
+
+    "record the execution time of UPDATE operation" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-update") {
+
+        val metricsListener = subscribeToMetrics()
+
+        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)
+      }
+    }
+
+    "record the execution time of DELETE operation" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-insert") {
+
+        val metricsListener = subscribeToMetrics()
+
+        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)
       }
     }
+
+    "record the execution time of SLOW QUERIES based on the kamon.jdbc.slow-query-threshold" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-slow") {
+
+        val metricsListener = subscribeToMetrics()
+
+        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)
+      }
+    }
+
+    "count the total ERRORS" in new StatementsMetricsListenerFixture {
+      TraceRecorder.withNewTraceContext("jdbc-trace-slow") {
+
+        val metricsListener = subscribeToMetrics()
+
+        for(_ <- 1 to 10) {
+          intercept[SQLException] {
+            val error = "SELECT * FROM NON_EXIST_TABLE"
+            val errorStatement = connection.createStatement()
+            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
+    }
+  }
+
+  def expectStatementsMetrics(listener: TestProbe, waitTime: FiniteDuration): StatementsMetricsSnapshot = {
+    val tickSnapshot = within(waitTime) {
+      listener.expectMsgType[TickMetricSnapshot]
+    }
+    val statementsMetricsOption = tickSnapshot.metrics.get(StatementsMetrics(StatementInstrumentation.Statements))
+    statementsMetricsOption should not be empty
+    statementsMetricsOption.get.asInstanceOf[StatementsMetricsSnapshot]
   }
 }
 
+class NOPSlowQueryProcessor extends SlowQueryProcessor {
+  override def process(sql: String, executionTimeInMillis: Long, queryThresholdInMillis: Long): Unit = {/*do nothing!!!*/}
+}
+
+