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

Signed-off-by: Siddhant Deshmukh <[email protected]>

* [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]>
(cherry picked from commit 8ff3bcc)

Author: 3 people

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))

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());
+    }
+}

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()