Add primary first calls for different monitor types (#1205)

Signed-off-by: Ashish Agrawal <ashisagr@amazon.com> (cherry picked from commit 2197d46) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
opensearch-project · Oct 2, 2023 · 418c6c7 · 418c6c7
1 parent 0951c26
commit 418c6c7
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt
@@ -30,6 +30,7 @@ import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.client.Client
 import org.opensearch.client.node.NodeClient
 import org.opensearch.cluster.metadata.IndexMetadata
+import org.opensearch.cluster.routing.Preference
 import org.opensearch.cluster.routing.ShardRouting
 import org.opensearch.cluster.service.ClusterService
 import org.opensearch.common.xcontent.XContentFactory
@@ -642,6 +643,7 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                     .query(boolQueryBuilder)
                     .size(10000) // fixme: make this configurable.
             )
+            .preference(Preference.PRIMARY_FIRST.type())
         val response: SearchResponse = monitorCtx.client!!.suspendUntil { monitorCtx.client!!.search(request, it) }
         if (response.status() !== RestStatus.OK) {
             throw IOException("Failed to search shard: $shard")
@@ -674,7 +676,7 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                 OpenSearchStatusException(message, RestStatus.INTERNAL_SERVER_ERROR)
             )
         }
-        val searchRequest = SearchRequest(queryIndex)
+        val searchRequest = SearchRequest(queryIndex).preference(Preference.PRIMARY_FIRST.type())
         val searchSourceBuilder = SearchSourceBuilder()
         searchSourceBuilder.query(boolQueryBuilder)
         searchRequest.source(searchSourceBuilder)

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt
@@ -19,6 +19,7 @@ import org.opensearch.alerting.util.clusterMetricsMonitorHelpers.toMap
 import org.opensearch.alerting.util.getRoleFilterEnabled
 import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.client.Client
+import org.opensearch.cluster.routing.Preference
 import org.opensearch.cluster.service.ClusterService
 import org.opensearch.common.io.stream.BytesStreamOutput
 import org.opensearch.common.settings.Settings
@@ -99,7 +100,9 @@ class InputService(
                             .newInstance(searchParams)
                             .execute()
 
-                        val searchRequest = SearchRequest().indices(*input.indices.toTypedArray())
+                        val searchRequest = SearchRequest()
+                            .indices(*input.indices.toTypedArray())
+                            .preference(Preference.PRIMARY_FIRST.type())
                         XContentType.JSON.xContent().createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, searchSource).use {
                             searchRequest.source(SearchSourceBuilder.fromXContent(it))
                         }
@@ -191,7 +194,9 @@ class InputService(
                 .newInstance(searchParams)
                 .execute()
 
-            val searchRequest = SearchRequest().indices(*input.indices.toTypedArray())
+            val searchRequest = SearchRequest()
+                .indices(*input.indices.toTypedArray())
+                .preference(Preference.PRIMARY_FIRST.type())
             XContentType.JSON.xContent().createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, searchSource).use {
                 searchRequest.source(SearchSourceBuilder.fromXContent(it))
             }