Merge branch 'single-kamon-instance-per-jvm' into release-legacy-akka-2.2

Conflicts:
	kamon-akka-remote/src/test/scala/kamon/akka/instrumentation/RemotingInstrumentationSpec.scala
	kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala
	kamon-core/src/main/scala/kamon/instrumentation/akka/AskPatternInstrumentation.scala
	kamon-core/src/main/scala/kamon/metric/MetricsExtension.scala
	kamon-core/src/main/scala/kamon/metric/Subscriptions.scala
	kamon-core/src/main/scala/kamon/metric/instrument/Gauge.scala
	kamon-core/src/main/scala/kamon/metric/instrument/MinMaxCounter.scala
	kamon-core/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala
	kamon-core/src/test/scala/kamon/metric/ActorMetricsSpec.scala
	kamon-core/src/test/scala/kamon/metric/RouterMetricsSpec.scala
	kamon-core/src/test/scala/kamon/metric/SubscriptionsProtocolSpec.scala
	kamon-core/src/test/scala/kamon/metric/TickMetricSnapshotBufferSpec.scala
	kamon-core/src/test/scala/kamon/metric/TraceMetricsSpec.scala
	kamon-core/src/test/scala/kamon/metric/UserMetricsSpec.scala
	kamon-core/src/test/scala/kamon/trace/TraceContextManipulationSpec.scala
	kamon-datadog/src/main/scala/kamon/datadog/Datadog.scala
	kamon-newrelic/src/main/scala/kamon/newrelic/Agent.scala
	kamon-newrelic/src/main/scala/kamon/newrelic/MetricReporter.scala
	kamon-play/src/main/scala/kamon/play/Play.scala
	kamon-play/src/main/scala/kamon/play/action/KamonTraceActions.scala
	kamon-play/src/main/scala/kamon/play/instrumentation/RequestInstrumentation.scala
	kamon-play/src/main/scala/kamon/play/instrumentation/WSInstrumentation.scala
	kamon-play/src/test/scala/kamon/play/RequestInstrumentationSpec.scala
	kamon-statsd/src/main/scala/kamon/statsd/StatsD.scala
	project/Dependencies.scala
	version.sbt

7 files changed, 1100 insertions, 0 deletions

diff --git a/kamon-akka/src/test/scala/kamon/akka/ActorMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/akka/ActorMetricsSpec.scala
new file mode 100644
index 00000000..0d8d41e3
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/ActorMetricsSpec.scala
@@ -0,0 +1,215 @@
+/* =========================================================================================
+ * 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.akka
+
+import java.nio.LongBuffer
+
+import akka.actor._
+import akka.testkit.TestProbe
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.akka.ActorMetricsTestActor._
+import kamon.metric.EntitySnapshot
+import kamon.metric.instrument.CollectionContext
+import kamon.testkit.BaseKamonSpec
+
+import scala.concurrent.duration._
+
+class ActorMetricsSpec extends BaseKamonSpec("actor-metrics-spec") {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |kamon.metric {
+        |  tick-interval = 1 hour
+        |  default-collection-context-buffer-size = 10
+        |
+        |  filters {
+        |    akka-actor {
+        |      includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect", "user/stop" ]
+        |      excludes = [ "user/tracked-explicitly-excluded", "user/non-tracked-actor" ]
+        |    }
+        |  }
+        |
+        |  instrument-settings {
+        |    akka-actor.mailbox-size.refresh-interval = 1 hour
+        |  }
+        |}
+        |
+        |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.counter("errors").get.count should be(1L)
+        firstSnapshot.minMaxCounter("mailbox-size").get.numberOfMeasurements should be > 0L
+        firstSnapshot.histogram("processing-time").get.numberOfMeasurements should be(102L) // 102 examples
+        firstSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(102L) // 102 examples
+
+        val secondSnapshot = collectMetricsOf(trackedActor).get // Ensure that the recorders are clean
+        secondSnapshot.counter("errors").get.count should be(0L)
+        secondSnapshot.minMaxCounter("mailbox-size").get.numberOfMeasurements should be(3L) // min, max and current
+        secondSnapshot.histogram("processing-time").get.numberOfMeasurements should be(0L)
+        secondSnapshot.histogram("time-in-mailbox").get.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.histogram("processing-time").get.numberOfMeasurements should be(1L)
+      snapshot.histogram("processing-time").get.recordsIterator.next().count should be(1L)
+      snapshot.histogram("processing-time").get.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.counter("errors").get.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.minMaxCounter("mailbox-size").get.min should be(0L)
+      snapshot.minMaxCounter("mailbox-size").get.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.histogram("time-in-mailbox").get.numberOfMeasurements should be(1L)
+      snapshot.histogram("time-in-mailbox").get.recordsIterator.next().count should be(1L)
+      snapshot.histogram("time-in-mailbox").get.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 firstRecorder = actorMetricsRecorderOf(trackedActor).get
+
+      // Killing the actor should remove it's ActorMetrics and registering again bellow should create a new one.
+      val deathWatcher = TestProbe()
+      deathWatcher.watch(trackedActor)
+      trackedActor ! PoisonPill
+      deathWatcher.expectTerminated(trackedActor)
+
+      actorMetricsRecorderOf(trackedActor).get shouldNot be theSameInstanceAs (firstRecorder)
+    }
+  }
+
+  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[ActorMetrics] =
+      Kamon.metrics.register(ActorMetrics, actorRecorderName(ref)).map(_.recorder)
+
+    def collectMetricsOf(ref: ActorRef): Option[EntitySnapshot] = {
+      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/akka/DispatcherMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/akka/DispatcherMetricsSpec.scala
new file mode 100644
index 00000000..dd5cfa45
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/DispatcherMetricsSpec.scala
@@ -0,0 +1,207 @@
+/* =========================================================================================
+ * 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 akka.actor.ActorRef
+import akka.dispatch.MessageDispatcher
+import akka.testkit.TestProbe
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.metric.{ EntityRecorder, EntitySnapshot }
+import kamon.testkit.BaseKamonSpec
+
+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()
+}
+
diff --git a/kamon-akka/src/test/scala/kamon/akka/RouterMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/akka/RouterMetricsSpec.scala
new file mode 100644
index 00000000..c4c1d9ad
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/RouterMetricsSpec.scala
@@ -0,0 +1,184 @@
+/* =========================================================================================
+ * 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._
+import akka.routing._
+import akka.testkit.TestProbe
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.akka.RouterMetricsTestActor._
+import kamon.metric.EntitySnapshot
+import kamon.metric.instrument.CollectionContext
+import kamon.testkit.BaseKamonSpec
+
+import scala.concurrent.duration._
+
+class RouterMetricsSpec extends BaseKamonSpec("router-metrics-spec") {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |kamon.metric {
+        |  tick-interval = 1 hour
+        |  default-collection-context-buffer-size = 10
+        |
+        |  filters = {
+        |    akka-router {
+        |      includes = [ "user/tracked-*", "user/measuring-*", "user/stop-*" ]
+        |      excludes = [ "user/tracked-explicitly-excluded-*"]
+        |    }
+        |  }
+        |}
+        |
+        |akka.loglevel = OFF
+        |
+      """.stripMargin)
+
+  "the Kamon router metrics" should {
+    "respect the configured include and exclude filters" in new RouterMetricsFixtures {
+      createTestRouter("tracked-router")
+      createTestRouter("non-tracked-router")
+      createTestRouter("tracked-explicitly-excluded-router")
+
+      routerMetricsRecorderOf("user/tracked-router") should not be empty
+      routerMetricsRecorderOf("user/non-tracked-router") shouldBe empty
+      routerMetricsRecorderOf("user/tracked-explicitly-excluded-router") shouldBe empty
+    }
+
+    "record the routing-time of the receive function for routers" in new RouterMetricsFixtures {
+      val listener = TestProbe()
+      val router = createTestRouter("measuring-routing-time-in-router")
+
+      router.tell(Ping, listener.ref)
+      listener.expectMsg(Pong)
+      val routerSnapshot = collectMetricsOf("user/measuring-routing-time-in-router").get
+
+      routerSnapshot.histogram("routing-time").get.numberOfMeasurements should be(1L)
+    }
+
+    "record the processing-time of the receive function for routers" in new RouterMetricsFixtures {
+      val timingsListener = TestProbe()
+      val router = createTestRouter("measuring-processing-time-in-router")
+
+      router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+      val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+      val routerSnapshot = collectMetricsOf("user/measuring-processing-time-in-router").get
+
+      routerSnapshot.histogram("processing-time").get.numberOfMeasurements should be(1L)
+      routerSnapshot.histogram("processing-time").get.recordsIterator.next().count should be(1L)
+      routerSnapshot.histogram("processing-time").get.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos)
+    }
+
+    "record the number of errors for routers" in new RouterMetricsFixtures {
+      val listener = TestProbe()
+      val router = createTestRouter("measuring-errors-in-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-router").get
+      routerSnapshot.counter("errors").get.count should be(10L)
+    }
+
+    "record the time-in-mailbox for routers" in new RouterMetricsFixtures {
+      val timingsListener = TestProbe()
+      val router = createTestRouter("measuring-time-in-mailbox-in-router")
+
+      router.tell(RouterTrackTimings(sleep = Some(1 second)), timingsListener.ref)
+      val timings = timingsListener.expectMsgType[RouterTrackedTimings]
+      val routerSnapshot = collectMetricsOf("user/measuring-time-in-mailbox-in-router").get
+
+      routerSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(1L)
+      routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().count should be(1L)
+      routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos)
+    }
+
+    "clean up the associated recorder when the router is stopped" in new RouterMetricsFixtures {
+      val trackedRouter = createTestRouter("stop-in-router")
+      val firstRecorder = routerMetricsRecorderOf("user/stop-in-router").get
+
+      // Killing the router should remove it's RouterMetrics and registering again bellow should create a new one.
+      val deathWatcher = TestProbe()
+      deathWatcher.watch(trackedRouter)
+      trackedRouter ! PoisonPill
+      deathWatcher.expectTerminated(trackedRouter)
+
+      routerMetricsRecorderOf("user/stop-in-router").get shouldNot be theSameInstanceAs (firstRecorder)
+    }
+  }
+
+  trait RouterMetricsFixtures {
+    val collectionContext = new CollectionContext {
+      val buffer: LongBuffer = LongBuffer.allocate(10000)
+    }
+
+    def routerMetricsRecorderOf(routerName: String): Option[RouterMetrics] =
+      Kamon.metrics.register(RouterMetrics, routerName).map(_.recorder)
+
+    def collectMetricsOf(routerName: String): Option[EntitySnapshot] = {
+      Thread.sleep(5) // Just in case the test advances a bit faster than the actor being tested.
+      routerMetricsRecorderOf(routerName).map(_.collect(collectionContext))
+    }
+
+    def createTestRouter(routerName: String): ActorRef = {
+      val router = system.actorOf(Props[RouterMetricsTestActor].withRouter(RoundRobinRouter(nrOfInstances = 5)), 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
+    }
+  }
+}
+
+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
+  }
+}
diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala
new file mode 100644
index 00000000..593a7baa
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala
@@ -0,0 +1,87 @@
+/* ===================================================
+ * 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.instrumentation.akka
+
+import akka.actor.{ Actor, Props }
+import akka.pattern.{ ask, pipe }
+import akka.routing._
+import akka.util.Timeout
+import kamon.testkit.BaseKamonSpec
+import kamon.trace.TraceContext
+
+import scala.concurrent.duration._
+
+class ActorCellInstrumentationSpec extends BaseKamonSpec("actor-cell-instrumentation-spec") {
+  implicit lazy val executionContext = system.dispatcher
+
+  "the message passing instrumentation" should {
+    "propagate the TraceContext using bang" in new EchoActorFixture {
+      val testTraceContext = TraceContext.withContext(newContext("bang-reply")) {
+        ctxEchoActor ! "test"
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+    "propagate the TraceContext using tell" in new EchoActorFixture {
+      val testTraceContext = TraceContext.withContext(newContext("tell-reply")) {
+        ctxEchoActor.tell("test", testActor)
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+    "propagate the TraceContext using ask" in new EchoActorFixture {
+      implicit val timeout = Timeout(1 seconds)
+      val testTraceContext = TraceContext.withContext(newContext("ask-reply")) {
+        // The pipe pattern use Futures internally, so FutureTracing test should cover the underpinnings of it.
+        (ctxEchoActor ? "test") pipeTo (testActor)
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+    "propagate the TraceContext to actors behind a pool router" in new RoundRobinRouterFixture {
+      val testTraceContext = TraceContext.withContext(newContext("router-reply")) {
+        router ! "test"
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+  }
+
+  trait EchoActorFixture {
+    val ctxEchoActor = system.actorOf(Props[TraceContextEcho])
+  }
+
+  trait RoundRobinRouterFixture {
+    val router = system.actorOf(Props[TraceContextEcho].withRouter(
+      RoundRobinRouter(nrOfInstances = 5)), "pool-router")
+  }
+
+}
+
+class TraceContextEcho extends Actor {
+  def receive = {
+    case msg: String ⇒ sender ! TraceContext.currentContext
+  }
+}
+
diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala
new file mode 100644
index 00000000..143c816d
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala
@@ -0,0 +1,74 @@
+/* ===================================================
+ * 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.instrumentation.akka
+
+import akka.actor.{ Actor, ActorLogging, Props }
+import akka.event.Logging.LogEvent
+import com.typesafe.config.ConfigFactory
+import kamon.testkit.BaseKamonSpec
+import kamon.trace.TraceLocal.AvailableToMdc
+import kamon.trace.logging.MdcKeysSupport
+import kamon.trace.{ TraceContextAware, TraceLocal, TraceContext }
+import org.scalatest.Inspectors
+import org.slf4j.MDC
+
+class ActorLoggingInstrumentationSpec extends BaseKamonSpec("actor-logging-instrumentation-spec") with Inspectors with MdcKeysSupport {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |akka {
+        |  loggers = ["akka.event.slf4j.Slf4jLogger"]
+        |}
+      """.stripMargin)
+
+  "the ActorLogging instrumentation" should {
+    "attach the TraceContext (if available) to log events" in {
+      val loggerActor = system.actorOf(Props[LoggerActor])
+      system.eventStream.subscribe(testActor, classOf[LogEvent])
+
+      val testTraceContext = TraceContext.withContext(newContext("logging")) {
+        loggerActor ! "info"
+        TraceContext.currentContext
+      }
+
+      fishForMessage() {
+        case event: LogEvent if event.message.toString startsWith "TraceContext" ⇒
+          val ctxInEvent = event.asInstanceOf[TraceContextAware].traceContext
+          ctxInEvent === testTraceContext
+
+        case event: LogEvent ⇒ false
+      }
+    }
+
+    "allow retrieve a value from the MDC when was created a key of type AvailableToMdc" in {
+      val testString = "Hello World"
+      TraceContext.withContext(newContext("logging-with-mdc")) {
+        TraceLocal.store(AvailableToMdc("some-cool-key"))(testString)
+
+        withMdc {
+          MDC.get("other-key") shouldBe (null)
+          MDC.get("some-cool-key") should equal(testString)
+        }
+      }
+    }
+  }
+}
+
+class LoggerActor extends Actor with ActorLogging {
+  def receive = {
+    case "info" ⇒ log.info("TraceContext(name = {}, token = {})", TraceContext.currentContext.name, TraceContext.currentContext.token)
+  }
+}
diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala
new file mode 100644
index 00000000..cf5f1b5b
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala
@@ -0,0 +1,196 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2015 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.instrumentation.akka
+
+import akka.actor.SupervisorStrategy.{ Escalate, Restart, Resume, Stop }
+import akka.actor._
+import akka.testkit.ImplicitSender
+import com.typesafe.config.ConfigFactory
+import kamon.testkit.BaseKamonSpec
+import kamon.trace.{ EmptyTraceContext, TraceContext }
+import org.scalatest.WordSpecLike
+
+import scala.concurrent.duration._
+import scala.util.control.NonFatal
+
+class ActorSystemMessageInstrumentationSpec extends BaseKamonSpec("actor-system-message-instrumentation-spec") with WordSpecLike {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |akka {
+        |  loglevel = OFF
+        |}
+      """.stripMargin)
+
+  implicit lazy val executionContext = system.dispatcher
+
+  "the system message passing instrumentation" should {
+    "keep the TraceContext while processing the Create message in top level actors" in {
+      val testTraceContext = TraceContext.withContext(newContext("creating-top-level-actor")) {
+        system.actorOf(Props(new Actor {
+          testActor ! TraceContext.currentContext
+          def receive: Actor.Receive = { case any ⇒ }
+        }))
+
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+    "keep the TraceContext while processing the Create message in non top level actors" in {
+      val testTraceContext = TraceContext.withContext(newContext("creating-non-top-level-actor")) {
+        system.actorOf(Props(new Actor {
+          def receive: Actor.Receive = {
+            case any ⇒
+              context.actorOf(Props(new Actor {
+                testActor ! TraceContext.currentContext
+                def receive: Actor.Receive = { case any ⇒ }
+              }))
+          }
+        })) ! "any"
+
+        TraceContext.currentContext
+      }
+
+      expectMsg(testTraceContext)
+    }
+
+    "keep the TraceContext in the supervision cycle" when {
+      "the actor is resumed" in {
+        val supervisor = supervisorWithDirective(Resume)
+
+        val testTraceContext = TraceContext.withContext(newContext("fail-and-resume")) {
+          supervisor ! "fail"
+          TraceContext.currentContext
+        }
+
+        expectMsg(testTraceContext) // From the parent executing the supervision strategy
+
+        // Ensure we didn't tie the actor with the context
+        supervisor ! "context"
+        expectMsg(EmptyTraceContext)
+      }
+
+      "the actor is restarted" in {
+        val supervisor = supervisorWithDirective(Restart, sendPreRestart = true, sendPostRestart = true)
+
+        val testTraceContext = TraceContext.withContext(newContext("fail-and-restart")) {
+          supervisor ! "fail"
+          TraceContext.currentContext
+        }
+
+        expectMsg(testTraceContext) // From the parent executing the supervision strategy
+        expectMsg(testTraceContext) // From the preRestart hook
+        expectMsg(testTraceContext) // From the postRestart hook
+
+        // Ensure we didn't tie the actor with the context
+        supervisor ! "context"
+        expectMsg(EmptyTraceContext)
+      }
+
+      "the actor is stopped" in {
+        val supervisor = supervisorWithDirective(Stop, sendPostStop = true)
+
+        val testTraceContext = TraceContext.withContext(newContext("fail-and-stop")) {
+          supervisor ! "fail"
+          TraceContext.currentContext
+        }
+
+        expectMsg(testTraceContext) // From the parent executing the supervision strategy
+        expectMsg(testTraceContext) // From the postStop hook
+        expectNoMsg(1 second)
+      }
+
+      "the failure is escalated" in {
+        val supervisor = supervisorWithDirective(Escalate, sendPostStop = true)
+
+        val testTraceContext = TraceContext.withContext(newContext("fail-and-escalate")) {
+          supervisor ! "fail"
+          TraceContext.currentContext
+        }
+
+        expectMsg(testTraceContext) // From the parent executing the supervision strategy
+        expectMsg(testTraceContext) // From the grandparent executing the supervision strategy
+        expectMsg(testTraceContext) // From the postStop hook in the child
+        expectMsg(testTraceContext) // From the postStop hook in the parent
+        expectNoMsg(1 second)
+      }
+    }
+  }
+
+  def supervisorWithDirective(directive: SupervisorStrategy.Directive, sendPreRestart: Boolean = false, sendPostRestart: Boolean = false,
+    sendPostStop: Boolean = false, sendPreStart: Boolean = false): ActorRef = {
+    class GrandParent extends Actor {
+      val child = context.actorOf(Props(new Parent))
+
+      override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() {
+        case NonFatal(throwable) ⇒ testActor ! TraceContext.currentContext; Stop
+      }
+
+      def receive = {
+        case any ⇒ child forward any
+      }
+    }
+
+    class Parent extends Actor {
+      val child = context.actorOf(Props(new Child))
+
+      override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() {
+        case NonFatal(throwable) ⇒ testActor ! TraceContext.currentContext; directive
+      }
+
+      def receive: Actor.Receive = {
+        case any ⇒ child forward any
+      }
+
+      override def postStop(): Unit = {
+        if (sendPostStop) testActor ! TraceContext.currentContext
+        super.postStop()
+      }
+    }
+
+    class Child extends Actor {
+      def receive = {
+        case "fail"    ⇒ throw new ArithmeticException("Division by zero.")
+        case "context" ⇒ sender ! TraceContext.currentContext
+      }
+
+      override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
+        if (sendPreRestart) testActor ! TraceContext.currentContext
+        super.preRestart(reason, message)
+      }
+
+      override def postRestart(reason: Throwable): Unit = {
+        if (sendPostRestart) testActor ! TraceContext.currentContext
+        super.postRestart(reason)
+      }
+
+      override def postStop(): Unit = {
+        if (sendPostStop) testActor ! TraceContext.currentContext
+        super.postStop()
+      }
+
+      override def preStart(): Unit = {
+        if (sendPreStart) testActor ! TraceContext.currentContext
+        super.preStart()
+      }
+    }
+
+    system.actorOf(Props(new GrandParent))
+  }
+}
diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala
new file mode 100644
index 00000000..d925fbf6
--- /dev/null
+++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala
@@ -0,0 +1,137 @@
+/*
+ * =========================================================================================
+ * 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.instrumentation.akka
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import akka.actor._
+import akka.event.Logging.Warning
+import akka.pattern.ask
+import akka.testkit.TestProbe
+import akka.util.Timeout
+import com.typesafe.config.ConfigFactory
+import kamon.Kamon
+import kamon.akka.Akka
+import kamon.testkit.BaseKamonSpec
+import kamon.trace.{ TraceContext, TraceContextAware }
+
+import scala.concurrent.duration._
+
+class AskPatternInstrumentationSpec extends BaseKamonSpec("ask-pattern-tracing-spec") {
+  override lazy val config =
+    ConfigFactory.parseString(
+      """
+        |akka {
+        |  loglevel = OFF
+        |}
+      """.stripMargin)
+
+  implicit lazy val ec = system.dispatcher
+  implicit val askTimeout = Timeout(10 millis)
+
+  // TODO: Make this work with ActorSelections
+
+  "the AskPatternInstrumentation" when {
+    "configured in heavyweight mode" should {
+      "log a warning with a full stack trace and the TraceContext taken from the moment the ask was triggered for a actor" in new NoReplyFixture {
+        setAskPatternTimeoutWarningMode("heavyweight")
+
+        expectTimeoutWarning() {
+          TraceContext.withContext(newContext("ask-timeout-warning")) {
+            noReplyActorRef ? "hello"
+            TraceContext.currentContext
+          }
+        }
+      }
+    }
+
+    "configured in lightweight mode" should {
+      "log a warning with a short source location description and the TraceContext taken from the moment the ask was triggered for a actor" in new NoReplyFixture {
+        setAskPatternTimeoutWarningMode("lightweight")
+
+        expectTimeoutWarning(messageSizeLimit = Some(1)) {
+          TraceContext.withContext(newContext("ask-timeout-warning")) {
+            noReplyActorRef ? "hello"
+            TraceContext.currentContext
+          }
+        }
+      }
+    }
+
+    "configured in off mode" should {
+      "should not log any warning messages" in new NoReplyFixture {
+        setAskPatternTimeoutWarningMode("off")
+
+        expectTimeoutWarning(expectWarning = false) {
+          TraceContext.withContext(newContext("ask-timeout-warning")) {
+            noReplyActorRef ? "hello"
+            TraceContext.currentContext
+          }
+        }
+      }
+    }
+  }
+
+  def expectTimeoutWarning(messageSizeLimit: Option[Int] = None, expectWarning: Boolean = true)(thunk: ⇒ TraceContext): Unit = {
+    val listener = warningListener()
+    val testTraceContext = thunk
+
+    if (expectWarning) {
+      val warning = listener.fishForMessage() {
+        case Warning(_, _, msg) if msg.toString.startsWith("Timeout triggered for ask pattern registered at") ⇒ true
+        case others ⇒ false
+      }.asInstanceOf[Warning]
+
+      warning.asInstanceOf[TraceContextAware].traceContext should equal(testTraceContext)
+      messageSizeLimit.map { messageLimit ⇒
+        warning.message.toString.lines.size should be(messageLimit)
+      }
+    } else {
+      listener.expectNoMsg()
+    }
+  }
+
+  def warningListener(): TestProbe = {
+    val listener = TestProbe()
+    system.eventStream.subscribe(listener.ref, classOf[Warning])
+    listener
+  }
+
+  def setAskPatternTimeoutWarningMode(mode: String): Unit = {
+    val target = Kamon(Akka)
+    val field = target.getClass.getDeclaredField("askPatternTimeoutWarning")
+    field.setAccessible(true)
+    field.set(target, mode)
+  }
+
+  val fixtureCounter = new AtomicInteger(0)
+
+  trait NoReplyFixture {
+    def noReplyActorRef: ActorRef = system.actorOf(Props[NoReply], "no-reply-" + fixtureCounter.incrementAndGet())
+
+    def noReplyActorSelection: ActorSelection = {
+      val target = noReplyActorRef
+      system.actorSelection(target.path)
+    }
+  }
+}
+
+class NoReply extends Actor {
+  def receive = {
+    case any ⇒
+  }
+}