Skip to content

Commit

Permalink
Create listener to refresh search thread resource usage (opensearch-p…
Browse files Browse the repository at this point in the history
…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]>
  • Loading branch information
3 people authored Jul 23, 2024
1 parent c82a282 commit 8ff3bcc
Show file tree
Hide file tree
Showing 3 changed files with 42 additions and 7 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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());
}
}
18 changes: 11 additions & 7 deletions server/src/main/java/org/opensearch/node/Node.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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()
Expand Down

0 comments on commit 8ff3bcc

Please sign in to comment.