feat: metric for dropped messages in Shard Region buffer

Author: sebastian-alfers

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

@@ -1296,6 +1296,7 @@ private[akka] class ShardRegion(
         loggedFullBufferWarning = true
       }
       context.system.deadLetters ! msg
+      instrumentation.dropMessagesShardRegion(cluster.selfAddress, self, typeName)
     } else {
       shardBuffers.append(shardId, msg, snd)
       instrumentation.incrementShardRegionBufferSize(cluster.selfAddress, self, typeName)

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

@@ -94,6 +94,9 @@ class ClusterShardingTelemetryEnsemble(val instrumentations: Seq[ClusterSharding
       typeName: String): Unit =
     instrumentations.foreach(_.incrementShardRegionBufferSize(selfAddress, shardRegionActor, typeName))
 
+  override def dropMessagesShardRegion(selfAddress: Address, self: ActorRef, typeName: String): Unit =
+    instrumentations.foreach(_.dropMessagesShardRegion(selfAddress, self, typeName))
+
   override def dependencies: immutable.Seq[String] =
     instrumentations.flatMap(_.dependencies)
 }
@@ -121,6 +124,8 @@ class EmptyClusterShardingInstrumentation extends ClusterShardingInstrumentation
       shardRegionActor: ActorRef,
       typeName: String): Unit = ()
 
+  override def dropMessagesShardRegion(selfAddress: Address, self: ActorRef, typeName: String): Unit = ()
+
   override def dependencies: immutable.Seq[String] = Nil
 }
 
@@ -140,6 +145,11 @@ trait ClusterShardingInstrumentation {
    */
   def incrementShardRegionBufferSize(selfAddress: Address, shardRegionActor: ActorRef, typeName: String): Unit
 
+  /**
+   * Drop a mesage send to a Shard Region as the buffer is full
+   */
+  def dropMessagesShardRegion(selfAddress: Address, self: ActorRef, typeName: String): Unit
+
   /**
    * Optional dependencies for this instrumentation.
    *

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

@@ -51,6 +51,7 @@ class SpecClusterShardingTelemetry(@nowarn("msg=never used") system: ExtendedAct
     extends ClusterShardingInstrumentation {
 
   val counter = new AtomicInteger(0)
+  val dropMessageCounter = new AtomicInteger(0)
 
   override def shardRegionBufferSize(
       selfAddress: Address,
@@ -67,6 +68,9 @@ class SpecClusterShardingTelemetry(@nowarn("msg=never used") system: ExtendedAct
     counter.incrementAndGet()
   }
 
+  override def dropMessagesShardRegion(selfAddress: Address, self: ActorRef, typeName: String): Unit =
+    dropMessageCounter.incrementAndGet()
+
   override def dependencies: Seq[String] = Nil
 }
 
@@ -113,6 +117,9 @@ abstract class ClusterShardingInstrumentationSpec
   private val counter = ClusterShardingInstrumentationProvider(system).instrumentation
     .asInstanceOf[SpecClusterShardingTelemetry]
     .counter
+  private val dropMessageCounter = ClusterShardingInstrumentationProvider(system).instrumentation
+    .asInstanceOf[SpecClusterShardingTelemetry]
+    .dropMessageCounter
 
   override implicit val patienceConfig: PatienceConfig = {
     import akka.testkit.TestDuration
@@ -189,8 +196,9 @@ abstract class ClusterShardingInstrumentationSpec
         }
         eventually {
           // we have 100 in the buffer, and our cap is 120 (per config in this test)
-          // 10 are dropped. Should we have a metric on this? Or custom events?
+          // 10 are dropped.
           counter.get() shouldBe 120
+          dropMessageCounter.get() shouldBe 10
         }
       }
       enterBarrier("buffer-overflow")