Watcher: Ensure watcher thread pool size is reasonably bound (elastic/x-pack-elasticsearch#3056)

The watcher threadpool size was always five times the CPU core
count, resulting in a huge threadpool when with even 24 cores.

This changes the behaviour to be five times the number of cores
by default - as watcher is usually waiting on I/O you should have more
threads than cores, but it maxes out with 50 threads, unless the number
of available cores is higher than that.

relates elastic/x-pack-elasticsearch#3052

Original commit: elastic/x-pack-elasticsearch@eab5deb113

plugin/src/main/java/org/elasticsearch/xpack/watcher/Watcher.java

@@ -449,7 +449,7 @@ public class Watcher implements ActionPlugin {
                     new FixedExecutorBuilder(
                             settings,
                             InternalWatchExecutor.THREAD_POOL_NAME,
-                            5 * EsExecutors.numberOfProcessors(settings),
+                            getWatcherThreadPoolSize(settings),
                             1000,
                             "xpack.watcher.thread_pool");
             return Collections.singletonList(builder);
@@ -457,6 +457,28 @@ public class Watcher implements ActionPlugin {
         return Collections.emptyList();
     }
 
+    /**
+     * A method to indicate the size of the watcher thread pool
+     * As watches are primarily bound on I/O waiting and execute
+     * synchronously, it makes sense to have a certain minimum of a
+     * threadpool size. This means you should start with a fair number
+     * of threads which is more than the number of CPUs, but you also need
+     * to ensure that this number does not go crazy high if you have really
+     * beefy machines. This can still be configured manually.
+     *
+     * Calculation is as follows:
+     * Use five times the number of processors up until 50, then stick with the
+     * number of processors.
+     *
+     * @param settings The current settings
+     * @return A number between 5 and the number of processors
+     */
+    static int getWatcherThreadPoolSize(Settings settings) {
+        int numberOfProcessors = EsExecutors.numberOfProcessors(settings);
+        long size = Math.max(Math.min(5 * numberOfProcessors, 50), numberOfProcessors);
+        return Math.toIntExact(size);
+    }
+
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
         if (false == enabled) {

plugin/src/test/java/org/elasticsearch/xpack/watcher/WatcherPluginTests.java

@@ -20,6 +20,7 @@ import java.util.List;
 import static java.util.Collections.emptyMap;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
 
 public class WatcherPluginTests extends ESTestCase {
 
@@ -74,4 +75,19 @@ public class WatcherPluginTests extends ESTestCase {
         // also no component creation if not enabled
         assertThat(watcher.createComponents(null, null, null, null, null, null, null, null), hasSize(0));
     }
+
+    public void testThreadPoolSize() {
+        // old calculation was 5 * number of processors
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 1).build()), is(5));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 2).build()), is(10));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 4).build()), is(20));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 8).build()), is(40));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 9).build()), is(45));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 10).build()), is(50));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 16).build()), is(50));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 24).build()), is(50));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 50).build()), is(50));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 51).build()), is(51));
+        assertThat(Watcher.getWatcherThreadPoolSize(Settings.builder().put("processors", 96).build()), is(96));
+    }
 }