reduce utilization EWMA to parallel polling: allow utilization to be polled by allocation

Author: DiannaHohensee

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

@@ -41,10 +41,8 @@
 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.DEFAULT_WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA;
-import static org.elasticsearch.threadpool.ThreadPool.DEFAULT_WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA;
 import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING;
 import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA;
-import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -242,15 +240,12 @@ public void testWriteThreadpoolsEwmaAlphaSetting() {
         Settings settings = Settings.EMPTY;
         var executionEwmaAlpha = DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA;
         var queueLatencyEwmaAlpha = DEFAULT_WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA;
-        var threadUtilizationEwmaAlpha = DEFAULT_WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA;
         if (randomBoolean()) {
             executionEwmaAlpha = randomDoubleBetween(0.0, 1.0, true);
             queueLatencyEwmaAlpha = randomDoubleBetween(0.0, 1.0, true);
-            threadUtilizationEwmaAlpha = randomDoubleBetween(0.0, 1.0, true);
             settings = Settings.builder()
                 .put(WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING.getKey(), executionEwmaAlpha)
                 .put(WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA.getKey(), queueLatencyEwmaAlpha)
-                .put(WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA.getKey(), threadUtilizationEwmaAlpha)
                 .build();
         }
         var nodeName = internalCluster().startNode(settings);
@@ -265,13 +260,10 @@ public void testWriteThreadpoolsEwmaAlphaSetting() {
             // Only the WRITE thread pool should enable further tracking.
             if (name.equals(ThreadPool.Names.WRITE) == false) {
                 assertFalse(executor.trackingQueueLatencyEwma());
-                assertFalse(executor.trackUtilizationEwma());
             } else {
                 // Verify that the WRITE thread pool has extra tracking enabled.
                 assertTrue(executor.trackingQueueLatencyEwma());
-                assertTrue(executor.trackUtilizationEwma());
                 assertThat(Double.compare(executor.getQueueLatencyEwmaAlpha(), queueLatencyEwmaAlpha), CoreMatchers.equalTo(0));
-                assertThat(Double.compare(executor.getPoolUtilizationEwmaAlpha(), threadUtilizationEwmaAlpha), CoreMatchers.equalTo(0));
             }
         }
     }

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -548,7 +548,6 @@ public void apply(Settings value, Settings current, Settings previous) {
         ThreadPool.SLOW_SCHEDULER_TASK_WARN_THRESHOLD_SETTING,
         ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING,
         ThreadPool.WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA,
-        ThreadPool.WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA,
         FastVectorHighlighter.SETTING_TV_HIGHLIGHT_MULTI_VALUE,
         Node.BREAKER_TYPE_KEY,
         OperationRouting.USE_ADAPTIVE_REPLICA_SELECTION_SETTING,

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

@@ -579,43 +579,30 @@ public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
     public static class TaskTrackingConfig {
         public static final double DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST = 0.3;
         public static final double DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST = 0.6;
-        public static final double DEFAULT_POOL_UTILIZATION_EWMA_ALPHA_FOR_TEST = 0.6;
 
         private final boolean trackExecutionTime;
         private final boolean trackOngoingTasks;
         private final boolean trackQueueLatencyEWMA;
-        private final boolean trackPoolUtilizationEWMA;
         private final double executionTimeEwmaAlpha;
         private final double queueLatencyEWMAAlpha;
-        private final double poolUtilizationEWMAAlpha;
 
         public static final TaskTrackingConfig DO_NOT_TRACK = new TaskTrackingConfig(
             false,
             false,
             false,
             DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST,
-            DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST,
-            DEFAULT_POOL_UTILIZATION_EWMA_ALPHA_FOR_TEST
+            DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST
         );
         public static final TaskTrackingConfig DEFAULT = new TaskTrackingConfig(
             true,
             false,
             false,
             DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST,
-            DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST,
-            DEFAULT_POOL_UTILIZATION_EWMA_ALPHA_FOR_TEST
+            DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST
         );
 
         public TaskTrackingConfig(boolean trackOngoingTasks, double executionTimeEWMAAlpha) {
-            this(
-                true,
-                trackOngoingTasks,
-                false,
-                false,
-                executionTimeEWMAAlpha,
-                DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST,
-                DEFAULT_POOL_UTILIZATION_EWMA_ALPHA_FOR_TEST
-            );
+            this(true, trackOngoingTasks, false, executionTimeEWMAAlpha, DEFAULT_QUEUE_LATENCY_EWMA_ALPHA_FOR_TEST);
         }
 
         /**
@@ -624,47 +611,31 @@ public TaskTrackingConfig(boolean trackOngoingTasks, double executionTimeEWMAAlp
         public TaskTrackingConfig(
             boolean trackOngoingTasks,
             boolean trackQueueLatencyEWMA,
-            boolean trackPoolUtilizationEWMA,
             double executionTimeEWMAAlpha,
-            double queueLatencyEWMAAlpha,
-            double poolUtilizationEWMAAlpha
+            double queueLatencyEWMAAlpha
         ) {
-            this(
-                true,
-                trackOngoingTasks,
-                trackQueueLatencyEWMA,
-                trackPoolUtilizationEWMA,
-                executionTimeEWMAAlpha,
-                queueLatencyEWMAAlpha,
-                poolUtilizationEWMAAlpha
-            );
+            this(true, trackOngoingTasks, trackQueueLatencyEWMA, 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 trackPoolUtilizationEWMA Whether to track the EWMA for thread pool thread utilization (percent use).
          * @param executionTimeEWMAAlpha The alpha seed for execution time EWMA (ExponentiallyWeightedMovingAverage).
          * @param queueLatencyEWMAAlpha The alpha seed for task queue latency EWMA (ExponentiallyWeightedMovingAverage).
-         * @param poolUtilizationEWMAAlpha The alpha seed for pool utilization EWMA (ExponentiallyWeightedMovingAverage).
          */
         private TaskTrackingConfig(
             boolean trackExecutionTime,
             boolean trackOngoingTasks,
             boolean trackQueueLatencyEWMA,
-            boolean trackPoolUtilizationEWMA,
             double executionTimeEWMAAlpha,
-            double queueLatencyEWMAAlpha,
-            double poolUtilizationEWMAAlpha
+            double queueLatencyEWMAAlpha
         ) {
             this.trackExecutionTime = trackExecutionTime;
             this.trackOngoingTasks = trackOngoingTasks;
             this.trackQueueLatencyEWMA = trackQueueLatencyEWMA;
-            this.trackPoolUtilizationEWMA = trackPoolUtilizationEWMA;
             this.executionTimeEwmaAlpha = executionTimeEWMAAlpha;
             this.queueLatencyEWMAAlpha = queueLatencyEWMAAlpha;
-            this.poolUtilizationEWMAAlpha = poolUtilizationEWMAAlpha;
         }
 
         public boolean trackExecutionTime() {
@@ -679,21 +650,13 @@ public boolean trackQueueLatencyEWMA() {
             return trackQueueLatencyEWMA;
         }
 
-        public boolean trackPoolUtilizationEWMA() {
-            return trackPoolUtilizationEWMA;
-        }
-
         public double getExecutionTimeEwmaAlpha() {
             return executionTimeEwmaAlpha;
         }
 
         public double getQueueLatencyEwmaAlpha() {
             return queueLatencyEWMAAlpha;
         }
-
-        public double getPoolUtilizationEwmaAlpha() {
-            return poolUtilizationEWMAAlpha;
-        }
     }
 
 }

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

@@ -27,7 +27,6 @@
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
@@ -48,14 +47,13 @@ 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;
+    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 trackUtilizationEWMA;
     private final ExponentiallyWeightedMovingAverage queueLatencyMillisEWMA;
-    private final ExponentiallyWeightedMovingAverage percentPoolUtilizationEWMA;
-    private final AtomicReference<Double> lastUtilizationValue = new AtomicReference<>(0.0);
 
     TaskExecutionTimeTrackingEsThreadPoolExecutor(
         String name,
@@ -77,8 +75,6 @@ public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThrea
         this.trackOngoingTasks = trackingConfig.trackOngoingTasks();
         this.trackQueueLatencyEWMA = trackingConfig.trackQueueLatencyEWMA();
         this.queueLatencyMillisEWMA = new ExponentiallyWeightedMovingAverage(trackingConfig.getQueueLatencyEwmaAlpha(), 0);
-        this.trackUtilizationEWMA = trackingConfig.trackPoolUtilizationEWMA();
-        this.percentPoolUtilizationEWMA = new ExponentiallyWeightedMovingAverage(trackingConfig.getPoolUtilizationEwmaAlpha(), 0);
     }
 
     public List<Instrument> setupMetrics(MeterRegistry meterRegistry, String threadPoolName) {
@@ -106,7 +102,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(), Map.of())
+                () -> new DoubleWithAttributes(pollUtilization(true, false), Map.of())
             )
         );
     }
@@ -147,13 +143,6 @@ public int getCurrentQueueSize() {
         return getQueue().size();
     }
 
-    public double getPercentPoolUtilizationEWMA() {
-        if (trackUtilizationEWMA == false) {
-            return 0;
-        }
-        return this.percentPoolUtilizationEWMA.getAverage();
-    }
-
     public double getQueuedTaskLatencyMillisEWMA() {
         if (trackQueueLatencyEWMA == false) {
             return 0;
@@ -162,38 +151,38 @@ public double getQueuedTaskLatencyMillisEWMA() {
     }
 
     /**
-     * Returns the fraction of the maximum possible thread time that was actually used since the last time
-     * this method was called.
+     * 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.
      *
-     * @return the utilization as a fraction, in the range [0, 1]
+     * @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() {
+    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 - lastTotalExecutionTime;
-        final long timeSinceLastPoll = currentPollTimeNanos - lastPollTime;
+        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;
 
-        lastTotalExecutionTime = currentTotalExecutionTimeNanos;
-        lastPollTime = currentPollTimeNanos;
-
-        if (trackUtilizationEWMA) {
-            percentPoolUtilizationEWMA.addValue(utilizationSinceLastPoll);
-            // Test only tracking.
-            assert setUtilizationSinceLastPoll(utilizationSinceLastPoll);
+        if (forAPM) {
+            lastTotalExecutionTimeAPM = currentTotalExecutionTimeNanos;
+            lastPollTimeAPM = currentPollTimeNanos;
+        } else {
+            assert forAllocation;
+            lastTotalExecutionTimeAllocation = currentTotalExecutionTimeNanos;
+            lastPollTimeNanosAllocation = currentPollTimeNanos;
         }
 
         return utilizationSinceLastPoll;
     }
 
-    // Test only
-    private boolean setUtilizationSinceLastPoll(double utilizationSinceLastPoll) {
-        lastUtilizationValue.set(utilizationSinceLastPoll);
-        return true;
-    }
-
     @Override
     protected void beforeExecute(Thread t, Runnable r) {
         if (trackOngoingTasks) {
@@ -209,9 +198,7 @@ protected void beforeExecute(Thread t, Runnable r) {
         queueLatencyMillisHistogram.addObservation(queueLatencyMillis);
 
         if (trackQueueLatencyEWMA) {
-            if (queueLatencyMillis > 0) {
-                queueLatencyMillisEWMA.addValue(queueLatencyMillis);
-            }
+            queueLatencyMillisEWMA.addValue(queueLatencyMillis);
         }
     }
 
@@ -257,9 +244,6 @@ protected void appendThreadPoolExecutorDetails(StringBuilder sb) {
         if (trackQueueLatencyEWMA) {
             sb.append("task queue EWMA = ").append(TimeValue.timeValueMillis((long) getQueuedTaskLatencyMillisEWMA())).append(", ");
         }
-        if (trackUtilizationEWMA) {
-            sb.append("thread pool utilization percentage EWMA = ").append(getPercentPoolUtilizationEWMA()).append(", ");
-        }
     }
 
     /**
@@ -283,18 +267,8 @@ public double getQueueLatencyEwmaAlpha() {
         return queueLatencyMillisEWMA.getAlpha();
     }
 
-    // Used for testing
-    public double getPoolUtilizationEwmaAlpha() {
-        return percentPoolUtilizationEWMA.getAlpha();
-    }
-
     // Used for testing
     public boolean trackingQueueLatencyEwma() {
         return trackQueueLatencyEWMA;
     }
-
-    // Used for testing
-    public boolean trackUtilizationEwma() {
-        return trackUtilizationEWMA;
-    }
 }

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

@@ -22,7 +22,6 @@
 import static java.util.Collections.unmodifiableMap;
 import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING;
 import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA;
-import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA;
 import static org.elasticsearch.threadpool.ThreadPool.searchAutoscalingEWMA;
 
 public class DefaultBuiltInExecutorBuilders implements BuiltInExecutorBuilders {
@@ -35,7 +34,6 @@ public Map<String, ExecutorBuilder> getBuilders(Settings settings, int allocated
         final int genericThreadPoolMax = ThreadPool.boundedBy(4 * allocatedProcessors, 128, 512);
         final double indexAutoscalingEWMA = WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING.get(settings);
         final double queueLatencyEWMAAlpha = WRITE_THREAD_POOL_QUEUE_LATENCY_EWMA_ALPHA.get(settings);
-        final double threadUtilizationEWMAAlpha = WRITE_THREAD_POOL_THREAD_UTILIZATION_EWMA_ALPHA.get(settings);
 
         Map<String, ExecutorBuilder> result = new HashMap<>();
         result.put(
@@ -59,14 +57,7 @@ public Map<String, ExecutorBuilder> getBuilders(Settings settings, int allocated
                 ThreadPool.Names.WRITE,
                 allocatedProcessors,
                 10000,
-                new EsExecutors.TaskTrackingConfig(
-                    true,
-                    true,
-                    true,
-                    indexAutoscalingEWMA,
-                    queueLatencyEWMAAlpha,
-                    threadUtilizationEWMAAlpha
-                )
+                new EsExecutors.TaskTrackingConfig(true, true, indexAutoscalingEWMA, queueLatencyEWMAAlpha)
             )
         );
         int searchOrGetThreadPoolSize = ThreadPool.searchOrGetThreadPoolSize(allocatedProcessors);