Added percentiles and pool name to default micrometer metrics

Author: RSam25

src/main/java/com/uid2/operator/Main.java

@@ -492,7 +492,7 @@ private static Vertx createVertx() {
 
         MicrometerMetricsOptions metricOptions = new MicrometerMetricsOptions()
             .setPrometheusOptions(prometheusOptions)
-            .setLabels(EnumSet.of(Label.HTTP_METHOD, Label.HTTP_CODE, Label.HTTP_PATH))
+            .setLabels(EnumSet.of(Label.HTTP_METHOD, Label.HTTP_CODE, Label.HTTP_PATH, Label.POOL_NAME))
             .setJvmMetricsEnabled(true)
             .setEnabled(true);
         setupMetrics(metricOptions);
@@ -529,6 +529,7 @@ private static void setupMetrics(MicrometerMetricsOptions metricOptions) {
                     Objects.equals(id.getTag(Label.HTTP_CODE.toString()), "404")))
                 .meterFilter(new MeterFilter() {
                     private final String httpServerResponseTime = MetricsDomain.HTTP_SERVER.getPrefix() + MetricsNaming.v4Names().getHttpResponseTime();
+                    private final String poolQueueTime = MetricsDomain.HTTP_SERVER.getPrefix() + MetricsNaming.v4Names().getPoolQueueTime();
 
                     @Override
                     public DistributionStatisticConfig configure(Meter.Id id, DistributionStatisticConfig config) {
@@ -538,6 +539,12 @@ public DistributionStatisticConfig configure(Meter.Id id, DistributionStatisticC
                                 .build()
                                 .merge(config);
                         }
+                        if (id.getName().equals(poolQueueTime)) {
+                            return DistributionStatisticConfig.builder()
+                                .percentiles(0.50, 0.90, 0.95, 0.99)
+                                .build()
+                                .merge(config);
+                        }
                         return config;
                     }
                 })

src/main/java/com/uid2/operator/service/ComputePoolService.java

@@ -12,6 +12,7 @@
 import java.time.Duration;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class ComputePoolService {
@@ -22,15 +23,15 @@ public class ComputePoolService {
 
     private final WorkerExecutor workerExecutor;
 
-    // Queue length: incremented on queue, decremented on completion
-    // since Vert.x worker executor doesn't expose queue length
-    private final AtomicLong queueLength = new AtomicLong(0);
+    // Queue pending: incremented on queue, decremented on dispatch
+    // Tracks tasks waiting in queue but not yet started
+    private final AtomicLong queuePending = new AtomicLong(0);
 
     // Prometheus histogram for queue wait time (time between queued and dispatched)
     private final Timer queueWaitTimer;
 
-    // Prometheus gauge for queue length
-    private final Gauge queueLengthGauge;
+    // Prometheus gauge for queue pending
+    private final Gauge queuePendingGauge;
 
     /**
      * Creates a ComputePoolService with default pool size (available processors - 2, minimum 1).
@@ -58,8 +59,8 @@ public ComputePoolService(Vertx vertx, int poolSize) {
                 .maximumExpectedValue(Duration.ofMillis(500))      // 500ms
                 .register(Metrics.globalRegistry);
 
-        this.queueLengthGauge = Gauge.builder(METRIC_PREFIX + "queue_length", queueLength::get)
-                .description("Number of tasks queued but not yet completed")
+        this.queuePendingGauge = Gauge.builder(METRIC_PREFIX + "queue_pending", queuePending::get)
+                .description("Number of tasks waiting in queue but not yet dispatched to a worker")
                 .register(Metrics.globalRegistry);
 
         LOGGER.info("ComputePoolService initialized with pool size: {}", poolSize);
@@ -76,18 +77,30 @@ public ComputePoolService(Vertx vertx, int poolSize) {
      */
     public <T> Future<T> executeBlocking(Callable<T> callable) {
         final long queuedAt = System.nanoTime();
-        queueLength.incrementAndGet();
+        final AtomicBoolean dispatched = new AtomicBoolean(false);
+
+        queuePending.incrementAndGet();
+
+        try {
+            return workerExecutor.<T>executeBlocking(() -> {
+                dispatched.set(true);
+                queuePending.decrementAndGet();
 
-        return workerExecutor.<T>executeBlocking(() -> {
-            try {
                 final long dispatchedAt = System.nanoTime();
                 queueWaitTimer.record(dispatchedAt - queuedAt, TimeUnit.NANOSECONDS);
 
                 return callable.call();
-            } finally {  
-                queueLength.decrementAndGet();
-            }
-        });
+            }).onComplete(ar -> {
+                // If task was never dispatched, clean up
+                if (!dispatched.get()) {
+                    queuePending.decrementAndGet();
+                }
+            });
+        } catch (Exception e) {
+            // executeBlocking threw before returning a Future
+            queuePending.decrementAndGet();
+            return Future.failedFuture(e);
+        }
     }
 
     /**
@@ -105,10 +118,10 @@ public Future<Void> executeBlocking(Runnable runnable) {
 
 
     /**
-     * Returns the current queue length (tasks queued but not yet completed).
+     * Returns the number of tasks waiting in queue but not yet dispatched to a worker.
      */
-    public long getQueueLength() {
-        return queueLength.get();
+    public long getQueuePending() {
+        return queuePending.get();
     }
 
     public void close() {