Make write thread pools EWMA configurable (#112283)

Relates #112206.

Author: pxsalehi

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

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
@@ -19,6 +20,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.hamcrest.CoreMatchers;
 
 import java.lang.management.ManagementFactory;
 import java.lang.management.ThreadInfo;
@@ -36,12 +38,15 @@
 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;
 
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0, numClientNodes = 0)
@@ -190,4 +195,19 @@ public void testThreadPoolMetrics() throws Exception {
         });
     }
 
+    public void testWriteThreadpoolEwmaAlphaSetting() {
+        Settings settings = Settings.EMPTY;
+        var ewmaAlpha = DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA;
+        if (randomBoolean()) {
+            ewmaAlpha = randomDoubleBetween(0.0, 1.0, true);
+            settings = Settings.builder().put(WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING.getKey(), ewmaAlpha).build();
+        }
+        var nodeName = internalCluster().startNode(settings);
+        var threadPool = internalCluster().getInstance(ThreadPool.class, nodeName);
+        for (var name : List.of(ThreadPool.Names.WRITE, ThreadPool.Names.SYSTEM_WRITE, ThreadPool.Names.SYSTEM_CRITICAL_WRITE)) {
+            assertThat(threadPool.executor(name), instanceOf(TaskExecutionTimeTrackingEsThreadPoolExecutor.class));
+            final var executor = (TaskExecutionTimeTrackingEsThreadPoolExecutor) threadPool.executor(name);
+            assertThat(Double.compare(executor.getEwmaAlpha(), ewmaAlpha), CoreMatchers.equalTo(0));
+        }
+    }
 }

server/src/main/java/org/elasticsearch/common/ExponentiallyWeightedMovingAverage.java

@@ -46,4 +46,9 @@ public void addValue(double newValue) {
             successful = averageBits.compareAndSet(currentBits, newBits);
         } while (successful == false);
     }
+
+    // Used for testing
+    public double getAlpha() {
+        return alpha;
+    }
 }

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

@@ -522,6 +522,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         ThreadPool.ESTIMATED_TIME_INTERVAL_SETTING,
         ThreadPool.LATE_TIME_INTERVAL_WARN_THRESHOLD_SETTING,
         ThreadPool.SLOW_SCHEDULER_TASK_WARN_THRESHOLD_SETTING,
+        ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING,
         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/TaskExecutionTimeTrackingEsThreadPoolExecutor.java

@@ -146,4 +146,9 @@ protected void appendThreadPoolExecutorDetails(StringBuilder sb) {
     public Map<Runnable, Long> getOngoingTasks() {
         return trackOngoingTasks ? Map.copyOf(ongoingTasks) : Map.of();
     }
+
+    // Used for testing
+    public double getEwmaAlpha() {
+        return executionEWMA.getAlpha();
+    }
 }

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

@@ -18,7 +18,7 @@
 import java.util.Map;
 
 import static java.util.Collections.unmodifiableMap;
-import static org.elasticsearch.threadpool.ThreadPool.indexAutoscalingEWMA;
+import static org.elasticsearch.threadpool.ThreadPool.WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING;
 import static org.elasticsearch.threadpool.ThreadPool.searchAutoscalingEWMA;
 
 public class DefaultBuiltInExecutorBuilders implements BuiltInExecutorBuilders {
@@ -29,6 +29,7 @@ public Map<String, ExecutorBuilder> getBuilders(Settings settings, int allocated
         final int halfProcMaxAt5 = ThreadPool.halfAllocatedProcessorsMaxFive(allocatedProcessors);
         final int halfProcMaxAt10 = ThreadPool.halfAllocatedProcessorsMaxTen(allocatedProcessors);
         final int genericThreadPoolMax = ThreadPool.boundedBy(4 * allocatedProcessors, 128, 512);
+        final double indexAutoscalingEWMA = WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING.get(settings);
 
         Map<String, ExecutorBuilder> result = new HashMap<>();
         result.put(

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

@@ -183,7 +183,7 @@ public static ThreadPoolType fromType(String type) {
     // EWMA value is at least within 90% of the new increased task duration. This value also determines the impact of a single
     // long-running task on the moving average and limits it roughly to 2% of the (long) task duration, e.g. if the current
     // moving average is 100ms, and we get one task which takes 20s the new EWMA will be ~500ms.
-    public static final double indexAutoscalingEWMA = 0.02;
+    public static final double DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA = 0.02;
 
     private final Map<String, ExecutorHolder> executors;
 
@@ -230,6 +230,15 @@ public Collection<ExecutorBuilder> builders() {
         Setting.Property.NodeScope
     );
 
+    // A setting to change the alpha parameter of the EWMA used in WRITE, SYSTEM_WRITE and SYSTEM_CRITICAL_WRITE thread pools
+    public static final Setting<Double> WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING = Setting.doubleSetting(
+        "thread_pool.write.ewma_alpha",
+        DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA,
+        0.0,
+        1.0,
+        Setting.Property.NodeScope
+    );
+
     /**
      * Defines and builds the many thread pools delineated in {@link Names}.
      *