! akka: prefix actors, routers and dispatchers with the actor system name.

2 files changed, 25 insertions, 15 deletions

diff --git a/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala
index d1d420fe..39bdcfe0 100644
--- a/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala
+++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala
@@ -34,7 +34,7 @@ class ActorCellInstrumentation {
 
   @After("actorCellCreation(cell, system, ref, props, dispatcher, parent)")
   def afterCreation(cell: ActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = {
-    val actorEntity = Entity(ref.path.elements.mkString("/"), ActorMetrics.category)
+    val actorEntity = Entity(system.name + "/" + ref.path.elements.mkString("/"), ActorMetrics.category)
 
     if (Kamon.metrics.shouldTrack(actorEntity)) {
       val actorMetricsRecorder = Kamon.metrics.entity(ActorMetrics, actorEntity)
@@ -60,19 +60,19 @@ class ActorCellInstrumentation {
         pjp.proceed()
       }
     } finally {
-      cellMetrics.recorder.map { am ⇒
-        val processingTime = System.nanoTime() - timestampBeforeProcessing
-        val timeInMailbox = timestampBeforeProcessing - contextAndTimestamp.captureNanoTime
+      val processingTime = System.nanoTime() - timestampBeforeProcessing
+      val timeInMailbox = timestampBeforeProcessing - contextAndTimestamp.captureNanoTime
 
+      cellMetrics.recorder.map { am ⇒
         am.processingTime.record(processingTime)
         am.timeInMailbox.record(timeInMailbox)
         am.mailboxSize.decrement()
+      }
 
-        // In case that this actor is behind a router, record the metrics for the router.
-        envelope.asInstanceOf[RouterAwareEnvelope].routerMetricsRecorder.map { rm ⇒
-          rm.processingTime.record(processingTime)
-          rm.timeInMailbox.record(timeInMailbox)
-        }
+      // In case that this actor is behind a router, record the metrics for the router.
+      envelope.asInstanceOf[RouterAwareEnvelope].routerMetricsRecorder.map { rm ⇒
+        rm.processingTime.record(processingTime)
+        rm.timeInMailbox.record(timeInMailbox)
       }
     }
   }
@@ -127,7 +127,7 @@ class RoutedActorCellInstrumentation {
 
   @After("routedActorCellCreation(cell, system, ref, props, dispatcher, routeeProps, supervisor)")
   def afterRoutedActorCellCreation(cell: RoutedActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, routeeProps: Props, supervisor: ActorRef): Unit = {
-    val routerEntity = Entity(ref.path.elements.mkString("/"), RouterMetrics.category)
+    val routerEntity = Entity(system.name + "/" + ref.path.elements.mkString("/"), RouterMetrics.category)
 
     if (Kamon.metrics.shouldTrack(routerEntity)) {
       val cellMetrics = cell.asInstanceOf[RoutedActorCellMetrics]
diff --git a/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala
index 4567d442..931d8a45 100644
--- a/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala
+++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala
@@ -59,16 +59,16 @@ class DispatcherInstrumentation {
   private def registerDispatcher(dispatcherName: String, executorService: ExecutorService, system: ActorSystem): Unit =
     executorService match {
       case fjp: AkkaForkJoinPool ⇒
-        val dispatcherEntity = Entity(dispatcherName, AkkaDispatcherMetrics.Category)
+        val dispatcherEntity = Entity(system.name + "/" + dispatcherName, AkkaDispatcherMetrics.Category, tags = Map("dispatcher-type" -> "fork-join-pool"))
 
         if (Kamon.metrics.shouldTrack(dispatcherEntity))
-          Kamon.metrics.entity(ForkJoinPoolDispatcherMetrics.factory(fjp), dispatcherName, Map("dispatcher-type" -> "fork-join-pool"))
+          Kamon.metrics.entity(ForkJoinPoolDispatcherMetrics.factory(fjp), dispatcherEntity)
 
       case tpe: ThreadPoolExecutor ⇒
-        val dispatcherEntity = Entity(dispatcherName, AkkaDispatcherMetrics.Category)
+        val dispatcherEntity = Entity(system.name + "/" + dispatcherName, AkkaDispatcherMetrics.Category, tags = Map("dispatcher-type" -> "thread-pool-executor"))
 
         if (Kamon.metrics.shouldTrack(dispatcherEntity))
-          Kamon.metrics.entity(ThreadPoolExecutorDispatcherMetrics.factory(tpe), dispatcherName, Map("dispatcher-type" -> "thread-pool-executor"))
+          Kamon.metrics.entity(ThreadPoolExecutorDispatcherMetrics.factory(tpe), dispatcherEntity)
 
       case others ⇒ // Currently not interested in other kinds of dispatchers.
     }
@@ -126,7 +126,17 @@ class DispatcherInstrumentation {
     import lazyExecutor.lookupData
 
     if (lookupData.actorSystem != null)
-      Kamon.metrics.removeEntity(Entity(lookupData.dispatcherName, AkkaDispatcherMetrics.Category))
+      lazyExecutor.asInstanceOf[ExecutorServiceDelegate].executor match {
+        case fjp: AkkaForkJoinPool ⇒
+          Kamon.metrics.removeEntity(Entity(lookupData.actorSystem.name + "/" + lookupData.dispatcherName,
+            AkkaDispatcherMetrics.Category, tags = Map("dispatcher-type" -> "fork-join-pool")))
+
+        case tpe: ThreadPoolExecutor ⇒
+          Kamon.metrics.removeEntity(Entity(lookupData.actorSystem.name + "/" + lookupData.dispatcherName,
+            AkkaDispatcherMetrics.Category, tags = Map("dispatcher-type" -> "thread-pool-executor")))
+
+        case other ⇒ // nothing to remove.
+      }
   }
 
 }