POC Dynamic threadpool

Signed-off-by: Gaurav Bafna <[email protected]>
opensearch-project · Oct 4, 2024 · 8df3a51 · 8df3a51
1 parent ebdf154
commit 8df3a51
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 2 deletions.
diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java
@@ -802,7 +802,10 @@ public void apply(Settings value, Settings current, Settings previous) {
                 // Settings to be used for limiting rest requests
                 ResponseLimitSettings.CAT_INDICES_RESPONSE_LIMIT_SETTING,
                 ResponseLimitSettings.CAT_SHARDS_RESPONSE_LIMIT_SETTING,
-                ResponseLimitSettings.CAT_SEGMENTS_RESPONSE_LIMIT_SETTING
+                ResponseLimitSettings.CAT_SEGMENTS_RESPONSE_LIMIT_SETTING,
+
+                //Thread pool
+                ThreadPool.THREADPOOL_SNAPSHOT_SETTING
             )
         )
     );

diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
@@ -563,7 +563,7 @@ protected Node(
             final List<ExecutorBuilder<?>> executorBuilders = pluginsService.getExecutorBuilders(settings);
 
             runnableTaskListener = new AtomicReference<>();
-            final ThreadPool threadPool = new ThreadPool(settings, runnableTaskListener, executorBuilders.toArray(new ExecutorBuilder[0]));
+            final ThreadPool threadPool = new ThreadPool(settings, runnableTaskListener,  executorBuilders.toArray(new ExecutorBuilder[0]));
 
             final IdentityService identityService = new IdentityService(settings, threadPool, identityPlugins);
 
@@ -619,6 +619,7 @@ protected Node(
                 additionalSettingsFilter,
                 settingsUpgraders
             );
+            threadPool.setClusterSettings(settingsModule.getClusterSettings());
             scriptModule.registerClusterSettingsListeners(scriptService, settingsModule.getClusterSettings());
             final NetworkService networkService = new NetworkService(
                 getCustomNameResolvers(pluginsService.filterPlugins(DiscoveryPlugin.class))

diff --git a/server/src/main/java/org/opensearch/threadpool/ThreadPool.java b/server/src/main/java/org/opensearch/threadpool/ThreadPool.java
@@ -38,6 +38,7 @@
 import org.opensearch.Version;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.SizeValue;
@@ -211,6 +212,9 @@ public static ThreadPoolType fromType(String type) {
 
     private final ScheduledThreadPoolExecutor scheduler;
 
+    private ClusterSettings clusterSettings = null;
+
+
     public Collection<ExecutorBuilder> builders() {
         return Collections.unmodifiableCollection(builders.values());
     }
@@ -226,6 +230,30 @@ public ThreadPool(final Settings settings, final ExecutorBuilder<?>... customBui
         this(settings, null, customBuilders);
     }
 
+    public static final Setting<Integer> THREADPOOL_SNAPSHOT_SETTING = Setting.intSetting(
+        "cluster.thread_pool.snapshot",
+        -1,
+        -1,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
+    public void setSnapshotThread(int snapshotThread) {
+        OpenSearchThreadPoolExecutor o = (OpenSearchThreadPoolExecutor) this.executors.get(Names.SNAPSHOT).executor;
+        if (snapshotThread != -1) {
+            o.setCorePoolSize(snapshotThread);
+            o.setMaximumPoolSize(snapshotThread);
+        }
+    }
+
+    public void setClusterSettings(ClusterSettings clusterSettings) {
+        this.clusterSettings = clusterSettings;
+        this.clusterSettings.addSettingsUpdateConsumer(
+            THREADPOOL_SNAPSHOT_SETTING,
+            this::setSnapshotThread
+        );
+    }
+
     public ThreadPool(
         final Settings settings,
         final AtomicReference<RunnableTaskExecutionListener> runnableTaskListener,