sub class pollUtilization into UtilizationTracker; rename from EWMA to average

Author: DiannaHohensee

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

@@ -582,9 +582,9 @@ public static class TaskTrackingConfig {
 
         private final boolean trackExecutionTime;
         private final boolean trackOngoingTasks;
-        private final boolean trackQueueLatencyEWMA;
+        private final boolean trackQueueLatencyAverage;
         private final double executionTimeEwmaAlpha;
-        private final double queueLatencyEWMAAlpha;
+        private final double queueLatencyEwmaAlpha;
 
         public static final TaskTrackingConfig DO_NOT_TRACK = new TaskTrackingConfig(
             false,
@@ -610,32 +610,32 @@ public TaskTrackingConfig(boolean trackOngoingTasks, double executionTimeEWMAAlp
          */
         public TaskTrackingConfig(
             boolean trackOngoingTasks,
-            boolean trackQueueLatencyEWMA,
-            double executionTimeEWMAAlpha,
-            double queueLatencyEWMAAlpha
+            boolean trackQueueLatencyAverage,
+            double executionTimeEwmaAlpha,
+            double queueLatencyEwmaAlpha
         ) {
-            this(true, trackOngoingTasks, trackQueueLatencyEWMA, executionTimeEWMAAlpha, queueLatencyEWMAAlpha);
+            this(true, trackOngoingTasks, trackQueueLatencyAverage, executionTimeEwmaAlpha, queueLatencyEwmaAlpha);
         }
 
         /**
          * @param trackExecutionTime Whether to track execution stats
          * @param trackOngoingTasks Whether to track ongoing task execution time, not just finished tasks
-         * @param trackQueueLatencyEWMA Whether to track queue latency {@link org.elasticsearch.common.ExponentiallyWeightedMovingAverage}
+         * @param trackQueueLatencyAverage Whether to track the average queue latency.
          * @param executionTimeEWMAAlpha The alpha seed for execution time EWMA (ExponentiallyWeightedMovingAverage).
-         * @param queueLatencyEWMAAlpha The alpha seed for task queue latency EWMA (ExponentiallyWeightedMovingAverage).
+         * @param queueLatencyEwmaAlpha The alpha seed for task queue latency EWMA (ExponentiallyWeightedMovingAverage).
          */
         private TaskTrackingConfig(
             boolean trackExecutionTime,
             boolean trackOngoingTasks,
-            boolean trackQueueLatencyEWMA,
+            boolean trackQueueLatencyAverage,
             double executionTimeEWMAAlpha,
-            double queueLatencyEWMAAlpha
+            double queueLatencyEwmaAlpha
         ) {
             this.trackExecutionTime = trackExecutionTime;
             this.trackOngoingTasks = trackOngoingTasks;
-            this.trackQueueLatencyEWMA = trackQueueLatencyEWMA;
+            this.trackQueueLatencyAverage = trackQueueLatencyAverage;
             this.executionTimeEwmaAlpha = executionTimeEWMAAlpha;
-            this.queueLatencyEWMAAlpha = queueLatencyEWMAAlpha;
+            this.queueLatencyEwmaAlpha = queueLatencyEwmaAlpha;
         }
 
         public boolean trackExecutionTime() {
@@ -646,16 +646,16 @@ public boolean trackOngoingTasks() {
             return trackOngoingTasks;
         }
 
-        public boolean trackQueueLatencyEWMA() {
-            return trackQueueLatencyEWMA;
+        public boolean trackQueueLatencyAverage() {
+            return trackQueueLatencyAverage;
         }
 
         public double getExecutionTimeEwmaAlpha() {
             return executionTimeEwmaAlpha;
         }
 
         public double getQueueLatencyEwmaAlpha() {
-            return queueLatencyEWMAAlpha;
+            return queueLatencyEwmaAlpha;
         }
     }
 

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

@@ -47,14 +47,18 @@ 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 lastPollTimeAPM = System.nanoTime();
-    private volatile long lastTotalExecutionTimeAPM = 0;
-    private volatile long lastPollTimeNanosAllocation = System.nanoTime();
-    private volatile long lastTotalExecutionTimeAllocation = 0;
     private final ExponentialBucketHistogram queueLatencyMillisHistogram = new ExponentialBucketHistogram(QUEUE_LATENCY_HISTOGRAM_BUCKETS);
-    private final boolean trackQueueLatencyEWMA;
+    private final boolean trackQueueLatencyAverage;
     private final ExponentiallyWeightedMovingAverage queueLatencyMillisEWMA;
 
+    public enum UtilizationTrackingPurpose {
+        APM,
+        ALLOCATION,
+    }
+
+    private volatile UtilizationTracker apmUtilizationTracker;
+    private volatile UtilizationTracker allocationUtilizationTracker;
+
     TaskExecutionTimeTrackingEsThreadPoolExecutor(
         String name,
         int corePoolSize,
@@ -73,8 +77,10 @@ public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThrea
         this.runnableWrapper = runnableWrapper;
         this.executionEWMA = new ExponentiallyWeightedMovingAverage(trackingConfig.getExecutionTimeEwmaAlpha(), 0);
         this.trackOngoingTasks = trackingConfig.trackOngoingTasks();
-        this.trackQueueLatencyEWMA = trackingConfig.trackQueueLatencyEWMA();
+        this.trackQueueLatencyAverage = trackingConfig.trackQueueLatencyAverage();
         this.queueLatencyMillisEWMA = new ExponentiallyWeightedMovingAverage(trackingConfig.getQueueLatencyEwmaAlpha(), 0);
+        this.apmUtilizationTracker = new UtilizationTracker();
+        this.allocationUtilizationTracker = new UtilizationTracker();
     }
 
     public List<Instrument> setupMetrics(MeterRegistry meterRegistry, String threadPoolName) {
@@ -102,7 +108,7 @@ public List<Instrument> setupMetrics(MeterRegistry meterRegistry, String threadP
                 ThreadPool.THREAD_POOL_METRIC_PREFIX + threadPoolName + THREAD_POOL_METRIC_NAME_UTILIZATION,
                 "fraction of maximum thread time utilized for " + threadPoolName,
                 "fraction",
-                () -> new DoubleWithAttributes(pollUtilization(true, false), Map.of())
+                () -> new DoubleWithAttributes(pollUtilization(UtilizationTrackingPurpose.APM), Map.of())
             )
         );
     }
@@ -143,44 +149,30 @@ public int getCurrentQueueSize() {
         return getQueue().size();
     }
 
-    public double getQueuedTaskLatencyMillisEWMA() {
-        if (trackQueueLatencyEWMA == false) {
+    public double getQueuedTaskLatencyMillis() {
+        if (trackQueueLatencyAverage == false) {
             return 0;
         }
         return queueLatencyMillisEWMA.getAverage();
     }
 
     /**
      * Returns the fraction of the maximum possible thread time that was actually used since the last time this method was called.
-     * One of the two boolean parameters must be true, while the other false. There are two periodic pulling mechanisms that access
-     * utilization reporting.
+     * There are two periodic pulling mechanisms that access utilization reporting: {@link UtilizationTrackingPurpose} distinguishes the
+     * caller.
      *
      * @return the utilization as a fraction, in the range [0, 1]. This may return >1 if a task completed in the time range but started
      * earlier, contributing a larger execution time.
      */
-    public double pollUtilization(boolean forAPM, boolean forAllocation) {
-        assert forAPM ^ forAllocation : "Can only collect one or the other, APM: " + forAPM + ", Allocation: " + forAllocation;
-
-        final long currentTotalExecutionTimeNanos = totalExecutionTime.sum();
-        final long currentPollTimeNanos = System.nanoTime();
-
-        final long totalExecutionTimeSinceLastPollNanos = currentTotalExecutionTimeNanos - (forAPM
-            ? lastTotalExecutionTimeAPM
-            : lastTotalExecutionTimeAllocation);
-        final long timeSinceLastPoll = currentPollTimeNanos - (forAPM ? lastPollTimeAPM : lastPollTimeNanosAllocation);
-        final long maximumExecutionTimeSinceLastPollNanos = timeSinceLastPoll * getMaximumPoolSize();
-        final double utilizationSinceLastPoll = (double) totalExecutionTimeSinceLastPollNanos / maximumExecutionTimeSinceLastPollNanos;
-
-        if (forAPM) {
-            lastTotalExecutionTimeAPM = currentTotalExecutionTimeNanos;
-            lastPollTimeAPM = currentPollTimeNanos;
-        } else {
-            assert forAllocation;
-            lastTotalExecutionTimeAllocation = currentTotalExecutionTimeNanos;
-            lastPollTimeNanosAllocation = currentPollTimeNanos;
+    public double pollUtilization(UtilizationTrackingPurpose utilizationTrackingPurpose) {
+        switch (utilizationTrackingPurpose) {
+            case APM:
+                return apmUtilizationTracker.pollUtilization();
+            case ALLOCATION:
+                return allocationUtilizationTracker.pollUtilization();
+            default:
+                throw new IllegalStateException("No operation defined for [" + utilizationTrackingPurpose + "]");
         }
-
-        return utilizationSinceLastPoll;
     }
 
     @Override
@@ -197,7 +189,7 @@ protected void beforeExecute(Thread t, Runnable r) {
         var queueLatencyMillis = TimeUnit.NANOSECONDS.toMillis(taskQueueLatency);
         queueLatencyMillisHistogram.addObservation(queueLatencyMillis);
 
-        if (trackQueueLatencyEWMA) {
+        if (trackQueueLatencyAverage) {
             queueLatencyMillisEWMA.addValue(queueLatencyMillis);
         }
     }
@@ -241,8 +233,8 @@ protected void appendThreadPoolExecutorDetails(StringBuilder sb) {
             .append("total task execution time = ")
             .append(TimeValue.timeValueNanos(getTotalTaskExecutionTime()))
             .append(", ");
-        if (trackQueueLatencyEWMA) {
-            sb.append("task queue EWMA = ").append(TimeValue.timeValueMillis((long) getQueuedTaskLatencyMillisEWMA())).append(", ");
+        if (trackQueueLatencyAverage) {
+            sb.append("task queue EWMA = ").append(TimeValue.timeValueMillis((long) getQueuedTaskLatencyMillis())).append(", ");
         }
     }
 
@@ -269,6 +261,33 @@ public double getQueueLatencyEwmaAlpha() {
 
     // Used for testing
     public boolean trackingQueueLatencyEwma() {
-        return trackQueueLatencyEWMA;
+        return trackQueueLatencyAverage;
+    }
+
+    /**
+     * Supports periodic polling for thread pool utilization. Tracks state since the last polling request so that the average utilization
+     * since the last poll can be calculated for the next polling request.
+     *
+     * Uses the difference of {@link #totalExecutionTime} since the last polling request to determine how much activity has occurred.
+     */
+    private class UtilizationTracker {
+        volatile long lastPollTime = System.nanoTime();;
+        volatile long lastTotalExecutionTime = 0;
+
+        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;
+        }
     }
 }

server/src/test/java/org/elasticsearch/common/util/concurrent/TaskExecutionTimeTrackingEsThreadPoolExecutorTests.java

@@ -116,18 +116,18 @@ public void testQueueLatencyEWMACalculation() throws Exception {
         executor.prestartAllCoreThreads();
         logger.info("--> executor: {}", executor);
 
-        assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.0);
+        assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.0);
         // Using the settableQueuingWrapper each task will report being queued for 1ms
         executeTask(executor, 1);
-        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.6); });
+        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.6); });
         executeTask(executor, 1);
-        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.84); });
+        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.84); });
         executeTask(executor, 1);
-        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.936); });
+        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.936); });
         executeTask(executor, 1);
-        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.9744); });
+        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.9744); });
         executeTask(executor, 1);
-        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillisEWMA(), 0.98976); });
+        assertBusy(() -> { assertDoublesEqual(executor.getQueuedTaskLatencyMillis(), 0.98976); });
         assertThat(executor.getOngoingTasks().toString(), executor.getOngoingTasks().size(), equalTo(0));
         executor.shutdown();
         executor.awaitTermination(10, TimeUnit.SECONDS);

server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.FutureUtils;
 import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor.UtilizationTrackingPurpose;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.telemetry.InstrumentType;
@@ -509,7 +510,7 @@ public void testDetailedUtilizationMetric() throws Exception {
 
             final long beforePreviousCollectNanos = System.nanoTime();
             meterRegistry.getRecorder().collect();
-            double allocationUtilization = executor.pollUtilization(false, true);
+            double allocationUtilization = executor.pollUtilization(UtilizationTrackingPurpose.ALLOCATION);
             final long afterPreviousCollectNanos = System.nanoTime();
 
             var metricValue = metricAsserter.assertLatestMetricValueMatches(
@@ -539,7 +540,7 @@ public void testDetailedUtilizationMetric() throws Exception {
 
             final long beforeMetricsCollectedNanos = System.nanoTime();
             meterRegistry.getRecorder().collect();
-            allocationUtilization = executor.pollUtilization(false, true);
+            allocationUtilization = executor.pollUtilization(UtilizationTrackingPurpose.ALLOCATION);
             final long afterMetricsCollectedNanos = System.nanoTime();
 
             // Calculate upper bound on utilisation metric