Capture per-entity metrics on Repository and EventStore

framework-client-api/src/main/java/io/axoniq/platform/framework/api/metrics/EntityStatisticIdentifier.kt

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2022-2025. AxonIQ B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.axoniq.platform.framework.api.metrics
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+data class EntityStatisticIdentifier(
+    @JsonProperty("n")
+    val entityName: String,
+)

framework-client-api/src/main/java/io/axoniq/platform/framework/api/metrics/EntityStatistics.kt

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2022-2025. AxonIQ B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.axoniq.platform.framework.api.metrics
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+data class EntityStatistics(
+    @JsonProperty("c")
+    val count: Double,
+    @JsonProperty("f")
+    val failed: Double,
+    @JsonProperty("t")
+    val timer: StatisticDistribution?,
+    @JsonProperty("m")
+    val metrics: Map<String, StatisticDistribution>,
+)

framework-client-api/src/main/java/io/axoniq/platform/framework/api/metrics/EntityStatisticsWithIdentifier.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2022-2025. AxonIQ B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.axoniq.platform.framework.api.metrics
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+data class EntityStatisticsWithIdentifier(
+    @JsonProperty("e")
+    val entity: EntityStatisticIdentifier,
+    @JsonProperty("s")
+    val statistics: EntityStatistics,
+)

framework-client-api/src/main/java/io/axoniq/platform/framework/api/metrics/StatisticReport.kt

@@ -25,4 +25,6 @@ data class StatisticReport(
     val dispatchers: List<DispatcherStatisticsWithIdentifier>,
     @JsonProperty("a")
     val aggregates: List<AggregateStatisticsWithIdentifier>,
+    @JsonProperty("e")
+    val entities: List<EntityStatisticsWithIdentifier> = emptyList(),
 )

framework-client/src/main/java/io/axoniq/platform/framework/AxoniqPlatformConfigurerEnhancer.java

@@ -38,6 +38,7 @@
 import io.axoniq.platform.framework.messaging.AxoniqPlatformCommandBus;
 import io.axoniq.platform.framework.messaging.AxoniqPlatformQueryBus;
 import io.axoniq.platform.framework.messaging.HandlerMetricsRegistry;
+import io.axoniq.platform.framework.modelling.EntityMetricsRegistry;
 import org.axonframework.common.configuration.ComponentDefinition;
 import org.axonframework.common.configuration.ComponentRegistry;
 import org.axonframework.common.configuration.Configuration;
@@ -140,12 +141,16 @@ public void enhance(ComponentRegistry registry) {
                                            // The start handler will allow for eager creation
                                            .onStart(Phase.EXTERNAL_CONNECTIONS, c -> {
                                            }))
+                .registerComponent(ComponentDefinition
+                                           .ofType(EntityMetricsRegistry.class)
+                                           .withBuilder(c -> new EntityMetricsRegistry()))
                 .registerComponent(ComponentDefinition
                                            .ofType(HandlerMetricsRegistry.class)
                                            .withBuilder(c -> new HandlerMetricsRegistry(
                                                    c.getComponent(AxoniqConsoleRSocketClient.class),
                                                    c.getComponent(PlatformClientConnectionService.class),
-                                                   c.getComponent(AxoniqPlatformConfiguration.class))))
+                                                   c.getComponent(AxoniqPlatformConfiguration.class),
+                                                   c.getComponent(EntityMetricsRegistry.class))))
                 .registerComponent(ComponentDefinition
                                            .ofType(ApplicationThreadDumpProvider.class)
                                            .withBuilder(c -> new ApplicationThreadDumpProvider()))

framework-client/src/main/java/io/axoniq/platform/framework/eventsourcing/AxoniqPlatformEventStorageEngine.kt

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2022-2026. AxonIQ B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.axoniq.platform.framework.eventsourcing
+
+import io.axoniq.platform.framework.api.metrics.PreconfiguredMetric
+import io.axoniq.platform.framework.messaging.HandlerMeasurement
+import io.axoniq.platform.framework.messaging.HandlerMeasurement.Companion.RESOURCE_KEY
+import io.axoniq.platform.framework.messaging.HandlerMetricsRegistry
+import io.axoniq.platform.framework.messaging.toInformation
+import io.axoniq.platform.framework.modelling.CurrentEntityContext
+import io.axoniq.platform.framework.modelling.EntityMetricsRegistry
+import org.axonframework.common.infra.ComponentDescriptor
+import org.axonframework.eventsourcing.eventstore.AppendCondition
+import org.axonframework.eventsourcing.eventstore.EventStorageEngine
+import org.axonframework.eventsourcing.eventstore.SourcingCondition
+import org.axonframework.eventsourcing.eventstore.TaggedEventMessage
+import org.axonframework.messaging.core.MessageStream
+import org.axonframework.messaging.core.unitofwork.ProcessingContext
+import org.axonframework.messaging.eventhandling.EventMessage
+import org.axonframework.messaging.eventhandling.processing.streaming.token.TrackingToken
+import org.axonframework.messaging.eventstreaming.StreamingCondition
+import java.time.Instant
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
+
+class AxoniqPlatformEventStorageEngine(
+        private val delegate: EventStorageEngine,
+        private val registry: HandlerMetricsRegistry,
+        private val entityMetricsRegistry: EntityMetricsRegistry,
+) : EventStorageEngine {
+
+    override fun appendEvents(condition: AppendCondition, context: ProcessingContext?, events: List<TaggedEventMessage<*>>): CompletableFuture<EventStorageEngine.AppendTransaction<*>> {
+        // First, report dispatches
+        val container = context?.getResource(RESOURCE_KEY)
+        if (container == null) {
+            events.forEach { tm ->
+                val event: EventMessage = tm.event()
+                registry.registerMessageDispatchedWithoutHandling(event.toInformation())
+            }
+        } else {
+            events.forEach { tm ->
+                val event: EventMessage = tm.event()
+                container.reportMessageDispatched(event.toInformation())
+            }
+        }
+        if (context == null) {
+            return delegate.appendEvents(condition, context, events)
+        }
+        // Second, measure commit time if measurement is ongoing
+        val startTime = System.nanoTime()
+        val currentEntity = context.getResource(CurrentEntityContext.RESOURCE_KEY)
+        return delegate.appendEvents(condition, context, events)
+                .whenComplete { _, _ ->
+                    val endTime = System.nanoTime()
+                    HandlerMeasurement.onContext(context) {
+                        it.registerMetricValue(PreconfiguredMetric.EVENT_COMMIT_TIME, endTime - startTime)
+                    }
+                    currentEntity?.let {
+                        entityMetricsRegistry.registerAdditionalTimer(
+                                it,
+                                EntityMetricsRegistry.METRIC_EVENT_COMMIT_TIME,
+                                endTime - startTime,
+                                TimeUnit.NANOSECONDS,
+                        )
+                    }
+                }
+    }
+
+    override fun appendEvents(condition: AppendCondition, context: ProcessingContext?, vararg events: TaggedEventMessage<*>): CompletableFuture<EventStorageEngine.AppendTransaction<*>> {
+        return appendEvents(condition, context, events.toList())
+    }
+
+    override fun source(condition: SourcingCondition): MessageStream<EventMessage> {
+        return delegate.source(condition)
+    }
+
+    override fun stream(condition: StreamingCondition): MessageStream<EventMessage> {
+        return delegate.stream(condition)
+    }
+
+    override fun firstToken(): CompletableFuture<TrackingToken> = delegate.firstToken()
+
+    override fun latestToken(): CompletableFuture<TrackingToken> = delegate.latestToken()
+
+    override fun tokenAt(at: Instant): CompletableFuture<TrackingToken> = delegate.tokenAt(at)
+
+    override fun describeTo(descriptor: ComponentDescriptor) {
+        descriptor.describeWrapperOf(delegate)
+    }
+}