Add search backpressure cancellation at the coordinator level

CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add getter for path field in NestedQueryBuilder ([#4636](https://github.com/opensearch-project/OpenSearch/pull/4636))
 - Allow mmap to use new JDK-19 preview APIs in Apache Lucene 9.4+ ([#5151](https://github.com/opensearch-project/OpenSearch/pull/5151))
 - Add support for ppc64le architecture ([#5459](https://github.com/opensearch-project/OpenSearch/pull/5459))
+- Cancellation of in-flight SearchTasks based on resource consumption ([#5606](https://github.com/opensearch-project/OpenSearch/pull/5605))
 
 ### Dependencies
 - Bumps `log4j-core` from 2.18.0 to 2.19.0

server/src/main/java/org/opensearch/action/search/SearchShardTask.java

@@ -36,6 +36,7 @@
 import org.opensearch.search.fetch.ShardFetchSearchRequest;
 import org.opensearch.search.internal.ShardSearchRequest;
 import org.opensearch.tasks.CancellableTask;
+import org.opensearch.tasks.SearchBackpressureTask;
 import org.opensearch.tasks.TaskId;
 
 import java.util.Map;
@@ -47,7 +48,7 @@
  *
  * @opensearch.internal
  */
-public class SearchShardTask extends CancellableTask {
+public class SearchShardTask extends CancellableTask implements SearchBackpressureTask {
     // generating metadata in a lazy way since source can be quite big
     private final MemoizedSupplier<String> metadataSupplier;
 

server/src/main/java/org/opensearch/action/search/SearchTask.java

@@ -34,6 +34,7 @@
 
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.tasks.CancellableTask;
+import org.opensearch.tasks.SearchBackpressureTask;
 import org.opensearch.tasks.TaskId;
 
 import java.util.Map;
@@ -46,7 +47,7 @@
  *
  * @opensearch.internal
  */
-public class SearchTask extends CancellableTask {
+public class SearchTask extends CancellableTask implements SearchBackpressureTask {
     // generating description in a lazy way since source can be quite big
     private final Supplier<String> descriptionSupplier;
     private SearchProgressListener progressListener = SearchProgressListener.NOOP;

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -44,9 +44,7 @@
 import org.opensearch.search.backpressure.settings.NodeDuressSettings;
 import org.opensearch.search.backpressure.settings.SearchBackpressureSettings;
 import org.opensearch.search.backpressure.settings.SearchShardTaskSettings;
-import org.opensearch.search.backpressure.trackers.CpuUsageTracker;
-import org.opensearch.search.backpressure.trackers.ElapsedTimeTracker;
-import org.opensearch.search.backpressure.trackers.HeapUsageTracker;
+import org.opensearch.search.backpressure.settings.SearchTaskSettings;
 import org.opensearch.tasks.TaskManager;
 import org.opensearch.tasks.TaskResourceTrackingService;
 import org.opensearch.watcher.ResourceWatcherService;
@@ -599,18 +597,31 @@ public void apply(Settings value, Settings current, Settings previous) {
 
                 // Settings related to search backpressure
                 SearchBackpressureSettings.SETTING_MODE,
-                SearchBackpressureSettings.SETTING_CANCELLATION_RATIO,
-                SearchBackpressureSettings.SETTING_CANCELLATION_RATE,
-                SearchBackpressureSettings.SETTING_CANCELLATION_BURST,
+
                 NodeDuressSettings.SETTING_NUM_SUCCESSIVE_BREACHES,
                 NodeDuressSettings.SETTING_CPU_THRESHOLD,
                 NodeDuressSettings.SETTING_HEAP_THRESHOLD,
+                SearchTaskSettings.SETTING_CANCELLATION_RATIO,
+                SearchTaskSettings.SETTING_CANCELLATION_RATE,
+                SearchTaskSettings.SETTING_CANCELLATION_BURST,
+                SearchTaskSettings.SETTING_HEAP_PERCENT_THRESHOLD,
+                SearchTaskSettings.SETTING_HEAP_VARIANCE_THRESHOLD,
+                SearchTaskSettings.SETTING_HEAP_MOVING_AVERAGE_WINDOW_SIZE,
+                SearchTaskSettings.SETTING_CPU_TIME_MILLIS_THRESHOLD,
+                SearchTaskSettings.SETTING_ELAPSED_TIME_MILLIS_THRESHOLD,
+                SearchTaskSettings.SETTING_TOTAL_HEAP_PERCENT_THRESHOLD,
+                SearchShardTaskSettings.SETTING_CANCELLATION_RATIO,
+                SearchShardTaskSettings.SETTING_CANCELLATION_RATE,
+                SearchShardTaskSettings.SETTING_CANCELLATION_BURST,
+                SearchShardTaskSettings.SETTING_HEAP_PERCENT_THRESHOLD,
+                SearchShardTaskSettings.SETTING_HEAP_VARIANCE_THRESHOLD,
+                SearchShardTaskSettings.SETTING_HEAP_MOVING_AVERAGE_WINDOW_SIZE,
+                SearchShardTaskSettings.SETTING_CPU_TIME_MILLIS_THRESHOLD,
+                SearchShardTaskSettings.SETTING_ELAPSED_TIME_MILLIS_THRESHOLD,
                 SearchShardTaskSettings.SETTING_TOTAL_HEAP_PERCENT_THRESHOLD,
-                HeapUsageTracker.SETTING_HEAP_PERCENT_THRESHOLD,
-                HeapUsageTracker.SETTING_HEAP_VARIANCE_THRESHOLD,
-                HeapUsageTracker.SETTING_HEAP_MOVING_AVERAGE_WINDOW_SIZE,
-                CpuUsageTracker.SETTING_CPU_TIME_MILLIS_THRESHOLD,
-                ElapsedTimeTracker.SETTING_ELAPSED_TIME_MILLIS_THRESHOLD
+                SearchBackpressureSettings.SETTING_CANCELLATION_RATIO,  // deprecated
+                SearchBackpressureSettings.SETTING_CANCELLATION_RATE,   // deprecated
+                SearchBackpressureSettings.SETTING_CANCELLATION_BURST   // deprecated
             )
         )
     );

server/src/main/java/org/opensearch/node/Node.java

@@ -867,7 +867,8 @@ protected Node(
             final SearchBackpressureService searchBackpressureService = new SearchBackpressureService(
                 searchBackpressureSettings,
                 taskResourceTrackingService,
-                threadPool
+                threadPool,
+                transportService.getTaskManager()
             );
 
             final RecoverySettings recoverySettings = new RecoverySettings(settings, settingsModule.getClusterSettings());

server/src/main/java/org/opensearch/search/backpressure/CancellationSettingsListener.java

@@ -0,0 +1,21 @@
+/*
+ * 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.search.backpressure;
+
+/**
+ * Listener for callbacks related to cancellation settings
+ */
+public interface CancellationSettingsListener {
+
+    void onRatioChanged(double ratio);
+
+    void onRateChanged(double rate);
+
+    void onBurstChanged(double burst);
+}