add basic support for typed actors (#136)

Author: pjfanning

build.sbt

@@ -36,6 +36,7 @@ libraryDependencies ++= Seq(
   "org.apache.pekko" %% "pekko-slf4j" % pekkoVersion,
   "com.typesafe" % "config" % "1.4.3",
   "org.aspectj" % "aspectjweaver" % aspectjweaverVersion,
+  "org.apache.pekko" %% "pekko-actor-typed" % pekkoVersion % Test,
   "org.apache.pekko" %% "pekko-testkit" % pekkoVersion % Test,
   "org.scalatest" %% "scalatest" % "3.2.19" % Test,
   "ch.qos.logback" % "logback-classic" % "1.3.15" % Test

src/main/scala/com/github/pjfanning/micrometer/pekko/ActorMetrics.scala

@@ -35,6 +35,9 @@ object ActorMetrics {
     }
   }
   def hasMetricsFor(e: Entity): Boolean = map.contains(e)
+
+  private[pekko] def getMap(): Map[Entity, ActorMetrics] = map.toMap
+  private[pekko] def clear(): Unit = map.clear()
 }
 
 class ActorMetrics(entity: Entity) {

src/main/scala/org/apache/pekko/monitor/instrumentation/CellInfo.scala

@@ -17,7 +17,6 @@
 package org.apache.pekko.monitor.instrumentation
 
 import com.github.pjfanning.micrometer.pekko.{Entity, MetricsConfig}
-
 import org.apache.pekko.actor.{ActorRef, ActorSystem, Cell}
 import org.apache.pekko.routing.{NoRouter, RoutedActorRef}
 
@@ -26,14 +25,21 @@ case class CellInfo(entity: Entity, actorSystemName: String, isRouter: Boolean,
 
 object CellInfo {
 
+  // use String instead of Class[_] to avoid requiring a runtime dependency on the pekko-actor-typed jar
+  val TypedActorAdapterClassName = "org.apache.pekko.actor.typed.internal.adapter.ActorAdapter"
+
   def cellName(system: ActorSystem, ref: ActorRef): String =
     s"""${system.name}/${ref.path.elements.mkString("/")}"""
 
   def cellInfoFor(cell: Cell, system: ActorSystem, ref: ActorRef, parent: ActorRef, actorCellCreation: Boolean): CellInfo = {
     def hasRouterProps(cell: Cell): Boolean = cell.props.deploy.routerConfig != NoRouter
 
     val pathString = ref.path.elements.mkString("/")
-    val isRootSupervisor = pathString.isEmpty || pathString == "user" || pathString == "system"
+    val isTyped = cell.props.actorClass().getName == TypedActorAdapterClassName
+    val isRootSupervisor = if (isTyped)
+      pathString != "user"
+    else
+      pathString.isEmpty || pathString == "user" || pathString == "system"
     val isRouter = hasRouterProps(cell)
     val isRoutee = parent.isInstanceOf[RoutedActorRef]
 

src/test/resources/application.conf

@@ -7,7 +7,7 @@ pekko {
 micrometer.pekko {
   metric.filters {
     pekko-actor {
-      includes = [ "**/user/tracked-**", "*/user/measuring-**", "*/user/stop-**" ]
+      includes = [ "**/user/tracked-**", "*/user/measuring-**", "*/user/stop-**", "greet-service/**" ]
       excludes = [ "*/system/**", "*/user/IO-**", "**/user/tracked-explicitly-excluded-**" ]
     }
 

src/test/scala/com/github/pjfanning/micrometer/pekko/TypedActor.scala

@@ -0,0 +1,15 @@
+package com.github.pjfanning.micrometer.pekko
+
+import org.apache.pekko.actor.typed.{ActorRef, Behavior}
+import org.apache.pekko.actor.typed.scaladsl.Behaviors
+
+object TypedActor {
+  final case class Greet(whom: String, replyTo: ActorRef[Greeted])
+  final case class Greeted(whom: String, from: ActorRef[Greet])
+
+  def apply(): Behavior[Greet] = Behaviors.receive { (context, message) =>
+    context.log.info("Hello {}!", message.whom)
+    message.replyTo ! Greeted(message.whom, context.self)
+    Behaviors.same
+  }
+}

src/test/scala/com/github/pjfanning/micrometer/pekko/TypedActorMetricsSpec.scala

@@ -0,0 +1,36 @@
+package com.github.pjfanning.micrometer.pekko
+
+import com.github.pjfanning.micrometer.pekko.TypedActor.{Greet, Greeted}
+import org.apache.pekko.actor.typed.ActorSystem
+import org.apache.pekko.actor.typed.scaladsl.AskPattern.Askable
+import org.apache.pekko.util.Timeout
+
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.Await
+
+class TypedActorMetricsSpec extends BaseSpec {
+
+  private val system: ActorSystem[Greet] = ActorSystem(TypedActor(), "greet-service")
+
+  override def afterAll(): Unit = {
+    super.afterAll()
+    system.terminate()
+    ActorMetrics.clear()
+  }
+
+  "the typed actor metrics" should {
+    "respect the configured include and exclude filters" in {
+      sendMessage("world")
+      val map = ActorMetrics.getMap()
+      map should not be empty
+      val entity = Entity("greet-service/user", MetricsConfig.Actor)
+      val metrics = map(entity)
+      metrics.messages.count() shouldEqual 2.0
+    }
+  }
+
+  def sendMessage(name: String): Unit = {
+    val fut = system.ask[Greeted](Greet(name, _))(Timeout(10.seconds), system.scheduler)
+    Await.result(fut, 10.seconds)
+  }
+}

src/test/scala/org/apache/pekko/actor/typed/instrumentation/ClassSpec.scala

@@ -0,0 +1,14 @@
+package org.apache.pekko.actor.typed.instrumentation
+
+import com.github.pjfanning.micrometer.pekko.BaseSpec
+import org.apache.pekko.actor.typed.internal.adapter.ActorAdapter
+import org.apache.pekko.monitor.instrumentation.CellInfo
+
+class ClassSpec extends BaseSpec {
+
+  "Classloading" should {
+    "match TypedActorAdapterClassName" in {
+      Class.forName(CellInfo.TypedActorAdapterClassName) shouldEqual classOf[ActorAdapter[_]]
+    }
+  }
+}