Create listener to refresh search thread resource usage (opensearch-p…

…roject#14832) * [bug fix] fix incorrect coordinator node search resource usages Signed-off-by: Chenyang Ji <[email protected]> * fix bug on serialization when passing task resource usage to coordinator Signed-off-by: Chenyang Ji <[email protected]> * add more unit tests Signed-off-by: Chenyang Ji <[email protected]> * remove query insights plugin related code Signed-off-by: Chenyang Ji <[email protected]> * create per request listener to refresh task resource usage Signed-off-by: Chenyang Ji <[email protected]> * Make new listener API public Signed-off-by: Siddhant Deshmukh <[email protected]> * Add changelog Signed-off-by: Siddhant Deshmukh <[email protected]> * Remove wrong files added Signed-off-by: Siddhant Deshmukh <[email protected]> * Address review comments Signed-off-by: Siddhant Deshmukh <[email protected]> * Build fix Signed-off-by: Siddhant Deshmukh <[email protected]> * Make singleton Signed-off-by: Siddhant Deshmukh <[email protected]> * Address review comments Signed-off-by: Siddhant Deshmukh <[email protected]> * Make sure listener runs before plugin listeners Signed-off-by: Siddhant Deshmukh <[email protected]> * Spotless Signed-off-by: Siddhant Deshmukh <[email protected]> * Minor fix Signed-off-by: Siddhant Deshmukh <[email protected]> --------- Signed-off-by: Chenyang Ji <[email protected]> Signed-off-by: Siddhant Deshmukh <[email protected]> Signed-off-by: Jay Deng <[email protected]> Co-authored-by: Chenyang Ji <[email protected]> Co-authored-by: Jay Deng <[email protected]>
kkewwei · Jul 23, 2024 · 8ff3bcc · 8ff3bcc
1 parent c82a282
commit 8ff3bcc
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 7 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -27,6 +27,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Reduce logging in DEBUG for MasterService:run ([#14795](https://github.com/opensearch-project/OpenSearch/pull/14795))
 - Enabling term version check on local state for all ClusterManager Read Transport Actions ([#14273](https://github.com/opensearch-project/OpenSearch/pull/14273))
 - Add persian_stem filter (([#14847](https://github.com/opensearch-project/OpenSearch/pull/14847)))
+- Create listener to refresh search thread resource usage ([#14832](https://github.com/opensearch-project/OpenSearch/pull/14832))
 - Add rest, transport layer changes for hot to warm tiering - dedicated setup (([#13980](https://github.com/opensearch-project/OpenSearch/pull/13980))
 
 ### Dependencies

diff --git a/server/src/main/java/org/opensearch/action/search/SearchTaskRequestOperationsListener.java b/server/src/main/java/org/opensearch/action/search/SearchTaskRequestOperationsListener.java
@@ -0,0 +1,30 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.search;
+
+import org.opensearch.tasks.TaskResourceTrackingService;
+
+/**
+ * SearchTaskRequestOperationsListener subscriber for operations on search tasks resource usages.
+ * Listener ensures to refreshResourceStats on request end capturing the search task resource usage
+ * upon request completion.
+ *
+ */
+public final class SearchTaskRequestOperationsListener extends SearchRequestOperationsListener {
+    private final TaskResourceTrackingService taskResourceTrackingService;
+
+    public SearchTaskRequestOperationsListener(TaskResourceTrackingService taskResourceTrackingService) {
+        this.taskResourceTrackingService = taskResourceTrackingService;
+    }
+
+    @Override
+    public void onRequestEnd(SearchPhaseContext context, SearchRequestContext searchRequestContext) {
+        taskResourceTrackingService.refreshResourceStats(context.getTask());
+    }
+}
diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
@@ -52,6 +52,7 @@
 import org.opensearch.action.search.SearchRequestOperationsListener;
 import org.opensearch.action.search.SearchRequestSlowLog;
 import org.opensearch.action.search.SearchRequestStats;
+import org.opensearch.action.search.SearchTaskRequestOperationsListener;
 import org.opensearch.action.search.SearchTransportService;
 import org.opensearch.action.support.TransportAction;
 import org.opensearch.action.update.UpdateHelper;
@@ -855,8 +856,17 @@ protected Node(
                 threadPool
             );
 
+            final TaskResourceTrackingService taskResourceTrackingService = new TaskResourceTrackingService(
+                settings,
+                clusterService.getClusterSettings(),
+                threadPool
+            );
+
             final SearchRequestStats searchRequestStats = new SearchRequestStats(clusterService.getClusterSettings());
             final SearchRequestSlowLog searchRequestSlowLog = new SearchRequestSlowLog(clusterService);
+            final SearchTaskRequestOperationsListener searchTaskRequestOperationsListener = new SearchTaskRequestOperationsListener(
+                taskResourceTrackingService
+            );
 
             remoteStoreStatsTrackerFactory = new RemoteStoreStatsTrackerFactory(clusterService, settings);
             CacheModule cacheModule = new CacheModule(pluginsService.filterPlugins(CachePlugin.class), settings);
@@ -988,7 +998,7 @@ protected Node(
             final SearchRequestOperationsCompositeListenerFactory searchRequestOperationsCompositeListenerFactory =
                 new SearchRequestOperationsCompositeListenerFactory(
                     Stream.concat(
-                        Stream.of(searchRequestStats, searchRequestSlowLog),
+                        Stream.of(searchRequestStats, searchRequestSlowLog, searchTaskRequestOperationsListener),
                         pluginComponents.stream()
                             .filter(p -> p instanceof SearchRequestOperationsListener)
                             .map(p -> (SearchRequestOperationsListener) p)
@@ -1117,12 +1127,6 @@ protected Node(
             // development. Then we can deprecate Getter and Setter for IndexingPressureService in ClusterService (#478).
             clusterService.setIndexingPressureService(indexingPressureService);
 
-            final TaskResourceTrackingService taskResourceTrackingService = new TaskResourceTrackingService(
-                settings,
-                clusterService.getClusterSettings(),
-                threadPool
-            );
-
             final SearchBackpressureSettings searchBackpressureSettings = new SearchBackpressureSettings(
                 settings,
                 clusterService.getClusterSettings()