rename spec, add local address

Author: sebastian-alfers

akka-cluster-sharding/src/main/scala/akka/cluster/sharding/ShardRegion.scala

@@ -690,7 +690,7 @@ private[akka] class ShardRegion(
     timers.startTimerWithFixedDelay(Retry, Retry, retryInterval)
     startRegistration()
     logPassivationStrategy()
-    instrumentation.shardRegionBufferSize(typeName, 0)
+    instrumentation.shardRegionBufferSize(cluster.selfAddress, typeName, 0)
   }
 
   override def postStop(): Unit = {
@@ -699,7 +699,7 @@ private[akka] class ShardRegion(
     coordinator.foreach(_ ! RegionStopped(context.self))
     cluster.unsubscribe(self)
     gracefulShutdownProgress.trySuccess(Done)
-    instrumentation.shardRegionBufferSize(typeName, 0)
+    instrumentation.shardRegionBufferSize(cluster.selfAddress, typeName, 0)
   }
 
   private def logPassivationStrategy(): Unit = {
@@ -754,7 +754,7 @@ private[akka] class ShardRegion(
     coordinator.map { coordRef =>
       val a = coordRef.path.address
 
-      if (a.hasLocalScope) cluster.selfMember.address
+      if (a.hasLocalScope) cluster.selfAddress
       else a
     }
 
@@ -954,7 +954,7 @@ private[akka] class ShardRegion(
             dropped,
             shard)
           // better to decrease by "dropped" to avoid calculating the size?
-          instrumentation.shardRegionBufferSize(typeName, shardBuffers.totalSize)
+          instrumentation.shardRegionBufferSize(cluster.selfAddress, typeName, shardBuffers.totalSize)
         }
         loggedFullBufferWarning = false
       }
@@ -1298,7 +1298,7 @@ private[akka] class ShardRegion(
       context.system.deadLetters ! msg
     } else {
       shardBuffers.append(shardId, msg, snd)
-      instrumentation.incrementShardRegionBufferSize(typeName)
+      instrumentation.incrementShardRegionBufferSize(cluster.selfAddress, typeName)
       // log some insight to how buffers are filled up every 10% of the buffer capacity
       val tot = totBufSize + 1
       if (tot % (bufferSize / 10) == 0) {
@@ -1331,7 +1331,7 @@ private[akka] class ShardRegion(
       }
 
       shardBuffers.remove(shardId)
-      instrumentation.shardRegionBufferSize(typeName, shardBuffers.totalSize)
+      instrumentation.shardRegionBufferSize(cluster.selfAddress, typeName, shardBuffers.totalSize)
     }
     loggedFullBufferWarning = false
     retryCount = 0
@@ -1374,7 +1374,7 @@ private[akka] class ShardRegion(
               shardId,
               buf.size + 1)
             shardBuffers.append(shardId, msg, snd)
-            instrumentation.incrementShardRegionBufferSize(typeName)
+            instrumentation.incrementShardRegionBufferSize(cluster.selfAddress, typeName)
         }
 
       case _ =>

akka-cluster-sharding/src/main/scala/akka/cluster/sharding/internal/ClusterShardingInstrumentation.scala

@@ -6,17 +6,16 @@ package akka.cluster.sharding.internal
 
 import akka.actor.ExtendedActorSystem
 import akka.annotation.InternalStableApi
-
 import scala.collection.immutable
 import scala.jdk.CollectionConverters._
 
 import akka.actor.ActorSystem
+import akka.actor.Address
 import akka.actor.ClassicActorSystemProvider
 import akka.actor.Extension
 import akka.actor.ExtensionId
 import akka.actor.ExtensionIdProvider
 import akka.event.Logging
-
 import akka.util.TopologicalSort.topologicalSort
 
 /**
@@ -81,11 +80,11 @@ class ClusterShardingInstrumentationProvider(system: ExtendedActorSystem) extend
 class ClusterShardingTelemetryEnsemble(val instrumentations: Seq[ClusterShardingInstrumentation])
     extends ClusterShardingInstrumentation {
 
-  override def shardRegionBufferSize(typeName: String, size: Int): Unit =
-    instrumentations.foreach(_.shardRegionBufferSize(typeName, size))
+  override def shardRegionBufferSize(selfAddress: Address, typeName: String, size: Int): Unit =
+    instrumentations.foreach(_.shardRegionBufferSize(selfAddress, typeName, size))
 
-  override def incrementShardRegionBufferSize(typeName: String): Unit =
-    instrumentations.foreach(_.incrementShardRegionBufferSize(typeName))
+  override def incrementShardRegionBufferSize(selfAddress: Address, typeName: String): Unit =
+    instrumentations.foreach(_.incrementShardRegionBufferSize(selfAddress, typeName))
 
   override def dependencies: immutable.Seq[String] =
     instrumentations.flatMap(_.dependencies)
@@ -103,9 +102,9 @@ object EmptyClusterShardingInstrumentation extends EmptyClusterShardingInstrumen
 @InternalStableApi
 class EmptyClusterShardingInstrumentation extends ClusterShardingInstrumentation {
 
-  override def shardRegionBufferSize(typeName: String, size: Int): Unit = ()
+  override def shardRegionBufferSize(selfAddress: Address, typeName: String, size: Int): Unit = ()
 
-  override def incrementShardRegionBufferSize(typeName: String): Unit = ()
+  override def incrementShardRegionBufferSize(selfAddress: Address, typeName: String): Unit = ()
 
   override def dependencies: immutable.Seq[String] = Nil
 }
@@ -119,12 +118,12 @@ trait ClusterShardingInstrumentation {
   /**
    * @param size set current size of the buffer.
    */
-  def shardRegionBufferSize(typeName: String, size: Int): Unit
+  def shardRegionBufferSize(selfAddress: Address, typeName: String, size: Int): Unit
 
   /**
    * Increase the current size of the buffer by one.
    */
-  def incrementShardRegionBufferSize(typeName: String): Unit
+  def incrementShardRegionBufferSize(selfAddress: Address, typeName: String): Unit
 
   /**
    * Optional dependencies for this instrumentation.

akka-cluster-sharding/src/multi-jvm/scala/akka/cluster/sharding/ClusterShardingInstrumentationSpec.scala

@@ -15,14 +15,14 @@ import akka.actor.ExtendedActorSystem
 import akka.actor.{ Actor, ActorLogging, Address, Props }
 import akka.cluster.Cluster
 import akka.cluster.MemberStatus
-import akka.cluster.sharding.ClusterShardInstrumentatioSpec.GiveMeYourHome.{ Get, Home }
+import akka.cluster.sharding.ClusterShardingInstrumentationSpec.GiveMeYourHome.{ Get, Home }
 import akka.cluster.sharding.internal.{ ClusterShardingInstrumentation, ClusterShardingInstrumentationProvider }
 import akka.remote.testkit.Direction
 import akka.testkit.TestProbe
 import akka.serialization.jackson.CborSerializable
 import akka.testkit.ImplicitSender
 
-object ClusterShardInstrumentatioSpecConfig
+object ClusterShardingInstrumentationSpecConfig
     extends MultiNodeClusterShardingConfig(
       //loglevel = "DEBUG",
       additionalConfig = """
@@ -39,27 +39,27 @@ object ClusterShardInstrumentatioSpecConfig
 
 }
 
-class ClusterShardInstrumentatioSpecMultiJvmNode1 extends ClusterShardInstrumentatioSpec
+class ClusterShardingInstrumentationSpecMultiJvmNode1 extends ClusterShardingInstrumentationSpec
 
-class ClusterShardInstrumentatioSpecMultiJvmNode2 extends ClusterShardInstrumentatioSpec
+class ClusterShardingInstrumentationSpecMultiJvmNode2 extends ClusterShardingInstrumentationSpec
 
 class SpecClusterShardingTelemetry(@nowarn("msg=never used") system: ExtendedActorSystem)
     extends ClusterShardingInstrumentation {
 
   val counter = new AtomicInteger(0)
 
-  override def shardRegionBufferSize(typeName: String, size: Int): Unit = {
+  override def shardRegionBufferSize(selfAddress: Address, typeName: String, size: Int): Unit = {
     counter.set(size)
   }
 
-  override def incrementShardRegionBufferSize(typeName: String): Unit = {
+  override def incrementShardRegionBufferSize(selfAddress: Address, typeName: String): Unit = {
     counter.incrementAndGet()
   }
 
   override def dependencies: Seq[String] = Nil
 }
 
-object ClusterShardInstrumentatioSpec {
+object ClusterShardingInstrumentationSpec {
 
   object GiveMeYourHome {
     case class Get(id: String) extends CborSerializable
@@ -90,14 +90,14 @@ object ClusterShardInstrumentatioSpec {
   }
 }
 
-abstract class ClusterShardInstrumentatioSpec
-    extends MultiNodeClusterShardingSpec(ClusterShardInstrumentatioSpecConfig)
+abstract class ClusterShardingInstrumentationSpec
+    extends MultiNodeClusterShardingSpec(ClusterShardingInstrumentationSpecConfig)
     with ImplicitSender
     with ScalaFutures {
 
-  import ClusterShardInstrumentatioSpec._
-  import ClusterShardInstrumentatioSpec.GiveMeYourHome._
-  import ClusterShardInstrumentatioSpecConfig._
+  import ClusterShardingInstrumentationSpec._
+  import ClusterShardingInstrumentationSpec.GiveMeYourHome._
+  import ClusterShardingInstrumentationSpecConfig._
 
   private val counter = ClusterShardingInstrumentationProvider(system).instrumentation
     .asInstanceOf[SpecClusterShardingTelemetry]