Add thread pool utilisation metric

docs/changelog/120363.yaml

@@ -0,0 +1,5 @@
+pr: 120363
+summary: Add thread pool utilization metric
+area: "Infra/Metrics"
+type: enhancement
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/threadpool/SimpleThreadPoolIT.java

@@ -11,6 +11,7 @@
 
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugins.Plugin;
@@ -31,22 +32,18 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Set;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
-import static java.util.function.Function.identity;
-import static org.elasticsearch.common.util.Maps.toUnmodifiableSortedMap;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.elasticsearch.threadpool.ThreadPool.DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA;
 import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.hamcrest.Matchers.hasEntry;
-import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.matchesRegex;
 
@@ -165,37 +162,78 @@ public void testThreadPoolMetrics() throws Exception {
         registeredMetrics.addAll(plugin.getRegisteredMetrics(InstrumentType.LONG_ASYNC_COUNTER));
 
         tps[0].forEach(stats -> {
-            Map<String, Long> threadPoolStats = List.of(
-                Map.entry(ThreadPool.THREAD_POOL_METRIC_NAME_COMPLETED, stats.completed()),
-                Map.entry(ThreadPool.THREAD_POOL_METRIC_NAME_ACTIVE, 0L),
-                Map.entry(ThreadPool.THREAD_POOL_METRIC_NAME_CURRENT, 0L),
-                Map.entry(ThreadPool.THREAD_POOL_METRIC_NAME_LARGEST, (long) stats.largest()),
-                Map.entry(ThreadPool.THREAD_POOL_METRIC_NAME_QUEUE, 0L)
-            ).stream().collect(toUnmodifiableSortedMap(e -> stats.name() + e.getKey(), Entry::getValue));
-
-            Function<String, List<Long>> measurementExtractor = name -> {
-                String metricName = ThreadPool.THREAD_POOL_METRIC_PREFIX + name;
-                assertThat(metricName, in(registeredMetrics));
-
-                List<Measurement> measurements = name.endsWith(ThreadPool.THREAD_POOL_METRIC_NAME_COMPLETED)
-                    ? plugin.getLongAsyncCounterMeasurement(metricName)
-                    : plugin.getLongGaugeMeasurement(metricName);
-                return measurements.stream().map(Measurement::getLong).toList();
-            };
-
-            Map<String, List<Long>> measurements = threadPoolStats.keySet()
-                .stream()
-                .collect(toUnmodifiableSortedMap(identity(), measurementExtractor));
+            Map<String, MetricDefinition<?>> metricDefinitions = Map.of(
+                ThreadPool.THREAD_POOL_METRIC_NAME_COMPLETED,
+                new MetricDefinition<>(stats.completed(), TestTelemetryPlugin::getLongAsyncCounterMeasurement, Measurement::getLong),
+                ThreadPool.THREAD_POOL_METRIC_NAME_ACTIVE,
+                new MetricDefinition<>(0L, TestTelemetryPlugin::getLongGaugeMeasurement, Measurement::getLong),
+                ThreadPool.THREAD_POOL_METRIC_NAME_CURRENT,
+                new MetricDefinition<>(0L, TestTelemetryPlugin::getLongGaugeMeasurement, Measurement::getLong),
+                ThreadPool.THREAD_POOL_METRIC_NAME_LARGEST,
+                new MetricDefinition<>((long) stats.largest(), TestTelemetryPlugin::getLongGaugeMeasurement, Measurement::getLong),
+                ThreadPool.THREAD_POOL_METRIC_NAME_QUEUE,
+                new MetricDefinition<>(0L, TestTelemetryPlugin::getLongGaugeMeasurement, Measurement::getLong)
+            );
 
-            logger.info("Stats of `{}`: {}", stats.name(), threadPoolStats);
-            logger.info("Measurements of `{}`: {}", stats.name(), measurements);
+            // TaskExecutionTimeTrackingEsThreadPoolExecutor also publishes a utilization metric
+            if (tp.executor(stats.name()) instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor) {
+                metricDefinitions = Maps.copyMapWithAddedEntry(
+                    metricDefinitions,
+                    ThreadPool.THREAD_POOL_METRIC_NAME_UTILIZATION,
+                    new MetricDefinition<>(0.0d, TestTelemetryPlugin::getDoubleGaugeMeasurement, Measurement::getDouble)
+                );
+            }
 
-            threadPoolStats.forEach(
-                (metric, value) -> assertThat(measurements, hasEntry(equalTo(metric), contains(greaterThanOrEqualTo(value))))
+            metricDefinitions = metricDefinitions.entrySet()
+                .stream()
+                .collect(Collectors.toUnmodifiableMap(e -> stats.name() + e.getKey(), Map.Entry::getValue));
+
+            logger.info(
+                "Measurements of `{}`: {}",
+                stats.name(),
+                metricDefinitions.entrySet()
+                    .stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getMeasurements(plugin, e.getKey())))
             );
+
+            // Validate all metrics
+            metricDefinitions.forEach((name, md) -> md.assertValid(plugin, name));
         });
     }
 
+    private static class MetricDefinition<T extends Comparable<T>> {
+
+        private final T minimumValue;
+        private final BiFunction<TestTelemetryPlugin, String, List<Measurement>> metricExtractor;
+        private final Function<Measurement, T> valueExtractor;
+
+        MetricDefinition(
+            T minimumValue,
+            BiFunction<TestTelemetryPlugin, String, List<Measurement>> metricExtractor,
+            Function<Measurement, T> valueExtractor
+        ) {
+            this.minimumValue = minimumValue;
+            this.metricExtractor = metricExtractor;
+            this.valueExtractor = valueExtractor;
+        }
+
+        public List<T> getMeasurements(TestTelemetryPlugin testTelemetryPlugin, String metricSuffix) {
+            return metricExtractor.apply(testTelemetryPlugin, ThreadPool.THREAD_POOL_METRIC_PREFIX + metricSuffix)
+                .stream()
+                .map(valueExtractor)
+                .toList();
+        }
+
+        public void assertValid(TestTelemetryPlugin testTelemetryPlugin, String metricSuffix) {
+            List<T> metrics = getMeasurements(testTelemetryPlugin, metricSuffix);
+            assertThat(
+                ThreadPool.THREAD_POOL_METRIC_PREFIX + metricSuffix + " is populated",
+                metrics,
+                contains(greaterThanOrEqualTo(minimumValue))
+            );
+        }
+    }
+
     public void testWriteThreadpoolEwmaAlphaSetting() {
         Settings settings = Settings.EMPTY;
         var ewmaAlpha = DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA;

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

@@ -33,6 +33,8 @@ public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThrea
     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 volatile long lastPollTime = System.nanoTime();
+    private volatile long lastTotalExecutionTime = 0;
 
     TaskExecutionTimeTrackingEsThreadPoolExecutor(
         String name,
@@ -89,6 +91,26 @@ public int getCurrentQueueSize() {
         return getQueue().size();
     }
 
+    /**
+     * Returns the fraction of the maximum possible thread time that was actually used since the last time
+     * this method was called.
+     *
+     * @return the utilization as a fraction, in the range [0, 1]
+     */
+    public double pollUtilization() {
+        final long currentTotalExecutionTimeNanos = totalExecutionTime.sum();
+        final long currentPollTimeNanos = System.nanoTime();
+
+        final long totalExecutionTimeSinceLastPollNanos = currentTotalExecutionTimeNanos - lastTotalExecutionTime;
+        final long timeSinceLastPoll = currentPollTimeNanos - lastPollTime;
+        final long maximumExecutionTimeSinceLastPollNanos = timeSinceLastPoll * getMaximumPoolSize();
+        final double utilizationSinceLastPoll = (double) totalExecutionTimeSinceLastPollNanos / maximumExecutionTimeSinceLastPollNanos;
+
+        lastTotalExecutionTime = currentTotalExecutionTimeNanos;
+        lastPollTime = currentPollTimeNanos;
+        return utilizationSinceLastPoll;
+    }
+
     @Override
     protected void beforeExecute(Thread t, Runnable r) {
         if (trackOngoingTasks) {

server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -24,12 +24,14 @@
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionHandler;
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.node.ReportingService;
+import org.elasticsearch.telemetry.metric.DoubleWithAttributes;
 import org.elasticsearch.telemetry.metric.Instrument;
 import org.elasticsearch.telemetry.metric.LongAsyncCounter;
 import org.elasticsearch.telemetry.metric.LongGauge;
@@ -149,6 +151,7 @@ public static class Names {
     public static final String THREAD_POOL_METRIC_NAME_CURRENT = ".threads.count.current";
     public static final String THREAD_POOL_METRIC_NAME_QUEUE = ".threads.queue.size";
     public static final String THREAD_POOL_METRIC_NAME_ACTIVE = ".threads.active.current";
+    public static final String THREAD_POOL_METRIC_NAME_UTILIZATION = ".threads.utilization.current";
     public static final String THREAD_POOL_METRIC_NAME_LARGEST = ".threads.largest.current";
     public static final String THREAD_POOL_METRIC_NAME_REJECTED = ".threads.rejected.total";
 
@@ -374,6 +377,17 @@ private static ArrayList<Instrument> setupMetrics(MeterRegistry meterRegistry, S
             if (rejectedExecutionHandler instanceof EsRejectedExecutionHandler handler) {
                 handler.registerCounter(meterRegistry, prefix + THREAD_POOL_METRIC_NAME_REJECTED, name);
             }
+
+            if (holder.executor() instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor timeTrackingExecutor) {
+                instruments.add(
+                    meterRegistry.registerDoubleGauge(
+                        prefix + THREAD_POOL_METRIC_NAME_UTILIZATION,
+                        "fraction of maximum thread time utilized for " + name,
+                        "fraction",
+                        () -> new DoubleWithAttributes(timeTrackingExecutor.pollUtilization(), at)
+                    )
+                );
+            }
         }
         return instruments;
     }