+ core: added tests to dispatcher metrics

2 files changed, 112 insertions, 7 deletions

diff --git a/kamon-core/src/main/scala/kamon/metrics/dispatcher/DispatcherMetricsCollector.scala b/kamon-core/src/main/scala/kamon/metrics/dispatcher/DispatcherMetricsCollector.scala
index e7edce51..529cc542 100644
--- a/kamon-core/src/main/scala/kamon/metrics/dispatcher/DispatcherMetricsCollector.scala
+++ b/kamon-core/src/main/scala/kamon/metrics/dispatcher/DispatcherMetricsCollector.scala
@@ -18,17 +18,17 @@ package kamon.metrics.dispatcher
 
 import scala.concurrent.forkjoin.ForkJoinPool
 import java.util.concurrent.ThreadPoolExecutor
-import akka.dispatch.{ExecutorServiceDelegate, Dispatcher, MessageDispatcher}
+import akka.dispatch.{ ExecutorServiceDelegate, Dispatcher, MessageDispatcher }
 import java.lang.reflect.Method
 
 object DispatcherMetricsCollector {
 
-  private[this] type DispatcherMetrics = (Long, Long, Long, Long)
+  private[this]type DispatcherMetrics = (Long, Long, Long, Long)
 
-  private[this] def collectForkJoinMetrics(pool: ForkJoinPool):DispatcherMetrics = {
+  private[this] def collectForkJoinMetrics(pool: ForkJoinPool): DispatcherMetrics = {
     (pool.getParallelism, pool.getActiveThreadCount, (pool.getQueuedTaskCount + pool.getQueuedSubmissionCount), pool.getPoolSize)
   }
-  private[this] def collectExecutorMetrics(pool: ThreadPoolExecutor):DispatcherMetrics = {
+  private[this] def collectExecutorMetrics(pool: ThreadPoolExecutor): DispatcherMetrics = {
     (pool.getMaximumPoolSize, pool.getActiveCount, pool.getQueue.size(), pool.getPoolSize)
   }
 
@@ -44,13 +44,13 @@ object DispatcherMetricsCollector {
       case x: Dispatcher ⇒ {
         val executor = executorServiceMethod.invoke(x) match {
           case delegate: ExecutorServiceDelegate ⇒ delegate.executor
-          case other ⇒ other
+          case other                             ⇒ other
         }
 
         executor match {
-          case fjp: ForkJoinPool ⇒ collectForkJoinMetrics(fjp)
+          case fjp: ForkJoinPool       ⇒ collectForkJoinMetrics(fjp)
           case tpe: ThreadPoolExecutor ⇒ collectExecutorMetrics(tpe)
-          case anything ⇒ (0L, 0L, 0L, 0L)
+          case anything                ⇒ (0L, 0L, 0L, 0L)
         }
       }
       case _ ⇒ new DispatcherMetrics(0L, 0L, 0L, 0L)
diff --git a/kamon-core/src/test/scala/kamon/metrics/DispatcherMetricsSpec.scala b/kamon-core/src/test/scala/kamon/metrics/DispatcherMetricsSpec.scala
new file mode 100644
index 00000000..a5594ee2
--- /dev/null
+++ b/kamon-core/src/test/scala/kamon/metrics/DispatcherMetricsSpec.scala
@@ -0,0 +1,105 @@
+/* =========================================================================================
+ * 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 org.scalatest.{ WordSpecLike, Matchers }
+import akka.testkit.{ TestProbe, TestKitBase }
+import akka.actor.{ ActorRef, Props, ActorSystem }
+import com.typesafe.config.ConfigFactory
+import scala.concurrent.duration._
+import kamon.Kamon
+import kamon.metrics.Subscriptions.TickMetricSnapshot
+import kamon.metrics.DispatcherMetrics.DispatcherMetricSnapshot
+
+class DispatcherMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
+  implicit lazy val system: ActorSystem = ActorSystem("dispatcher-metrics-spec", ConfigFactory.parseString(
+    """
+      |kamon.metrics {
+      |  filters = [
+      |    {
+      |      dispatcher {
+      |        includes = ["*"]
+      |        excludes = ["dispatcher-explicitly-excluded"]
+      |      }
+      |    }
+      |  ]
+      |}
+      |
+      |dispatcher-explicitly-excluded {
+      |   type = "Dispatcher"
+      |   executor = "fork-join-executor"
+      |}
+      |
+      |tracked-dispatcher {
+      |   type = "Dispatcher"
+      |   executor = "thread-pool-executor"
+      |}
+      |
+    """.stripMargin))
+
+  "the Kamon dispatcher metrics" should {
+    "respect the configured include and exclude filters" in {
+      system.actorOf(Props[DelayableActor].withDispatcher("tracked-dispatcher"), "actor-with-tracked-dispatcher")
+      system.actorOf(Props[DelayableActor].withDispatcher("dispatcher-explicitly-excluded"), "actor-with-excluded-dispatcher")
+
+      Kamon(Metrics).subscribe(DispatcherMetrics, "*", testActor, permanently = true)
+      expectMsgType[TickMetricSnapshot]
+
+      within(2 seconds) {
+        val tickSnapshot = expectMsgType[TickMetricSnapshot]
+        tickSnapshot.metrics.keys should contain(DispatcherMetrics("tracked-dispatcher"))
+        tickSnapshot.metrics.keys should not contain (DispatcherMetrics("dispatcher-explicitly-excluded"))
+      }
+    }
+
+    "record maximumPoolSize, runningThreadCount, queueTaskCount, poolSize metrics" in new DelayableActorFixture {
+      val (delayable, metricsListener) = delayableActor("worker-actor", "tracked-dispatcher")
+
+      for (_ ← 1 to 100) {
+        delayable ! Discard
+      }
+
+      val dispatcherMetrics = expectDispatcherMetrics("tracked-dispatcher", metricsListener, 3 seconds)
+      dispatcherMetrics.maximumPoolSize.max should be <= 18L //6 cores
+      dispatcherMetrics.poolSize.max should be(12L)
+      dispatcherMetrics.queueTaskCount.max should be(0L)
+      dispatcherMetrics.runningThreadCount.max should be(0L)
+    }
+
+  }
+
+  def expectDispatcherMetrics(dispatcherId: String, listener: TestProbe, waitTime: FiniteDuration): DispatcherMetricSnapshot = {
+    val tickSnapshot = within(waitTime) {
+      listener.expectMsgType[TickMetricSnapshot]
+    }
+    val dispatcherMetricsOption = tickSnapshot.metrics.get(DispatcherMetrics(dispatcherId))
+    dispatcherMetricsOption should not be empty
+    dispatcherMetricsOption.get.asInstanceOf[DispatcherMetricSnapshot]
+  }
+
+  trait DelayableActorFixture {
+    def delayableActor(name: String, dispatcher: String): (ActorRef, TestProbe) = {
+      val actor = system.actorOf(Props[DelayableActor].withDispatcher(dispatcher), name)
+      val metricsListener = TestProbe()
+
+      Kamon(Metrics).subscribe(DispatcherMetrics, "*", metricsListener.ref, permanently = true)
+      // Wait for one empty snapshot before proceeding to the test.
+      metricsListener.expectMsgType[TickMetricSnapshot]
+
+      (actor, metricsListener)
+    }
+  }
+}