From 56dbb94aaf41731cdea12f85d5c5784f6538eb9c Mon Sep 17 00:00:00 2001
From: Mitchell Herrijgers <mitchellherrijgers@gmail.com>
Date: Fri, 3 Jan 2025 15:31:36 +0000
Subject: [PATCH] Add processing latency to the client

Measures the time between the currently processing message and the current clock. This will allow us to detect stalled segments, as this time will increase for every report.

Before, we only had ingest latency, but this will no longer show after 30 minutes. In addition, the nature of that metric does not allow it to increase. It will remain fixed during the processing.
---
 console-framework-client-api/pom.xml          |  4 +-
 .../framework/api/eventProcessorApi.kt        |  3 +-
 .../pom.xml                                   |  4 +-
 console-framework-client/pom.xml              |  4 +-
 .../eventprocessor/ProcessorReportCreator.kt  |  3 +-
 .../AxoniqConsoleProcessorInterceptor.kt      | 33 ++++++++-------
 .../metrics/ProcessorMetricsRegistry.kt       | 40 ++++++++++++++++---
 pom.xml                                       |  4 +-
 8 files changed, 66 insertions(+), 29 deletions(-)

diff --git a/console-framework-client-api/pom.xml b/console-framework-client-api/pom.xml
index 2e3bad3..d89f386 100644
--- 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
index 72112b6..08774a3 100644
--- 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
index 29afd15..7285dfb 100644
--- 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
index 8cf4059..e33c231 100644
--- 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
index 4eea5ed..4c639ff 100644
--- 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
index f1ba070..54bbe54 100644
--- 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
index c05d564..516ac0b 100644
--- 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,7 +17,10 @@
 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
@@ -25,6 +28,7 @@ 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
index f89e9b2..3097d53 100644
--- 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>