Merge pull request #97 from AxonIQ/feature/processing-latency

CodeDrivenMitch · web-flow · commit ce2a0fe569a6 · 2025-01-23T16:47:47.000Z
Add processing latency to the client
diff --git a/console-framework-client-api/pom.xml b/console-framework-client-api/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright (c) 2022-2024. AxonIQ B.V.
+  ~ 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.
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.8.2-SNAPSHOT</version>
+        <version>1.9.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/eventProcessorApi.kt b/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/eventProcessorApi.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023. AxonIQ B.V.
+ * 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.
@@ -57,6 +57,7 @@ data class SegmentStatus(
         val errorMessage: String?,
         val ingestLatency: Double?,
         val commitLatency: Double?,
+        val processingLatency: Double?,
         val position: Long? = -1,
         val resetPosition: Long? = -1,
 )
diff --git a/console-framework-client-spring-boot-starter/pom.xml b/console-framework-client-spring-boot-starter/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright (c) 2022-2024. AxonIQ B.V.
+  ~ 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.
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.8.2-SNAPSHOT</version>
+        <version>1.9.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client/pom.xml b/console-framework-client/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright (c) 2022-2024. AxonIQ B.V.
+  ~ 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.
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.8.2-SNAPSHOT</version>
+        <version>1.9.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/ProcessorReportCreator.kt b/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/ProcessorReportCreator.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023. AxonIQ B.V.
+ * 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.
@@ -121,6 +121,7 @@ class ProcessorReportCreator(
             errorMessage = this.error?.message,
             ingestLatency = metricsRegistry.ingestLatencyForProcessor(name, this.segment.segmentId).getValue(),
             commitLatency = metricsRegistry.commitLatencyForProcessor(name, this.segment.segmentId).getValue(),
+            processingLatency = metricsRegistry.processingMessageLatencyForProcessor(name, this.segment.segmentId)?.toDouble() ?: -1.0,
             position = this.currentPosition?.orElse(-1) ?: -1,
             resetPosition = this.resetPosition?.orElse(-1) ?: -1,
     )
diff --git a/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/AxoniqConsoleProcessorInterceptor.kt b/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/AxoniqConsoleProcessorInterceptor.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024. AxonIQ B.V.
+ * 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.
@@ -30,8 +30,8 @@ import java.time.Instant
 import java.time.temporal.ChronoUnit
 
 class AxoniqConsoleProcessorInterceptor(
-    private val processorMetricsRegistry: ProcessorMetricsRegistry,
-    private val processorName: String,
+        private val processorMetricsRegistry: ProcessorMetricsRegistry,
+        private val processorName: String,
 ) : MessageHandlerInterceptor<Message<*>> {
     private val logger = LoggerFactory.getLogger(this::class.java)
 
@@ -40,31 +40,36 @@ class AxoniqConsoleProcessorInterceptor(
         if (uow == null || unitOfWork.message.payload is UnknownSerializedType) {
             return interceptorChain.proceed()
         }
+        val message = unitOfWork.message
+        if (message !is EventMessage) {
+            return interceptorChain.proceed()
+        }
         try {
             AxoniqConsoleSpanFactory.onTopLevelSpanIfActive {
                 it.reportProcessorName(processorName)
             }
-            val message = unitOfWork.message
-            if (message is EventMessage) {
-                val segment = unitOfWork.resources()["Processor[$processorName]/SegmentId"] as? Int ?: -1
-                processorMetricsRegistry.registerIngested(
+            val segment = unitOfWork.resources()["Processor[$processorName]/SegmentId"] as? Int ?: -1
+            processorMetricsRegistry.registerIngested(
                     processorName,
                     segment,
                     ChronoUnit.NANOS.between(message.timestamp, Instant.now())
-                )
-                if (unitOfWork !is BatchingUnitOfWork<*> || unitOfWork.isLastMessage) {
-                    unitOfWork.afterCommit {
-                        processorMetricsRegistry.registerCommitted(
+            )
+            if (unitOfWork !is BatchingUnitOfWork<*> || unitOfWork.isFirstMessage) {
+                unitOfWork.afterCommit {
+                    processorMetricsRegistry.registerCommitted(
                             processorName,
                             segment,
                             ChronoUnit.NANOS.between(message.timestamp, Instant.now())
-                        )
-                    }
+                    )
                 }
             }
+
+            return processorMetricsRegistry.doWithActiveMessageForSegment(processorName, segment, message.timestamp) {
+                interceptorChain.proceed()
+            }
         } catch (e: Exception) {
             logger.debug("AxonIQ Console could not register metrics for processor $processorName", e)
+            return interceptorChain.proceed()
         }
-        return interceptorChain.proceed()
     }
 }
diff --git a/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/ProcessorMetricsRegistry.kt b/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/ProcessorMetricsRegistry.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2023. AxonIQ B.V.
+ * 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.
@@ -17,14 +17,18 @@
 package io.axoniq.console.framework.eventprocessor.metrics
 
 import io.axoniq.console.framework.computeIfAbsentWithRetry
+import org.axonframework.messaging.unitofwork.CurrentUnitOfWork
 import java.time.Clock
+import java.time.Instant
+import java.time.temporal.ChronoUnit
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
 
 class ProcessorMetricsRegistry {
     private val ingestLatencyRegistry: MutableMap<String, MutableMap<Int, ExpiringLatencyValue>> = ConcurrentHashMap()
     private val commitLatencyRegistry: MutableMap<String, MutableMap<Int, ExpiringLatencyValue>> = ConcurrentHashMap()
+    private val processingLatencyRegistry: MutableMap<String, MutableMap<Int, Instant?>> = ConcurrentHashMap()
 
     fun registerIngested(processor: String, segment: Int, latencyInNanos: Long) {
         ingestLatencyForProcessor(processor, segment).setValue(latencyInNanos.toDouble() / 1000000)
@@ -34,18 +38,44 @@ class ProcessorMetricsRegistry {
         commitLatencyForProcessor(processor, segment).setValue(latencyInNanos.toDouble() / 1000000)
     }
 
+    fun <T> doWithActiveMessageForSegment(processor: String, segment: Int, messageTimestamp: Instant, action: () -> T?): T? {
+        val processingMessageTimestampsForSegment = getProcessingLatencySegmentMap(processor)
+
+        try {
+            processingMessageTimestampsForSegment[segment] = messageTimestamp
+            return action()
+        } finally {
+            CurrentUnitOfWork.get().afterCommit {
+                getProcessingLatencySegmentMap(processor)
+                        .remove(segment)
+            }
+        }
+    }
+
     fun ingestLatencyForProcessor(processor: String, segment: Int): ExpiringLatencyValue {
         return ingestLatencyRegistry
-            .computeIfAbsentWithRetry(processor) { mutableMapOf() }
-            .computeIfAbsentWithRetry(segment) { ExpiringLatencyValue() }
+                .computeIfAbsentWithRetry(processor) { ConcurrentHashMap() }
+                .computeIfAbsentWithRetry(segment) { ExpiringLatencyValue() }
     }
 
     fun commitLatencyForProcessor(processor: String, segment: Int): ExpiringLatencyValue {
         return commitLatencyRegistry
-            .computeIfAbsentWithRetry(processor) { mutableMapOf() }
-            .computeIfAbsentWithRetry(segment) { ExpiringLatencyValue() }
+                .computeIfAbsentWithRetry(processor) { ConcurrentHashMap() }
+                .computeIfAbsentWithRetry(segment) { ExpiringLatencyValue() }
     }
 
+    fun processingMessageLatencyForProcessor(processor: String, segment: Int): Long? {
+        val processingTimestamp = getProcessingLatencySegmentMap(processor)
+                .computeIfAbsentWithRetry(segment) { null }
+        if (processingTimestamp == null) {
+            return null
+        }
+        return ChronoUnit.MILLIS.between(processingTimestamp, Instant.now())
+    }
+
+    private fun getProcessingLatencySegmentMap(processor: String) = processingLatencyRegistry
+            .computeIfAbsentWithRetry(processor) { ConcurrentHashMap() }
+
     class ExpiringLatencyValue(
         private val expiryTime: Long = 30 * 60 * 1000 // Default to 1 hour
     ) {
diff --git a/pom.xml b/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright (c) 2022-2024. AxonIQ B.V.
+  ~ 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.
@@ -22,7 +22,7 @@
 
     <groupId>io.axoniq.console</groupId>
     <artifactId>console-framework-client-parent</artifactId>
-    <version>1.8.2-SNAPSHOT</version>
+    <version>1.9.0-SNAPSHOT</version>
 
     <modules>
         <module>console-framework-client-api</module>

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`/*`
`2`		`- * Copyright (c) 2022-2023. AxonIQ B.V.`
	`2`	`+ * Copyright (c) 2022-2025. AxonIQ B.V.`
`3`	`3`	`*`
`4`	`4`	`* Licensed under the Apache License, Version 2.0 (the "License");`
`5`	`5`	`* you may not use this file except in compliance with the License.`
`@@ -57,6 +57,7 @@ data class SegmentStatus(`
`57`	`57`	`val errorMessage: String?,`
`58`	`58`	`val ingestLatency: Double?,`
`59`	`59`	`val commitLatency: Double?,`
	`60`	`+ val processingLatency: Double?,`
`60`	`61`	`val position: Long? = -1,`
`61`	`62`	`val resetPosition: Long? = -1,`
`62`	`63`	`)`