Publish queue latency metrics from tracked thread pools

docs/changelog/120488.yaml

@@ -0,0 +1,5 @@
+pr: 120488
+summary: Publish queue latency metrics from tracked thread pools
+area: "Infra/Metrics"
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/common/metrics/ExponentialBucketHistogram.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.metrics;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * A histogram with a fixed number of buckets of exponentially increasing width.
+ * <p>
+ * The bucket boundaries are defined by increasing powers of two, e.g.
+ * <code>
+ *     (-&infin;, 1), [1, 2), [2, 4), [4, 8), ..., [2^({@link #BUCKET_COUNT}-2), &infin;)
+ * </code>
+ * There are {@link #BUCKET_COUNT} buckets.
+ */
+public class ExponentialBucketHistogram {
+
+    public static int[] getBucketUpperBounds() {
+        int[] bounds = new int[17];
+        for (int i = 0; i < bounds.length; i++) {
+            bounds[i] = 1 << i;
+        }
+        return bounds;
+    }
+
+    private static int getBucket(long observedValue) {
+        if (observedValue <= 0) {
+            return 0;
+        } else if (LAST_BUCKET_LOWER_BOUND <= observedValue) {
+            return BUCKET_COUNT - 1;
+        } else {
+            return Long.SIZE - Long.numberOfLeadingZeros(observedValue);
+        }
+    }
+
+    public static final int BUCKET_COUNT = getBucketUpperBounds().length + 1;
+
+    private static final long LAST_BUCKET_LOWER_BOUND = getBucketUpperBounds()[BUCKET_COUNT - 2];
+
+    private final LongAdder[] buckets;
+
+    public ExponentialBucketHistogram() {
+        buckets = new LongAdder[BUCKET_COUNT];
+        for (int i = 0; i < BUCKET_COUNT; i++) {
+            buckets[i] = new LongAdder();
+        }
+    }
+
+    public void addObservation(long observedValue) {
+        buckets[getBucket(observedValue)].increment();
+    }
+
+    /**
+     * @return An array of frequencies of handling times in buckets with upper bounds as returned by {@link #getBucketUpperBounds()}, plus
+     *         an extra bucket for handling times longer than the longest upper bound.
+     */
+    public long[] getHistogram() {
+        final long[] histogram = new long[BUCKET_COUNT];
+        for (int i = 0; i < BUCKET_COUNT; i++) {
+            histogram[i] = buckets[i].longValue();
+        }
+        return histogram;
+    }
+
+    /**
+     * Calculate the Nth percentile value
+     *
+     * @param percentile The percentile as a fraction (in [0, 1.0])
+     * @return A value greater than the specified fraction of values in the histogram
+     */
+    public long getPercentile(float percentile) {
+        assert percentile >= 0 && percentile <= 1;
+        final long[] snapshot = getHistogram();
+        final long totalCount = Arrays.stream(snapshot).sum();
+        long percentileIndex = (long) Math.ceil(totalCount * percentile);
+        for (int i = 0; i < BUCKET_COUNT; i++) {
+            percentileIndex -= snapshot[i];
+            if (percentileIndex <= 0) {
+                if (i == snapshot.length - 1) {
+                    return Long.MAX_VALUE;
+                } else {
+                    return getBucketUpperBounds()[i];
+                }
+            }
+        }
+        assert false : "We shouldn't ever get here";
+        return Long.MAX_VALUE;
+    }
+
+    /**
+     * Clear all values in the histogram (non-atomic)
+     */
+    public void clear() {
+        for (int i = 0; i < BUCKET_COUNT; i++) {
+            buckets[i].reset();
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/network/NetworkService.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.common.network;
 
+import org.elasticsearch.common.metrics.ExponentialBucketHistogram;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -85,14 +86,14 @@ public interface CustomNameResolver {
     }
 
     private final List<CustomNameResolver> customNameResolvers;
-    private final HandlingTimeTracker handlingTimeTracker = new HandlingTimeTracker();
+    private final ExponentialBucketHistogram handlingTimeTracker = new ExponentialBucketHistogram();
     private final ThreadWatchdog threadWatchdog = new ThreadWatchdog();
 
     public NetworkService(List<CustomNameResolver> customNameResolvers) {
         this.customNameResolvers = Objects.requireNonNull(customNameResolvers, "customNameResolvers must be non null");
     }
 
-    public HandlingTimeTracker getHandlingTimeTracker() {
+    public ExponentialBucketHistogram getHandlingTimeTracker() {
         return handlingTimeTracker;
     }
 

server/src/main/java/org/elasticsearch/common/util/concurrent/TaskExecutionTimeTrackingEsThreadPoolExecutor.java

@@ -10,9 +10,17 @@
 package org.elasticsearch.common.util.concurrent;
 
 import org.elasticsearch.common.ExponentiallyWeightedMovingAverage;
+import org.elasticsearch.common.metrics.ExponentialBucketHistogram;
 import org.elasticsearch.common.util.concurrent.EsExecutors.TaskTrackingConfig;
 import org.elasticsearch.core.TimeValue;
-
+import org.elasticsearch.telemetry.metric.Instrument;
+import org.elasticsearch.telemetry.metric.LongGauge;
+import org.elasticsearch.telemetry.metric.LongWithAttributes;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
+import org.elasticsearch.threadpool.ThreadPool;
+
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
@@ -27,12 +35,16 @@
  */
 public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThreadPoolExecutor {
 
+    private static final int[] LATENCY_PERCENTILES_TO_REPORT = { 50, 90, 99 };
+    public static final String THREAD_POOL_METRIC_NAME_QUEUE_TIME = ".threads.queue.latency.histogram";
+
     private final Function<Runnable, WrappedRunnable> runnableWrapper;
     private final ExponentiallyWeightedMovingAverage executionEWMA;
     private final LongAdder totalExecutionTime = new LongAdder();
     private final boolean trackOngoingTasks;
     // The set of currently running tasks and the timestamp of when they started execution in the Executor.
     private final Map<Runnable, Long> ongoingTasks = new ConcurrentHashMap<>();
+    private final ExponentialBucketHistogram queueLatencyHistogram = new ExponentialBucketHistogram();
 
     TaskExecutionTimeTrackingEsThreadPoolExecutor(
         String name,
@@ -53,6 +65,27 @@ public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThrea
         this.trackOngoingTasks = trackingConfig.trackOngoingTasks();
     }
 
+    public List<Instrument> setupMetrics(MeterRegistry meterRegistry, String threadPoolName) {
+        final LongGauge queueLatencyGauge = meterRegistry.registerLongsGauge(
+            ThreadPool.THREAD_POOL_METRIC_PREFIX + threadPoolName + THREAD_POOL_METRIC_NAME_QUEUE_TIME,
+            "Time tasks spent in the queue for the " + threadPoolName + " thread pool",
+            "milliseconds",
+            () -> {
+                List<LongWithAttributes> metricValues = Arrays.stream(LATENCY_PERCENTILES_TO_REPORT)
+                    .mapToObj(
+                        percentile -> new LongWithAttributes(
+                            queueLatencyHistogram.getPercentile(percentile / 100f),
+                            Map.of("percentile", String.valueOf(percentile))
+                        )
+                    )
+                    .toList();
+                queueLatencyHistogram.clear();
+                return metricValues;
+            }
+        );
+        return List.of(queueLatencyGauge);
+    }
+
     @Override
     protected Runnable wrapRunnable(Runnable command) {
         return super.wrapRunnable(this.runnableWrapper.apply(command));
@@ -94,6 +127,11 @@ protected void beforeExecute(Thread t, Runnable r) {
         if (trackOngoingTasks) {
             ongoingTasks.put(r, System.nanoTime());
         }
+        assert super.unwrap(r) instanceof TimedRunnable : "expected only TimedRunnables in queue";
+        final TimedRunnable timedRunnable = (TimedRunnable) super.unwrap(r);
+        timedRunnable.beforeExecute();
+        final long taskQueueLatency = timedRunnable.getQueueTimeNanos();
+        queueLatencyHistogram.addObservation(TimeUnit.NANOSECONDS.toMillis(taskQueueLatency));
     }
 
     @Override

server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java

@@ -18,6 +18,7 @@
 class TimedRunnable extends AbstractRunnable implements WrappedRunnable {
     private final Runnable original;
     private final long creationTimeNanos;
+    private long beforeExecuteTime = -1;
     private long startTimeNanos;
     private long finishTimeNanos = -1;
     private boolean failedOrRejected = false;
@@ -58,6 +59,19 @@ public boolean isForceExecution() {
         return original instanceof AbstractRunnable && ((AbstractRunnable) original).isForceExecution();
     }
 
+    /**
+     * Returns the time in nanoseconds between the creation time and the execution time
+     *
+     * @return The time in nanoseconds or -1 if the task was never de-queued
+     */
+    long getQueueTimeNanos() {
+        if (beforeExecuteTime == -1) {
+            assert false : "beforeExecute must be called before getQueueTimeNanos";
+            return -1;
+        }
+        return beforeExecuteTime - creationTimeNanos;
+    }
+
     /**
      * Return the time this task spent being run.
      * If the task is still running or has not yet been run, returns -1.
@@ -70,6 +84,13 @@ long getTotalExecutionNanos() {
         return Math.max(finishTimeNanos - startTimeNanos, 1);
     }
 
+    /**
+     * Called when the task has reached the front of the queue and is about to be executed
+     */
+    public void beforeExecute() {
+        beforeExecuteTime = System.nanoTime();
+    }
+
     /**
      * If the task was failed or rejected, return true.
      * Otherwise, false.

server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java

@@ -460,7 +460,7 @@ public void incomingRequest(final HttpRequest httpRequest, final HttpChannel htt
             handleIncomingRequest(httpRequest, trackingChannel, httpRequest.getInboundException());
         } finally {
             final long took = threadPool.rawRelativeTimeInMillis() - startTime;
-            networkService.getHandlingTimeTracker().addHandlingTime(took);
+            networkService.getHandlingTimeTracker().addObservation(took);
             final long logThreshold = slowLogThresholdMs;
             if (logThreshold > 0 && took > logThreshold) {
                 logger.warn(

server/src/main/java/org/elasticsearch/http/HttpRouteStats.java

@@ -12,7 +12,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.network.HandlingTimeTracker;
+import org.elasticsearch.common.metrics.ExponentialBucketHistogram;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xcontent.ToXContentObject;
@@ -36,7 +36,7 @@
  * @param totalResponseSize     the total body size (bytes) of responses produced by the HTTP route
  * @param responseSizeHistogram similar to {@code requestSizeHistogram} but for response size
  * @param responseTimeHistogram an array of frequencies of response time (millis) in buckets with upper bounds
- *                              as returned by {@link HandlingTimeTracker#getBucketUpperBounds()}, plus
+ *                              as returned by {@link ExponentialBucketHistogram#getBucketUpperBounds()}, plus
  *                              an extra bucket for handling response time larger than the longest upper bound (currently 65536ms).
  */
 public record HttpRouteStats(
@@ -89,7 +89,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             "millis",
             TimeValue::timeValueMillis,
             responseTimeHistogram,
-            HandlingTimeTracker.getBucketUpperBounds()
+            ExponentialBucketHistogram.getBucketUpperBounds()
         );
         builder.endObject();
 

server/src/main/java/org/elasticsearch/http/HttpRouteStatsTracker.java

@@ -9,7 +9,7 @@
 
 package org.elasticsearch.http;
 
-import org.elasticsearch.common.network.HandlingTimeTracker;
+import org.elasticsearch.common.metrics.ExponentialBucketHistogram;
 
 import java.util.concurrent.atomic.AtomicLongArray;
 import java.util.concurrent.atomic.LongAdder;
@@ -80,7 +80,7 @@ private static int bucket(int contentLength) {
 
     private final StatsTracker requestStats = new StatsTracker();
     private final StatsTracker responseStats = new StatsTracker();
-    private final HandlingTimeTracker responseTimeTracker = new HandlingTimeTracker();
+    private final ExponentialBucketHistogram responseTimeTracker = new ExponentialBucketHistogram();
 
     public void addRequestStats(int contentLength) {
         requestStats.addStats(contentLength);
@@ -91,7 +91,7 @@ public void addResponseStats(long contentLength) {
     }
 
     public void addResponseTime(long timeMillis) {
-        responseTimeTracker.addHandlingTime(timeMillis);
+        responseTimeTracker.addObservation(timeMillis);
     }
 
     public HttpRouteStats getStats() {