From 8ef2f9f4f4c2d3fd7cb5b234f031d46ff6391a03 Mon Sep 17 00:00:00 2001 From: Siddhant Deshmukh Date: Tue, 23 Jul 2024 08:44:51 -0700 Subject: [PATCH] Create listener to refresh search thread resource usage (#14832) (#14900) * [bug fix] fix incorrect coordinator node search resource usages Signed-off-by: Chenyang Ji * fix bug on serialization when passing task resource usage to coordinator Signed-off-by: Chenyang Ji * add more unit tests Signed-off-by: Chenyang Ji * remove query insights plugin related code Signed-off-by: Chenyang Ji * create per request listener to refresh task resource usage Signed-off-by: Chenyang Ji * Make new listener API public Signed-off-by: Siddhant Deshmukh * Add changelog Signed-off-by: Siddhant Deshmukh * Remove wrong files added Signed-off-by: Siddhant Deshmukh * Address review comments Signed-off-by: Siddhant Deshmukh * Build fix Signed-off-by: Siddhant Deshmukh * Make singleton Signed-off-by: Siddhant Deshmukh * Address review comments Signed-off-by: Siddhant Deshmukh * Make sure listener runs before plugin listeners Signed-off-by: Siddhant Deshmukh * Spotless Signed-off-by: Siddhant Deshmukh * Minor fix Signed-off-by: Siddhant Deshmukh --------- Signed-off-by: Chenyang Ji Signed-off-by: Siddhant Deshmukh Signed-off-by: Jay Deng Co-authored-by: Chenyang Ji Co-authored-by: Jay Deng (cherry picked from commit 8ff3bcc4632287d8a784a1cba662957d6f921851) Signed-off-by: Siddhant Deshmukh --- CHANGELOG.md | 1 + .../SearchTaskRequestOperationsListener.java | 30 +++++++++++++++++++ .../main/java/org/opensearch/node/Node.java | 18 ++++++----- 3 files changed, 42 insertions(+), 7 deletions(-) create mode 100644 server/src/main/java/org/opensearch/action/search/SearchTaskRequestOperationsListener.java diff --git a/CHANGELOG.md b/CHANGELOG.md index a2d965541ab67..7490e1612907e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,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 rest, transport layer changes for hot to warm tiering - dedicated setup (([#13980](https://github.com/opensearch-project/OpenSearch/pull/13980)) +- Create listener to refresh search thread resource usage ([#14832](https://github.com/opensearch-project/OpenSearch/pull/14832)) ### 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 new file mode 100644 index 0000000000000..4434d71793b23 --- /dev/null +++ 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 index 81d4ee31c3d0f..e5eea7327b711 100644 --- 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()