feat: Cluster Sharding telemetry #32878
+573
−10
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,217 @@ | ||
| /* | ||
| * Copyright (C) 2025 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.cluster.sharding.internal | ||
|
|
||
| import akka.actor.ExtendedActorSystem | ||
| import akka.annotation.InternalStableApi | ||
| import scala.collection.immutable | ||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| import akka.actor.ActorRef | ||
| 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 | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalStableApi | ||
| object ClusterShardingInstrumentationProvider | ||
| extends ExtensionId[ClusterShardingInstrumentationProvider] | ||
| with ExtensionIdProvider { | ||
| override def get(system: ActorSystem): ClusterShardingInstrumentationProvider = super.get(system) | ||
|
|
||
| override def get(system: ClassicActorSystemProvider): ClusterShardingInstrumentationProvider = super.get(system) | ||
|
|
||
| override def lookup = ClusterShardingInstrumentationProvider | ||
|
|
||
| override def createExtension(system: ExtendedActorSystem): ClusterShardingInstrumentationProvider = | ||
| new ClusterShardingInstrumentationProvider(system) | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalStableApi | ||
| class ClusterShardingInstrumentationProvider(system: ExtendedActorSystem) extends Extension { | ||
| private val fqcnConfigPath = "akka.cluster.sharding.telemetry.instrumentations" | ||
|
|
||
| lazy val instrumentation: ClusterShardingInstrumentation = { | ||
| if (!system.settings.config.hasPath(fqcnConfigPath)) { | ||
| EmptyClusterShardingInstrumentation | ||
| } else { | ||
| val fqcns = system.settings.config.getStringList(fqcnConfigPath).asScala.toVector | ||
| fqcns.size match { | ||
| case 0 => EmptyClusterShardingInstrumentation | ||
| case 1 => create(fqcns.head) | ||
| case _ => | ||
| val instrumentationsByFqcn = fqcns.iterator.map(fqcn => fqcn -> create(fqcn)).toMap | ||
| val sortedNames = topologicalSort[String](fqcns, fqcn => instrumentationsByFqcn(fqcn).dependencies.toSet) | ||
| val instrumentations = sortedNames.map(instrumentationsByFqcn).toVector | ||
| new ClusterShardingTelemetryEnsemble(instrumentations) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def create(fqcn: String): ClusterShardingInstrumentation = { | ||
| try { | ||
| system.dynamicAccess | ||
| .createInstanceFor[ClusterShardingInstrumentation](fqcn, immutable.Seq(classOf[ExtendedActorSystem] -> system)) | ||
| .get | ||
| } catch { | ||
| case t: Throwable => // Throwable, because instrumentation failure should not cause fatal shutdown | ||
| Logging(system.classicSystem, classOf[ClusterShardingInstrumentationProvider]) | ||
| .warning(t, "Cannot create instrumentation [{}]", fqcn) | ||
| EmptyClusterShardingInstrumentation | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalStableApi | ||
| class ClusterShardingTelemetryEnsemble(val instrumentations: Seq[ClusterShardingInstrumentation]) | ||
| extends ClusterShardingInstrumentation { | ||
|
|
||
| override def shardRegionBufferSize( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
|
Contributor
There was a problem hiding this comment. do we need the selfAddress and shardRegionActor? is that because Cinnamon has that existing structure?
Contributor
Author
There was a problem hiding this comment. Yes, in the current version both are used. I let @pvlugter share his thoughts.
Member
There was a problem hiding this comment. I didn't work on cluster sharding instrumentation originally, but if you're integrating with what's there already, these will be for the identity and for accessing metadata. But you can also have this new telemetry be completely separate. You'll mostly just want the entity type for a metric label.
Contributor
There was a problem hiding this comment. To me it feels like Cinnamon should already know the ActorSystem, and thereby the address, and I don't see why this metric should be coupled to shardRegionActor. The address + typeName should be enough to create a unique key. However, if that is needed because it makes it easier on the Cinnamon side, then so be it.
Member
There was a problem hiding this comment. Also not sure why self address is being passed and why the actor ref is being used. I see that the address is already accessed automatically from the actor system for some cluster instrumentation, and agree that the entity type name is what should be used for identifying.
Contributor
Author
There was a problem hiding this comment. Ok, then I remove it again (I thought somewhere Cinnamon needed it).
Member
There was a problem hiding this comment. You'll likely need it if you're cross-integrating this instrumentation with the existing sharding instrumentation. Otherwise you can define this SPI cleanly, which I think is preferable. |
||
| typeName: String, | ||
| size: Int): Unit = | ||
| instrumentations.foreach(_.shardRegionBufferSize(selfAddress, shardRegionActor, typeName, size)) | ||
|
|
||
| override def shardRegionBufferSizeIncremented( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String): Unit = | ||
| instrumentations.foreach(_.shardRegionBufferSizeIncremented(selfAddress, shardRegionActor, typeName)) | ||
|
|
||
| override def regionRequestedShardHome( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shardId: String): Unit = | ||
| instrumentations.foreach(_.regionRequestedShardHome(selfAddress, shardRegionActor, typeName, shardId)) | ||
|
|
||
| override def receivedShardHome( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shardId: String): Unit = | ||
| instrumentations.foreach(_.receivedShardHome(selfAddress, shardRegionActor, typeName, shardId)) | ||
|
|
||
| override def shardHandoffStarted( | ||
| selfAddress: Address, | ||
| shardCoordinatorActor: ActorRef, | ||
| typeName: String, | ||
| shard: String): Unit = | ||
| instrumentations.foreach(_.shardHandoffStarted(selfAddress, shardCoordinatorActor, typeName, shard)) | ||
|
|
||
| override def shardHandoffFinished( | ||
| selfAddress: Address, | ||
| shardCoordinatorActor: ActorRef, | ||
| typeName: String, | ||
| shard: String, | ||
| ok: Boolean): Unit = | ||
| instrumentations.foreach(_.shardHandoffFinished(selfAddress, shardCoordinatorActor, typeName, shard, ok)) | ||
|
|
||
| override def dependencies: immutable.Seq[String] = | ||
| instrumentations.flatMap(_.dependencies) | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalStableApi | ||
| object EmptyClusterShardingInstrumentation extends EmptyClusterShardingInstrumentation | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalStableApi | ||
| class EmptyClusterShardingInstrumentation extends ClusterShardingInstrumentation { | ||
|
|
||
| override def shardRegionBufferSize( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| size: Int): Unit = () | ||
|
|
||
| override def shardRegionBufferSizeIncremented( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String): Unit = () | ||
|
|
||
| override def regionRequestedShardHome( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shardId: String): Unit = () | ||
|
|
||
| override def receivedShardHome( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shardId: String): Unit = () | ||
|
|
||
| override def shardHandoffStarted( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shard: String): Unit = () | ||
|
|
||
| override def shardHandoffFinished( | ||
| selfAddress: Address, | ||
| self: ActorRef, | ||
| typeName: String, | ||
| shard: String, | ||
| ok: Boolean): Unit = () | ||
|
|
||
| override def dependencies: immutable.Seq[String] = Nil | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API: Instrumentation SPI for Akka Cluster. | ||
| */ | ||
| @InternalStableApi | ||
| trait ClusterShardingInstrumentation { | ||
|
|
||
| def shardRegionBufferSize(selfAddress: Address, shardRegionActor: ActorRef, typeName: String, size: Int): Unit | ||
|
|
||
| def shardRegionBufferSizeIncremented(selfAddress: Address, shardRegionActor: ActorRef, typeName: String): Unit | ||
|
|
||
| def regionRequestedShardHome( | ||
| selfAddress: Address, | ||
| shardRegionActor: ActorRef, | ||
| typeName: String, | ||
| shardId: String): Unit | ||
|
|
||
| def receivedShardHome(selfAddress: Address, shardRegionActor: ActorRef, typeName: String, shardId: String): Unit | ||
|
|
||
| def shardHandoffStarted(selfAddress: Address, shardCoordinatorActor: ActorRef, typeName: String, shard: String): Unit | ||
|
|
||
| def shardHandoffFinished( | ||
| selfAddress: Address, | ||
| shardCoordinatorActor: ActorRef, | ||
| typeName: String, | ||
| shard: String, | ||
| ok: Boolean): Unit | ||
|
|
||
| /** | ||
| * Optional dependencies for this instrumentation. | ||
| * | ||
| * Dependency instrumentations will always be ordered before this instrumentation. | ||
| * | ||
| * @return list of class names for optional instrumentation dependencies | ||
| */ | ||
| def dependencies: immutable.Seq[String] | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.