aboutsummaryrefslogtreecommitdiff
path: root/kamon-akka/src/test/scala/kamon/metric
diff options
context:
space:
mode:
authorIvan Topolnjak <ivantopo@gmail.com>2014-12-28 07:59:59 +0100
committerIvan Topolnjak <ivantopo@gmail.com>2014-12-28 18:15:48 +0100
commit4d0e37594cb223b091ba2ec126eabe89ac8c13f8 (patch)
tree5989bcd31a1055fb8a9481c764ae7c6e3ad668bb /kamon-akka/src/test/scala/kamon/metric
parent0be6a850b1e458bdc961c956daddb8d8cc1682b5 (diff)
downloadKamon-4d0e37594cb223b091ba2ec126eabe89ac8c13f8.tar.gz
Kamon-4d0e37594cb223b091ba2ec126eabe89ac8c13f8.tar.bz2
Kamon-4d0e37594cb223b091ba2ec126eabe89ac8c13f8.zip
! core,akka: separate all akka instrumentation into it's own kamon-akka module, related to #136.
All Akka-related instrumentation and code has been moved to the kamon-akka module, including the filters for actor, dispatcher and router metrics. Also the following changes are included: - Router Metrics are now working properly, related to #139. - Cleanup the log output for this module, related to #142. - Some minor cleanups in various tests. This PR breaks the reporting modules which will need to wait for #141 to be ready to come back to life.
Diffstat (limited to 'kamon-akka/src/test/scala/kamon/metric')
-rw-r--r--kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala228
-rw-r--r--kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala110
-rw-r--r--kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala296
3 files changed, 634 insertions, 0 deletions
diff --git a/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala
new file mode 100644
index 00000000..6d16386b
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala
@@ -0,0 +1,228 @@
+/* =========================================================================================
+ * Copyright © 2013 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.metric
+
+import java.nio.LongBuffer
+
+import kamon.Kamon
+import kamon.akka.ActorMetrics
+import kamon.metric.ActorMetricsTestActor._
+import org.scalatest.{ BeforeAndAfterAll, WordSpecLike, Matchers }
+import akka.testkit.{ ImplicitSender, TestProbe, TestKitBase }
+import akka.actor._
+import com.typesafe.config.ConfigFactory
+import scala.concurrent.duration._
+import ActorMetrics.{ ActorMetricsRecorder, ActorMetricSnapshot }
+
+class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with ImplicitSender with BeforeAndAfterAll {
+ implicit lazy val system: ActorSystem = ActorSystem("actor-metrics-spec", ConfigFactory.parseString(
+ """
+ |kamon.metrics {
+ | tick-interval = 1 hour
+ | default-collection-context-buffer-size = 10
+ |
+ | filters = [
+ | {
+ | actor {
+ | includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect", "user/stop" ]
+ | excludes = [ "user/tracked-explicitly-excluded"]
+ | }
+ | }
+ | ]
+ | precision.actor {
+ | processing-time {
+ | highest-trackable-value = 3600000000000
+ | significant-value-digits = 2
+ | }
+ |
+ | time-in-mailbox {
+ | highest-trackable-value = 3600000000000
+ | significant-value-digits = 2
+ | }
+ |
+ | mailbox-size {
+ | refresh-interval = 1 hour
+ | highest-trackable-value = 999999999
+ | significant-value-digits = 2
+ | }
+ | }
+ |}
+ |
+ |akka.loglevel = OFF
+ |
+ """.stripMargin))
+
+ "the Kamon actor metrics" should {
+ "respect the configured include and exclude filters" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("tracked-actor")
+ actorMetricsRecorderOf(trackedActor) should not be empty
+
+ val nonTrackedActor = createTestActor("non-tracked-actor")
+ actorMetricsRecorderOf(nonTrackedActor) shouldBe empty
+
+ val trackedButExplicitlyExcluded = createTestActor("tracked-explicitly-excluded")
+ actorMetricsRecorderOf(trackedButExplicitlyExcluded) shouldBe empty
+ }
+
+ "reset all recording instruments after taking a snapshot" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("clean-after-collect")
+
+ for (_ ← 1 to 100) {
+ for (i ← 1 to 100) {
+ trackedActor ! Discard
+ }
+ trackedActor ! Fail
+ trackedActor ! Ping
+ expectMsg(Pong)
+
+ val firstSnapshot = collectMetricsOf(trackedActor).get
+ firstSnapshot.errors.count should be(1L)
+ firstSnapshot.mailboxSize.numberOfMeasurements should be > 0L
+ firstSnapshot.processingTime.numberOfMeasurements should be(102L) // 102 examples
+ firstSnapshot.timeInMailbox.numberOfMeasurements should be(102L) // 102 examples
+
+ val secondSnapshot = collectMetricsOf(trackedActor).get // Ensure that the recorders are clean
+ secondSnapshot.errors.count should be(0L)
+ secondSnapshot.mailboxSize.numberOfMeasurements should be(3L) // min, max and current
+ secondSnapshot.processingTime.numberOfMeasurements should be(0L)
+ secondSnapshot.timeInMailbox.numberOfMeasurements should be(0L)
+ }
+ }
+
+ "record the processing-time of the receive function" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("measuring-processing-time")
+
+ trackedActor ! TrackTimings(sleep = Some(100 millis))
+ val timings = expectMsgType[TrackedTimings]
+ val snapshot = collectMetricsOf(trackedActor).get
+
+ snapshot.processingTime.numberOfMeasurements should be(1L)
+ snapshot.processingTime.recordsIterator.next().count should be(1L)
+ snapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos)
+ }
+
+ "record the number of errors" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("measuring-errors")
+
+ for (i ← 1 to 10) { trackedActor ! Fail }
+ trackedActor ! Ping
+ expectMsg(Pong)
+ val snapshot = collectMetricsOf(trackedActor).get
+
+ snapshot.errors.count should be(10)
+ }
+
+ "record the mailbox-size" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("measuring-mailbox-size")
+
+ trackedActor ! TrackTimings(sleep = Some(100 millis))
+ for (i ← 1 to 10) {
+ trackedActor ! Discard
+ }
+ trackedActor ! Ping
+
+ val timings = expectMsgType[TrackedTimings]
+ expectMsg(Pong)
+ val snapshot = collectMetricsOf(trackedActor).get
+
+ snapshot.mailboxSize.min should be(0L)
+ snapshot.mailboxSize.max should be(11L +- 1L)
+ }
+
+ "record the time-in-mailbox" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("measuring-time-in-mailbox")
+
+ trackedActor ! TrackTimings(sleep = Some(100 millis))
+ val timings = expectMsgType[TrackedTimings]
+ val snapshot = collectMetricsOf(trackedActor).get
+
+ snapshot.timeInMailbox.numberOfMeasurements should be(1L)
+ snapshot.timeInMailbox.recordsIterator.next().count should be(1L)
+ snapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos)
+ }
+
+ "clean up the associated recorder when the actor is stopped" in new ActorMetricsFixtures {
+ val trackedActor = createTestActor("stop")
+
+ val deathWatcher = TestProbe()
+ deathWatcher.watch(trackedActor)
+ trackedActor ! PoisonPill
+ deathWatcher.expectTerminated(trackedActor)
+
+ actorMetricsRecorderOf(trackedActor) shouldBe empty
+ }
+ }
+
+ override protected def afterAll(): Unit = shutdown()
+
+ trait ActorMetricsFixtures {
+ val collectionContext = new CollectionContext {
+ val buffer: LongBuffer = LongBuffer.allocate(10000)
+ }
+
+ def actorRecorderName(ref: ActorRef): String = ref.path.elements.mkString("/")
+
+ def actorMetricsRecorderOf(ref: ActorRef): Option[ActorMetricsRecorder] =
+ Kamon(Metrics)(system).storage.get(ActorMetrics(actorRecorderName(ref))).map(_.asInstanceOf[ActorMetricsRecorder])
+
+ def collectMetricsOf(ref: ActorRef): Option[ActorMetricSnapshot] = {
+ Thread.sleep(5) // Just in case the test advances a bit faster than the actor being tested.
+ actorMetricsRecorderOf(ref).map(_.collect(collectionContext))
+ }
+
+ def createTestActor(name: String): ActorRef = {
+ val actor = system.actorOf(Props[ActorMetricsTestActor], name)
+ val initialiseListener = TestProbe()
+
+ // Ensure that the router has been created before returning.
+ actor.tell(Ping, initialiseListener.ref)
+ initialiseListener.expectMsg(Pong)
+
+ // Cleanup all the metric recording instruments:
+ collectMetricsOf(actor)
+
+ actor
+ }
+ }
+}
+
+class ActorMetricsTestActor extends Actor {
+ def receive = {
+ case Discard ⇒
+ case Fail ⇒ throw new ArithmeticException("Division by zero.")
+ case Ping ⇒ sender ! Pong
+ case TrackTimings(sendTimestamp, sleep) ⇒ {
+ val dequeueTimestamp = System.nanoTime()
+ sleep.map(s ⇒ Thread.sleep(s.toMillis))
+ val afterReceiveTimestamp = System.nanoTime()
+
+ sender ! TrackedTimings(sendTimestamp, dequeueTimestamp, afterReceiveTimestamp)
+ }
+ }
+}
+
+object ActorMetricsTestActor {
+ case object Ping
+ case object Pong
+ case object Fail
+ case object Discard
+
+ case class TrackTimings(sendTimestamp: Long = System.nanoTime(), sleep: Option[Duration] = None)
+ case class TrackedTimings(sendTimestamp: Long, dequeueTimestamp: Long, afterReceiveTimestamp: Long) {
+ def approximateTimeInMailbox: Long = dequeueTimestamp - sendTimestamp
+ def approximateProcessingTime: Long = afterReceiveTimestamp - dequeueTimestamp
+ }
+}
diff --git a/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala
new file mode 100644
index 00000000..55af3f2e
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala
@@ -0,0 +1,110 @@
+/* =========================================================================================
+ * 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.metric
+
+import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.testkit.{ TestKitBase, TestProbe }
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.akka.DispatcherMetrics
+import DispatcherMetrics.DispatcherMetricSnapshot
+import kamon.metric.Subscriptions.TickMetricSnapshot
+import org.scalatest.{ Matchers, WordSpecLike }
+
+import scala.concurrent.duration._
+
+class DispatcherMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
+ implicit lazy val system: ActorSystem = ActorSystem("dispatcher-metrics-spec", ConfigFactory.parseString(
+ """
+ |kamon.metrics {
+ | tick-interval = 1 second
+ | default-collection-context-buffer-size = 10
+ |
+ | 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[ActorMetricsTestActor].withDispatcher("tracked-dispatcher"), "actor-with-tracked-dispatcher")
+ system.actorOf(Props[ActorMetricsTestActor].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 <= 64L //fail in travis
+ dispatcherMetrics.poolSize.max should be <= 22L //fail in travis
+ 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[ActorMetricsTestActor].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)
+ }
+ }
+}
diff --git a/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala
new file mode 100644
index 00000000..abc195ba
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala
@@ -0,0 +1,296 @@
+/* =========================================================================================
+ * 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.metric
+
+import java.nio.LongBuffer
+
+import akka.actor._
+import akka.kamon.instrumentation.ActorCellMetrics
+import akka.routing._
+import akka.testkit.{ ImplicitSender, TestKitBase, TestProbe }
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.akka.{ RouterMetrics, ActorMetrics }
+import ActorMetrics.{ ActorMetricSnapshot, ActorMetricsRecorder }
+import RouterMetrics._
+import kamon.metric.RouterMetricsTestActor._
+import kamon.metric.Subscriptions.TickMetricSnapshot
+import kamon.metric.instrument.{ Counter, Histogram }
+import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike }
+
+import scala.concurrent.duration._
+
+class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with ImplicitSender with BeforeAndAfterAll {
+ implicit lazy val system: ActorSystem = ActorSystem("router-metrics-spec", ConfigFactory.parseString(
+ """
+ |kamon.metrics {
+ | tick-interval = 1 hour
+ | default-collection-context-buffer-size = 10
+ |
+ | filters = [
+ | {
+ | router {
+ | includes = [ "user/tracked-*", "user/measuring-*", "user/stop-*" ]
+ | excludes = [ "user/tracked-explicitly-excluded-*"]
+ | }
+ | }
+ | ]
+ | precision {
+ | default-histogram-precision {
+ | highest-trackable-value = 3600000000000
+ | significant-value-digits = 2
+ | }
+ | }
+ |}
+ |
+ |akka.loglevel = OFF
+ |
+ """.stripMargin))
+
+ "the Kamon router metrics" should {
+ "respect the configured include and exclude filters" in new RouterMetricsFixtures {
+ createTestPoolRouter("tracked-pool-router")
+ createTestGroupRouter("tracked-group-router")
+ createTestPoolRouter("non-tracked-pool-router")
+ createTestGroupRouter("non-tracked-group-router")
+ createTestPoolRouter("tracked-explicitly-excluded-pool-router")
+ createTestGroupRouter("tracked-explicitly-excluded-group-router")
+
+ routerMetricsRecorderOf("user/tracked-pool-router") should not be empty
+ routerMetricsRecorderOf("user/tracked-group-router") should not be empty
+ routerMetricsRecorderOf("user/non-tracked-pool-router") shouldBe empty
+ routerMetricsRecorderOf("user/non-tracked-group-router") shouldBe empty
+ routerMetricsRecorderOf("user/tracked-explicitly-excluded-pool-router") shouldBe empty
+ routerMetricsRecorderOf("user/tracked-explicitly-excluded-group-router") shouldBe empty
+ }
+
+ "record the routing-time of the receive function for pool routers" in new RouterMetricsFixtures {
+ val listener = TestProbe()
+ val router = createTestPoolRouter("measuring-routing-time-in-pool-router")
+
+ router.tell(Ping, listener.ref)
+ listener.expectMsg(Pong)
+ val routerSnapshot = collectMetricsOf("user/measuring-routing-time-in-pool-router").get
+
+ routerSnapshot.routingTime.numberOfMeasurements should be(1L)
+ }
+
+ "record the routing-time of the receive function for group routers" in new RouterMetricsFixtures {
+ val listener = TestProbe()
+ val router = createTestGroupRouter("measuring-routing-time-in-group-router")
+
+ router.tell(Ping, listener.ref)
+ listener.expectMsg(Pong)
+ val routerSnapshot = collectMetricsOf("user/measuring-routing-time-in-group-router").get
+
+ routerSnapshot.routingTime.numberOfMeasurements should be(1L)
+ }
+
+ "record the processing-time of the receive function for pool routers" in new RouterMetricsFixtures {
+ val timingsListener = TestProbe()
+ val router = createTestPoolRouter("measuring-processing-time-in-pool-router")
+
+ router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+ val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+ val routerSnapshot = collectMetricsOf("user/measuring-processing-time-in-pool-router").get
+
+ routerSnapshot.processingTime.numberOfMeasurements should be(1L)
+ routerSnapshot.processingTime.recordsIterator.next().count should be(1L)
+ routerSnapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos)
+ }
+
+ "record the processing-time of the receive function for group routers" in new RouterMetricsFixtures {
+ val timingsListener = TestProbe()
+ val router = createTestGroupRouter("measuring-processing-time-in-group-router")
+
+ router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+ val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+ val routerSnapshot = collectMetricsOf("user/measuring-processing-time-in-group-router").get
+
+ routerSnapshot.processingTime.numberOfMeasurements should be(1L)
+ routerSnapshot.processingTime.recordsIterator.next().count should be(1L)
+ routerSnapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos)
+ }
+
+ "record the number of errors for pool routers" in new RouterMetricsFixtures {
+ val listener = TestProbe()
+ val router = createTestPoolRouter("measuring-errors-in-pool-router")
+
+ for (i ← 1 to 10) {
+ router.tell(Fail, listener.ref)
+ }
+
+ router.tell(Ping, listener.ref)
+ listener.expectMsg(Pong)
+
+ val routerSnapshot = collectMetricsOf("user/measuring-errors-in-pool-router").get
+ routerSnapshot.errors.count should be(10L)
+ }
+
+ "record the number of errors for group routers" in new RouterMetricsFixtures {
+ val listener = TestProbe()
+ val router = createTestGroupRouter("measuring-errors-in-group-router")
+
+ for (i ← 1 to 10) {
+ router.tell(Fail, listener.ref)
+ }
+
+ router.tell(Ping, listener.ref)
+ listener.expectMsg(Pong)
+
+ val routerSnapshot = collectMetricsOf("user/measuring-errors-in-group-router").get
+ routerSnapshot.errors.count should be(10L)
+ }
+
+ "record the time-in-mailbox for pool routers" in new RouterMetricsFixtures {
+ val timingsListener = TestProbe()
+ val router = createTestPoolRouter("measuring-time-in-mailbox-in-pool-router")
+
+ router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+ val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+ val routerSnapshot = collectMetricsOf("user/measuring-time-in-mailbox-in-pool-router").get
+
+ routerSnapshot.timeInMailbox.numberOfMeasurements should be(1L)
+ routerSnapshot.timeInMailbox.recordsIterator.next().count should be(1L)
+ routerSnapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos)
+ }
+
+ "record the time-in-mailbox for group routers" in new RouterMetricsFixtures {
+ val timingsListener = TestProbe()
+ val router = createTestGroupRouter("measuring-time-in-mailbox-in-group-router")
+
+ router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+ val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+ val routerSnapshot = collectMetricsOf("user/measuring-time-in-mailbox-in-group-router").get
+
+ routerSnapshot.timeInMailbox.numberOfMeasurements should be(1L)
+ routerSnapshot.timeInMailbox.recordsIterator.next().count should be(1L)
+ routerSnapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos)
+ }
+
+ "clean up the associated recorder when the pool router is stopped" in new RouterMetricsFixtures {
+ val trackedRouter = createTestPoolRouter("stop-in-pool-router")
+ routerMetricsRecorderOf("user/stop-in-pool-router") should not be empty
+
+ val deathWatcher = TestProbe()
+ deathWatcher.watch(trackedRouter)
+ trackedRouter ! PoisonPill
+ deathWatcher.expectTerminated(trackedRouter)
+
+ routerMetricsRecorderOf("user/stop-in-pool-router") shouldBe empty
+ }
+
+ "clean up the associated recorder when the group router is stopped" in new RouterMetricsFixtures {
+ val trackedRouter = createTestPoolRouter("stop-in-group-router")
+ routerMetricsRecorderOf("user/stop-in-group-router") should not be empty
+
+ val deathWatcher = TestProbe()
+ deathWatcher.watch(trackedRouter)
+ trackedRouter ! PoisonPill
+ deathWatcher.expectTerminated(trackedRouter)
+
+ routerMetricsRecorderOf("user/stop-in-group-router") shouldBe empty
+ }
+ }
+
+ override protected def afterAll(): Unit = shutdown()
+
+ trait RouterMetricsFixtures {
+ val collectionContext = new CollectionContext {
+ val buffer: LongBuffer = LongBuffer.allocate(10000)
+ }
+
+ def routerMetricsRecorderOf(routerName: String): Option[RouterMetricsRecorder] =
+ Kamon(Metrics)(system).storage.get(RouterMetrics(routerName)).map(_.asInstanceOf[RouterMetricsRecorder])
+
+ def collectMetricsOf(routerName: String): Option[RouterMetricSnapshot] = {
+ Thread.sleep(5) // Just in case the test advances a bit faster than the actor being tested.
+ routerMetricsRecorderOf(routerName).map(_.collect(collectionContext))
+ }
+
+ def createTestGroupRouter(routerName: String): ActorRef = {
+ val routees = Vector.fill(5) {
+ system.actorOf(Props[RouterMetricsTestActor])
+ }
+
+ val group = system.actorOf(RoundRobinGroup(routees.map(_.path.toStringWithoutAddress)).props(), routerName)
+
+ //val router = system.actorOf(RoundRobinPool(5).props(Props[RouterMetricsTestActor]), routerName)
+ val initialiseListener = TestProbe()
+
+ // Ensure that the router has been created before returning.
+ group.tell(Ping, initialiseListener.ref)
+ initialiseListener.expectMsg(Pong)
+
+ // Cleanup all the metric recording instruments:
+ collectMetricsOf("user/" + routerName)
+
+ group
+ }
+
+ def createTestPoolRouter(routerName: String): ActorRef = {
+ val router = system.actorOf(RoundRobinPool(5).props(Props[RouterMetricsTestActor]), routerName)
+ val initialiseListener = TestProbe()
+
+ // Ensure that the router has been created before returning.
+ router.tell(Ping, initialiseListener.ref)
+ initialiseListener.expectMsg(Pong)
+
+ // Cleanup all the metric recording instruments:
+ collectMetricsOf("user/" + routerName)
+
+ router
+ }
+ }
+
+ trait ActorMetricsFixtures {
+ val collectionContext = new CollectionContext {
+ val buffer: LongBuffer = LongBuffer.allocate(10000)
+ }
+
+ def createTestActor(name: String): ActorRef = system.actorOf(Props[ActorMetricsTestActor], name)
+
+ def takeSnapshotOf(amr: ActorMetricsRecorder): ActorMetricSnapshot = amr.collect(collectionContext)
+ }
+}
+
+class RouterMetricsTestActor extends Actor {
+ def receive = {
+ case Discard ⇒
+ case Fail ⇒ throw new ArithmeticException("Division by zero.")
+ case Ping ⇒ sender ! Pong
+ case RouterTrackTimings(sendTimestamp, sleep) ⇒ {
+ val dequeueTimestamp = System.nanoTime()
+ sleep.map(s ⇒ Thread.sleep(s.toMillis))
+ val afterReceiveTimestamp = System.nanoTime()
+
+ sender ! RouterTrackedTimings(sendTimestamp, dequeueTimestamp, afterReceiveTimestamp)
+ }
+ }
+}
+
+object RouterMetricsTestActor {
+ case object Ping
+ case object Pong
+ case object Fail
+ case object Discard
+
+ case class RouterTrackTimings(sendTimestamp: Long = System.nanoTime(), sleep: Option[Duration] = None)
+ case class RouterTrackedTimings(sendTimestamp: Long, dequeueTimestamp: Long, afterReceiveTimestamp: Long) {
+ def approximateTimeInMailbox: Long = dequeueTimestamp - sendTimestamp
+ def approximateProcessingTime: Long = afterReceiveTimestamp - dequeueTimestamp
+ }
+}