Create listener to refresh search thread resource usage (#14832)

Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * [bug fix] fix incorrect coordinator node search resource usages Signed-off-by: Chenyang Ji <cyji@amazon.com> * fix bug on serialization when passing task resource usage to coordinator Signed-off-by: Chenyang Ji <cyji@amazon.com> * add more unit tests Signed-off-by: Chenyang Ji <cyji@amazon.com> * remove query insights plugin related code Signed-off-by: Chenyang Ji <cyji@amazon.com> * create per request listener to refresh task resource usage Signed-off-by: Chenyang Ji <cyji@amazon.com> * Make new listener API public Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Add changelog Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Remove wrong files added Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Address review comments Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Build fix Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Make singleton Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Address review comments Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Make sure listener runs before plugin listeners Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Spotless Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> * Minor fix Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> --------- Signed-off-by: Chenyang Ji <cyji@amazon.com> Signed-off-by: Siddhant Deshmukh <deshsid@amazon.com> Signed-off-by: Jay Deng <jayd0104@gmail.com> Co-authored-by: Chenyang Ji <cyji@amazon.com> Co-authored-by: Jay Deng <jayd0104@gmail.com> (cherry picked from commit 8ff3bcc)
opensearch-project · Jul 23, 2024 · 4466c0f · 4466c0f
1 parent 36dc24a
commit 4466c0f
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 9 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,7 +4,6 @@ All notable changes to this project are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). See the [CONTRIBUTING guide](./CONTRIBUTING.md#Changelog) for instructions on how to add changelog entries.
 
 ## [Unreleased 2.x]
-### Added
 - Add fingerprint ingest processor ([#13724](https://github.com/opensearch-project/OpenSearch/pull/13724))
 - [Remote Store] Rate limiter for remote store low priority uploads ([#14374](https://github.com/opensearch-project/OpenSearch/pull/14374/))
 - Apply the date histogram rewrite optimization to range aggregation ([#13865](https://github.com/opensearch-project/OpenSearch/pull/13865))
@@ -25,7 +24,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add SplitResponseProcessor to Search Pipelines (([#14800](https://github.com/opensearch-project/OpenSearch/issues/14800)))
 - Optimize TransportNodesAction to not send DiscoveryNodes for NodeStats, NodesInfo and ClusterStats call ([14749](https://github.com/opensearch-project/OpenSearch/pull/14749))
 - Reduce logging in DEBUG for MasterService:run ([#14795](https://github.com/opensearch-project/OpenSearch/pull/14795))
-- Refactor remote-routing-table service inline with remote state interfaces([#14668](https://github.com/opensearch-project/OpenSearch/pull/14668))
+- 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
 - Update to Apache Lucene 9.11.1 ([#14042](https://github.com/opensearch-project/OpenSearch/pull/14042), [#14576](https://github.com/opensearch-project/OpenSearch/pull/14576))

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
@@ -51,6 +51,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;
@@ -852,8 +853,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);
@@ -982,7 +992,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)
@@ -1110,12 +1120,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()