Update/Add Required Metrics to match latest design doc

ggivo · ggivo · commit 26ccb0d993c0 · 2025-07-30T09:30:41.000+03:00
Added - redis.connections.total - redis.connection.drops.total - redis.pubsub.operations.total - redis.operations.total - redis.operation.duration Not implemented - redis_active_connections Ref : - https://redislabs.atlassian.net/wiki/spaces/CAE/pages/5011111938/High-level+design+Lettuce-based+test+application#Required-Metrics - https://github.com/redis-developer/observability-stack/blob/main/README.md
diff --git a/application.properties b/application.properties
@@ -15,7 +15,12 @@ management.influx.metrics.export.bucket=BUCKET
 management.influx.metrics.export.token=TOKEN
 management.influx.metrics.export.auto-create-bucket=true
 management.influx.metrics.export.consistency=one
-management.influx.metrics.export.step=PT5S
+management.influx.metrics.export.step=PT1S
+
+# OpenTelemetry Configuration for Micrometer (Spring Boot 3.x)
+management.otlp.metrics.export.enabled=false
+management.otlp.metrics.export.url=http://localhost:4318/v1/metrics
+management.otlp.metrics.export.step=PT1S
 
 #logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss} - %msg%n
 #logging.level.org.springframework=DEBUG
diff --git a/pom.xml b/pom.xml
@@ -33,6 +33,10 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-influx</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-otlp</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
diff --git a/src/main/java/io/lettuce/test/WorkloadRunnerBase.java b/src/main/java/io/lettuce/test/WorkloadRunnerBase.java
@@ -9,6 +9,7 @@
 import io.lettuce.core.cluster.event.ClusterTopologyChangedEvent;
 import io.lettuce.core.event.Event;
 import io.lettuce.core.event.EventBus;
+import io.lettuce.core.event.connection.DisconnectedEvent;
 import io.lettuce.core.event.connection.ReconnectAttemptEvent;
 import io.lettuce.core.event.connection.ReconnectEventHelper;
 import io.lettuce.core.event.connection.ReconnectFailedEvent;
@@ -24,6 +25,8 @@
 import io.lettuce.test.config.WorkloadRunnerConfig.WorkloadConfig;
 import io.lettuce.test.metrics.ConnectionKey;
 import io.lettuce.test.metrics.MetricsReporter;
+import io.lettuce.test.metrics.MetricsReporter.ReconnectAttemptKey;
+import io.lettuce.test.metrics.OperationStatus;
 import io.lettuce.test.workloads.BaseWorkload;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Timer;
@@ -172,15 +175,15 @@ private BaseWorkload withErrorHandler(BaseWorkload task, C client, Conn conn, Wo
 
             @Override
             public void run() {
-                Timer.Sample timer = metricsReporter.startTimer();
+                Timer.Sample timer = WorkloadRunnerBase.this.metricsReporter.startTimer();
                 try {
                     task.run();
-                    metricsReporter.recordWorkloadExecutionDuration(timer, config.getType(), BaseWorkload.Status.SUCCESSFUL);
+                    WorkloadRunnerBase.this.metricsReporter.recordWorkloadExecutionDuration(timer, config.getType(), BaseWorkload.Status.SUCCESSFUL);
                 } catch (Exception e) {
                     // Note: Use client and conn reference to track, which client and connection caused the error
                     // could not find other means to identify the client and connection
                     log.error("Error client: {} conn: {}", client, conn, e);
-                    metricsReporter.recordWorkloadExecutionDuration(timer, config.getType(),
+                    WorkloadRunnerBase.this.metricsReporter.recordWorkloadExecutionDuration(timer, config.getType(),
                             BaseWorkload.Status.COMPLETED_WITH_ERRORS);
                     throw e;
                 }
@@ -237,14 +240,19 @@ private void subscribeToReconnectEvents(EventBus eventBus) {
         Flux<Event> events = eventBus.get();
 
         events.subscribe(event -> {
-            if (event instanceof ReconnectAttemptEvent) {
-                ReconnectAttemptEvent reconnectEvent = (ReconnectAttemptEvent) event;
+            if (event instanceof ReconnectAttemptEvent reconnectEvent) {
                 ConnectionKey connectionKey = ReconnectEventHelper.connectionKey(reconnectEvent);
+                ReconnectAttemptKey reconnectionKey = new ReconnectAttemptKey(connectionKey, OperationStatus.INITIATED);
                 metricsReporter.incrementReconnectAttempt(connectionKey);
-            } else if (event instanceof ReconnectFailedEvent) {
-                ReconnectFailedEvent reconnectEvent = (ReconnectFailedEvent) event;
+                metricsReporter.incrementRedisConnectionAttempt(reconnectionKey);
+            } else if (event instanceof ReconnectFailedEvent reconnectEvent) {
                 ConnectionKey connectionKey = ReconnectEventHelper.connectionKey(reconnectEvent);
+                ReconnectAttemptKey reconnectionKey = new ReconnectAttemptKey(connectionKey, OperationStatus.ERROR);
                 metricsReporter.incrementReconnectFailure(connectionKey);
+                metricsReporter.incrementRedisConnectionAttempt(reconnectionKey);
+            } else if (event instanceof DisconnectedEvent disconnectedEvent) {
+                ConnectionKey connectionKey = ReconnectEventHelper.connectionKey(disconnectedEvent);
+                metricsReporter.incrementRedisConnectionDrops(connectionKey);
             } else if (event instanceof ClusterTopologyChangedEvent) {
                 ClusterTopologyChangedEvent ctcEvent = (ClusterTopologyChangedEvent) event;
                 log.info("ClusterTopologyChangedEvent: before={}, after={}", ctcEvent.before(), ctcEvent.after());
@@ -337,7 +345,8 @@ private void applyTimeoutOptions(TimeoutOptions.Builder builder, TimeoutOptionsC
                 timeoutsRelaxingDuringMaintenance.invoke(builder, config.getTimeoutsRelaxingDuringMaintenance());
                 log.info("ProactiveTimeoutsRelaxingMethod enabled successfully.");
             } catch (NoSuchMethodException e) {
-                throw new IllegalStateException("The method 'timeoutsRelaxingDuringMaintenance' is not available in this build.", e);
+                throw new IllegalStateException(
+                        "The method 'timeoutsRelaxingDuringMaintenance' is not available in this build.", e);
             } catch (IllegalAccessException | InvocationTargetException e) {
                 throw new RuntimeException("Failed to invoke 'timeoutsRelaxingDuringMaintenance' method.", e);
             }
diff --git a/src/main/java/io/lettuce/test/metrics/MeterRegistryConfiguration.java b/src/main/java/io/lettuce/test/metrics/MeterRegistryConfiguration.java
@@ -1,5 +1,6 @@
 package io.lettuce.test.metrics;
 
+import io.lettuce.core.LettuceVersion;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.config.MeterFilter;
 import io.micrometer.core.instrument.logging.LoggingRegistryConfig;
@@ -17,9 +18,24 @@ public class MeterRegistryConfiguration {
     @Value("${runId:${runner.test.workload.type}-#{T(org.apache.commons.lang3.RandomStringUtils).randomAlphanumeric(8)}}")
     private String runId;
 
+    @Value("${instanceId:${spring.application.name}-#{T(org.apache.commons.lang3.RandomStringUtils).randomAlphanumeric(8)}}")
+    private String instanceId;
+
+    @Value("${appName:lettuce-test-app}")
+    private String appName;
+
     @Bean
     public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
-        return (registry) -> registry.config().commonTags("runId", runId);
+        return (registry) -> {
+            registry.config().commonTags("app_name", appName).commonTags("run_id", runId).commonTags("instance_id", instanceId)
+                    .commonTags("version", LettuceVersion.getVersion());
+
+            // Ensure OTLP registry accepts all redis metrics
+            if (registry.getClass().getSimpleName().contains("Otlp")) {
+                registry.config().meterFilter(MeterFilter.acceptNameStartsWith("redis"))
+                        .meterFilter(MeterFilter.acceptNameStartsWith("lettuce"));
+            }
+        };
     }
 
     @Bean
@@ -56,8 +72,7 @@ public SimpleMeterRegistry simpleMeterRegistry() {
 
     // * Deny all meters by default and accept only specific meters for metrics stored in logs
     private void meterFilters(MeterRegistry registry) {
-        registry.config().meterFilter(MeterFilter.acceptNameStartsWith("redis.command"))
-                .meterFilter(MeterFilter.acceptNameStartsWith("redis.workload"))
+        registry.config().meterFilter(MeterFilter.acceptNameStartsWith("redis"))
                 .meterFilter(MeterFilter.acceptNameStartsWith("lettuce.connect"))
                 .meterFilter(MeterFilter.acceptNameStartsWith("lettuce.reconnect"))
                 .meterFilter(MeterFilter.acceptNameStartsWith("lettuce.reconnection")).meterFilter(MeterFilter.deny());
diff --git a/src/main/java/io/lettuce/test/metrics/MetricsProxy.java b/src/main/java/io/lettuce/test/metrics/MetricsProxy.java
@@ -9,6 +9,9 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 
+import static io.lettuce.test.metrics.MetricsReporter.cmdKeyError;
+import static io.lettuce.test.metrics.MetricsReporter.cmdKeyOk;
+
 public class MetricsProxy<T> implements InvocationHandler {
 
     private static final Logger log = LoggerFactory.getLogger(MetricsProxy.class);
@@ -39,17 +42,20 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
                 command.whenComplete((res, ex) -> {
                     if (ex != null) {
                         metricsReporter.incrementCommandError(commandName);
+                        metricsReporter.recordCommandLatency(cmdKeyError(commandName, ex), sample);
                         log.error("Command failed", ex);
+                    } else {
+                        metricsReporter.recordCommandLatency(cmdKeyOk(commandName), sample);
                     }
-                    metricsReporter.recordCommandLatency(commandName, sample);
                 });
                 return result;
             }
 
-            metricsReporter.recordCommandLatency(commandName, sample);
+            metricsReporter.recordCommandLatency(cmdKeyOk(commandName), sample);
             return result;
         } catch (InvocationTargetException ex) {
             metricsReporter.incrementCommandError(commandName);
+            metricsReporter.recordCommandLatency(cmdKeyError(commandName, ex.getCause()), sample);
             log.error("Command failed", ex.getCause());
             throw ex.getCause();
         }
diff --git a/src/main/java/io/lettuce/test/metrics/MetricsReporter.java b/src/main/java/io/lettuce/test/metrics/MetricsReporter.java
@@ -37,12 +37,14 @@ public class MetricsReporter {
     @Value("${simple.metrics.dumpRate:PT5S}")
     private Duration dumpRate;
 
-    private final Map<String, Timer> commandLatencyTimers = new ConcurrentHashMap<>();
+    private final Map<CommandKey, Timer> commandLatencyTimers = new ConcurrentHashMap<>();
 
     private final Timer commandLatencyTotalTimer;
 
     private final Map<String, Counter> commandErrorCounters = new ConcurrentHashMap<>();
 
+    private final Map<CommandKey, Counter> commandTotalCounter = new ConcurrentHashMap<>();
+
     private final Counter commandErrorTotalCounter;
 
     private final Timer connectionSuccessTimer;
@@ -51,6 +53,12 @@ public class MetricsReporter {
 
     private final Map<ConnectionKey, Counter> reconnectAttemptCounter = new ConcurrentHashMap<>();
 
+    private final Map<ReconnectAttemptKey, Counter> redisConnectionsTotal = new ConcurrentHashMap<>();
+
+    private final Map<ConnectionKey, Counter> redisConnectionsDrops = new ConcurrentHashMap<>();
+
+    private final Map<PubSubOpKey, Counter> pubSubOperationCounter = new ConcurrentHashMap<>();
+
     private final Counter reconnectAttemptTotalCounter;
 
     private final Map<ConnectionKey, Counter> reconnectFailureCounter = new ConcurrentHashMap<>();
@@ -87,11 +95,18 @@ Timer.Sample startCommandTimer() {
         return Timer.start(meterRegistry);
     }
 
-    void recordCommandLatency(String commandName, Timer.Sample sample) {
-        Timer timer = commandLatencyTimers.computeIfAbsent(commandName, this::createCommandLatencyTimer);
+    public record CommandKey(String commandName, OperationStatus status) {
+
+    }
+
+    void recordCommandLatency(CommandKey commandKey, Timer.Sample sample) {
+        Timer timer = commandLatencyTimers.computeIfAbsent(commandKey, this::createCommandLatencyTimer);
         long timeNs = sample.stop(timer);
 
         commandLatencyTotalTimer.record(Duration.ofNanos(timeNs));
+
+        Counter counter = commandTotalCounter.computeIfAbsent(commandKey, this::createCommandTotalCounter);
+        counter.increment();
     }
 
     void incrementCommandError(String commandName) {
@@ -101,8 +116,8 @@ void incrementCommandError(String commandName) {
 
     public void recordWorkloadExecutionDuration(Timer.Sample sample, String workloadName, BaseWorkload.Status status) {
         Timer workloadDuration = Timer.builder("redis.workload.execution.duration")
-                .description("Time taken to complete a workload").tag("workload", workloadName).tag("status", status.toString())
-                .register(meterRegistry);
+                .description("Time taken to complete a workload").tag("workload", workloadName)
+                .tag("status", status.name().toLowerCase()).register(meterRegistry);
         sample.stop(workloadDuration);
     }
 
@@ -119,21 +134,50 @@ public void incrementReconnectAttempt(ConnectionKey connectionKey) {
         reconnectAttemptTotalCounter.increment();
     }
 
+    public void incrementRedisConnectionAttempt(ReconnectAttemptKey reconnectAttempt) {
+        redisConnectionsTotal.computeIfAbsent(reconnectAttempt, this::createRedisConnectionsTotal).increment();
+    }
+
+    public void incrementRedisConnectionDrops(ConnectionKey connectionKey) {
+        redisConnectionsDrops.computeIfAbsent(connectionKey, this::createRedisConnectionsDrops).increment();
+    }
+
     public void incrementReconnectFailure(ConnectionKey connectionKey) {
         reconnectFailureCounter.computeIfAbsent(connectionKey, this::createReconnectFailedAttempCounter).increment();
         reconnectFailureTotalCounter.increment();
     }
 
-    private Timer createCommandLatencyTimer(String commandName) {
-        return Timer.builder("redis.command.latency").description(
+    public enum PubSubOperation {
+        SUBSCRIBE, UNSUBSCRIBE, PUBLISH, RECEIVE
+    }
+
+    public record PubSubOpKey(String channel, PubSubOperation operation, String subsriberId, OperationStatus status) {
+
+    }
+
+    public void incrementPubSubOperation(PubSubOpKey pubSubOpKey) {
+        pubSubOperationCounter.computeIfAbsent(pubSubOpKey, this::createPubSubOperationCounter).increment();
+    }
+
+    private Counter createPubSubOperationCounter(PubSubOpKey pubSubOpKey) {
+        return Counter.builder("redis.pubsub.operations.total")
+                .description("Total number of Redis pub/sub operations (publish and receive)")
+                .tag("channel", pubSubOpKey.channel).tag("operation_type", pubSubOpKey.operation.name().toLowerCase())
+                .tag("subscriber_id", pubSubOpKey.subsriberId).tag("status", pubSubOpKey.status.name().toLowerCase())
+                .register(meterRegistry);
+    }
+
+    private Timer createCommandLatencyTimer(CommandKey commandKey) {
+        return Timer.builder("redis.operation.duration").description(
                 "Measures the execution time of Redis commands from API invocation until command completion per command")
-                .tag("command", commandName).register(meterRegistry);
+                .tag("command", commandKey.commandName).tag("status", commandKey.status.name().toLowerCase())
+                .publishPercentileHistogram(true).publishPercentiles(0.5, 0.95, 0.99).register(meterRegistry);
     }
 
     private Timer createCommandLatencyTotalTimer() {
-        return Timer.builder("redis.command.total.latency")
+        return Timer.builder("redis.operation.total.duration")
                 .description("Measures the execution time of Redis commands from API invocation until command completion")
-                .register(meterRegistry);
+                .publishPercentileHistogram(true).publishPercentiles(0.5, 0.95, 0.99).register(meterRegistry);
     }
 
     private Counter createCommandErrorCounter(String commandName) {
@@ -142,12 +186,39 @@ private Counter createCommandErrorCounter(String commandName) {
                 .tag("command", commandName).register(meterRegistry);
     }
 
+    private Counter createCommandTotalCounter(CommandKey commandKey) {
+        return Counter.builder("redis.operations.total")
+                .description("Counts the number of total Redis command API calls completed successfully or with an error")
+                .tag("command", commandKey.commandName).tag("status", commandKey.status.name().toLowerCase())
+                .register(meterRegistry);
+    }
+
     private Counter createCommandErrorTotalCounter() {
         return Counter.builder("redis.command.total.errors")
                 .description("Counts the number of failed Redis command API calls that completed with an exception")
                 .register(meterRegistry);
     }
 
+    public record ReconnectAttemptKey(ConnectionKey connectionKey, OperationStatus status) {
+    }
+
+    private Counter createRedisConnectionsTotal(ReconnectAttemptKey reconnectAttempt) {
+        ConnectionKey connectionKey = reconnectAttempt.connectionKey;
+        OperationStatus status = reconnectAttempt.status;
+
+        return Counter.builder("redis.connections.total")
+                .description("Counts the number of Redis reconnect attempts per connection")
+                .tag("status", status.name().toLowerCase()).tag("epid", connectionKey.getEpId())
+                .tag("local", connectionKey.getLocalAddress().toString())
+                .tag("remote", connectionKey.getRemoteAddress().toString()).register(meterRegistry);
+    }
+
+    private Counter createRedisConnectionsDrops(ConnectionKey connectionKey) {
+        return Counter.builder("redis.connection.drops.total").description("Counts the number of disconnects")
+                .tag("epid", connectionKey.getEpId()).tag("local", connectionKey.getLocalAddress().toString())
+                .tag("remote", connectionKey.getRemoteAddress().toString()).register(meterRegistry);
+    }
+
     private Counter createReconnectAttemptCounter(ConnectionKey connectionKey) {
         return Counter.builder("lettuce.reconnect.attempts")
                 .description("Counts the number of Redis reconnect attempts per connection")
@@ -214,4 +285,12 @@ public void shutdown() {
         }
     }
 
+    public static CommandKey cmdKeyOk(String commandName) {
+        return new CommandKey(commandName, OperationStatus.SUCCESS);
+    }
+
+    public static CommandKey cmdKeyError(String commandName, Throwable throwable) {
+        return new CommandKey(commandName, OperationStatus.ERROR);
+    }
+
 }
diff --git a/src/main/java/io/lettuce/test/metrics/OperationStatus.java b/src/main/java/io/lettuce/test/metrics/OperationStatus.java
@@ -0,0 +1,5 @@
+package io.lettuce.test.metrics;
+
+public enum OperationStatus {
+    SUCCESS, ERROR, INITIATED
+}
diff --git a/src/main/java/io/lettuce/test/workloads/BaseWorkload.java b/src/main/java/io/lettuce/test/workloads/BaseWorkload.java
@@ -31,7 +31,7 @@ public enum Status {
 
     private static final Logger log = LoggerFactory.getLogger(BaseWorkload.class);
 
-    private MetricsReporter metricsReporter;
+    protected MetricsReporter metricsReporter;
 
     private final CommonWorkloadOptions options;
 
diff --git a/src/main/java/io/lettuce/test/workloads/PubSubWorkload.java b/src/main/java/io/lettuce/test/workloads/PubSubWorkload.java
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
