1 files changed, 211 insertions, 0 deletions
diff --git a/kamon-akka/src/test/scala/kamon/akka/DispatcherMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/akka/DispatcherMetricsSpec.scala
new file mode 100644
index 00000000..209b5cf5
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/DispatcherMetricsSpec.scala
@@ -0,0 +1,211 @@
+/* =========================================================================================
+ * 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.akka
+
+import java.nio.LongBuffer
+
+import akka.actor.{ ActorRef, ActorSystem }
+import akka.dispatch.MessageDispatcher
+import akka.testkit.{ TestKitBase, TestProbe }
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.metric.instrument.CollectionContext
+import kamon.metric.{ EntityRecorder, EntitySnapshot }
+import kamon.testkit.BaseKamonSpec
+import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike }
+
+import scala.concurrent.duration._
+import scala.concurrent.{ Await, Future }
+
+class DispatcherMetricsSpec extends BaseKamonSpec("dispatcher-metrics-spec") {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |kamon.metric {
+        |  tick-interval = 1 hour
+        |  default-collection-context-buffer-size = 10
+        |
+        |  filters = {
+        |    akka-dispatcher {
+        |      includes = [ "*" ]
+        |      excludes = [ "explicitly-excluded" ]
+        |    }
+        |  }
+        |
+        |  default-instrument-settings {
+        |    gauge.refresh-interval = 1 hour
+        |    min-max-counter.refresh-interval = 1 hour
+        |  }
+        |}
+        |
+        |explicitly-excluded {
+        |  type = "Dispatcher"
+        |  executor = "fork-join-executor"
+        |}
+        |
+        |tracked-fjp {
+        |  type = "Dispatcher"
+        |  executor = "fork-join-executor"
+        |
+        |  fork-join-executor {
+        |    parallelism-min = 8
+        |    parallelism-factor = 100.0
+        |    parallelism-max = 22
+        |  }
+        |}
+        |
+        |tracked-tpe {
+        |  type = "Dispatcher"
+        |  executor = "thread-pool-executor"
+        |
+        |  thread-pool-executor {
+        |    core-pool-size-min = 7
+        |    core-pool-size-factor = 100.0
+        |    max-pool-size-factor  = 100.0
+        |    max-pool-size-max = 21
+        |  }
+        |}
+        |
+      """.stripMargin)
+
+  "the Kamon dispatcher metrics" should {
+    "respect the configured include and exclude filters" in {
+      val defaultDispatcher = forceInit(system.dispatchers.lookup("akka.actor.default-dispatcher"))
+      val fjpDispatcher = forceInit(system.dispatchers.lookup("tracked-fjp"))
+      val tpeDispatcher = forceInit(system.dispatchers.lookup("tracked-tpe"))
+      val excludedDispatcher = forceInit(system.dispatchers.lookup("explicitly-excluded"))
+
+      findDispatcherRecorder(defaultDispatcher) shouldNot be(empty)
+      findDispatcherRecorder(fjpDispatcher) shouldNot be(empty)
+      findDispatcherRecorder(tpeDispatcher) shouldNot be(empty)
+      findDispatcherRecorder(excludedDispatcher) should be(empty)
+    }
+
+    "record metrics for a dispatcher with thread-pool-executor" in {
+      implicit val tpeDispatcher = system.dispatchers.lookup("tracked-tpe")
+      refreshDispatcherInstruments(tpeDispatcher)
+      collectDispatcherMetrics(tpeDispatcher)
+
+      Await.result({
+        Future.sequence {
+          for (_ ← 1 to 100) yield submit(tpeDispatcher)
+        }
+      }, 5 seconds)
+
+      refreshDispatcherInstruments(tpeDispatcher)
+      val snapshot = collectDispatcherMetrics(tpeDispatcher)
+
+      snapshot.gauge("active-threads") should not be empty
+      snapshot.gauge("pool-size").get.min should be >= 7L
+      snapshot.gauge("pool-size").get.max should be <= 21L
+      snapshot.gauge("max-pool-size").get.max should be(21)
+      snapshot.gauge("core-pool-size").get.max should be(21)
+      snapshot.gauge("processed-tasks").get.max should be(102L +- 5L)
+
+      // The processed tasks should be reset to 0 if no more tasks are submitted.
+      val secondSnapshot = collectDispatcherMetrics(tpeDispatcher)
+      secondSnapshot.gauge("processed-tasks").get.max should be(0)
+    }
+
+    "record metrics for a dispatcher with fork-join-executor" in {
+      implicit val fjpDispatcher = system.dispatchers.lookup("tracked-fjp")
+      collectDispatcherMetrics(fjpDispatcher)
+
+      Await.result({
+        Future.sequence {
+          for (_ ← 1 to 100) yield submit(fjpDispatcher)
+        }
+      }, 5 seconds)
+
+      refreshDispatcherInstruments(fjpDispatcher)
+      val snapshot = collectDispatcherMetrics(fjpDispatcher)
+
+      snapshot.minMaxCounter("parallelism").get.max should be(22)
+      snapshot.gauge("pool-size").get.min should be >= 0L
+      snapshot.gauge("pool-size").get.max should be <= 22L
+      snapshot.gauge("active-threads").get.max should be >= 0L
+      snapshot.gauge("running-threads").get.max should be >= 0L
+      snapshot.gauge("queued-task-count").get.max should be(0)
+
+    }
+
+    "clean up the metrics recorders after a dispatcher is shut down" in {
+      implicit val tpeDispatcher = system.dispatchers.lookup("tracked-tpe")
+      implicit val fjpDispatcher = system.dispatchers.lookup("tracked-fjp")
+
+      findDispatcherRecorder(fjpDispatcher) shouldNot be(empty)
+      findDispatcherRecorder(tpeDispatcher) shouldNot be(empty)
+
+      shutdownDispatcher(tpeDispatcher)
+      shutdownDispatcher(fjpDispatcher)
+
+      findDispatcherRecorder(fjpDispatcher) should be(empty)
+      findDispatcherRecorder(tpeDispatcher) should be(empty)
+    }
+
+  }
+
+  def actorRecorderName(ref: ActorRef): String = ref.path.elements.mkString("/")
+
+  def findDispatcherRecorder(dispatcher: MessageDispatcher): Option[EntityRecorder] =
+    Kamon.metrics.find(dispatcher.id, "akka-dispatcher")
+
+  def collectDispatcherMetrics(dispatcher: MessageDispatcher): EntitySnapshot =
+    findDispatcherRecorder(dispatcher).map(_.collect(collectionContext)).get
+
+  def refreshDispatcherInstruments(dispatcher: MessageDispatcher): Unit = {
+    findDispatcherRecorder(dispatcher) match {
+      case Some(tpe: ThreadPoolExecutorDispatcherMetrics) ⇒
+        tpe.processedTasks.refreshValue()
+        tpe.activeThreads.refreshValue()
+        tpe.maxPoolSize.refreshValue()
+        tpe.poolSize.refreshValue()
+        tpe.corePoolSize.refreshValue()
+
+      case Some(fjp: ForkJoinPoolDispatcherMetrics) ⇒
+        fjp.activeThreads.refreshValue()
+        fjp.poolSize.refreshValue()
+        fjp.queuedTaskCount.refreshValue()
+        fjp.paralellism.refreshValues()
+        fjp.runningThreads.refreshValue()
+
+      case other ⇒
+    }
+  }
+
+  def forceInit(dispatcher: MessageDispatcher): MessageDispatcher = {
+    val listener = TestProbe()
+    Future {
+      listener.ref ! "init done"
+    }(dispatcher)
+    listener.expectMsg("init done")
+
+    dispatcher
+  }
+
+  def submit(dispatcher: MessageDispatcher): Future[String] = Future {
+    "hello"
+  }(dispatcher)
+
+  def shutdownDispatcher(dispatcher: MessageDispatcher): Unit = {
+    val shutdownMethod = dispatcher.getClass.getDeclaredMethod("shutdown")
+    shutdownMethod.setAccessible(true)
+    shutdownMethod.invoke(dispatcher)
+  }
+
+  override protected def afterAll(): Unit = system.shutdown()
+}
+