Add ActionListener onFailure to ExtensionsRunner (opensearch-project#87)

* Add ActionListener onFailure to ExtensionsRunner Signed-off-by: Ryan Bogan <rbogan@amazon.com> * Addressed PR Comments Signed-off-by: Ryan Bogan <rbogan@amazon.com> * Addressed PR Comments Signed-off-by: Ryan Bogan <rbogan@amazon.com> * Addressed PR Comments Signed-off-by: Ryan Bogan <rbogan@amazon.com> * Removed test failure Signed-off-by: Ryan Bogan <rbogan@amazon.com> Signed-off-by: Ryan Bogan <rbogan@amazon.com>
kokibas · Mar 17, 2023 · 527de09 · 527de09
1 parent 337166f
commit 527de09
Show file tree

Hide file tree

Showing 3 changed files with 82 additions and 1 deletion.
diff --git a/src/main/java/org/opensearch/sdk/ExtensionsRunner.java b/src/main/java/org/opensearch/sdk/ExtensionsRunner.java
@@ -41,6 +41,7 @@
 import org.opensearch.rest.RestResponse;
 import org.opensearch.transport.netty4.Netty4Transport;
 import org.opensearch.transport.SharedGroupFactory;
+import org.opensearch.sdk.handlers.ActionListenerOnFailureResponseHandler;
 import org.opensearch.sdk.handlers.ClusterSettingsResponseHandler;
 import org.opensearch.sdk.handlers.ClusterStateResponseHandler;
 import org.opensearch.sdk.handlers.LocalNodeResponseHandler;
@@ -475,6 +476,30 @@ public void sendLocalNodeRequest(TransportService transportService) {
         }
     }
 
+    /**
+     * Requests the ActionListener onFailure method to be run by OpenSearch.  The result will be handled by a {@link ActionListenerOnFailureResponseHandler}.
+     *
+     * @param transportService  The TransportService defining the connection to OpenSearch.
+     * @param failureException The exception to be sent to OpenSearch
+     */
+    public void sendActionListenerOnFailureRequest(TransportService transportService, Exception failureException) {
+        logger.info("Sending ActionListener onFailure request to OpenSearch");
+        ActionListenerOnFailureResponseHandler listenerHandler = new ActionListenerOnFailureResponseHandler();
+        try {
+            transportService.sendRequest(
+                opensearchNode,
+                ExtensionsOrchestrator.REQUEST_EXTENSION_ACTION_LISTENER_ON_FAILURE,
+                new ExtensionRequest(
+                    ExtensionsOrchestrator.RequestType.REQUEST_EXTENSION_ACTION_LISTENER_ON_FAILURE,
+                    failureException.toString()
+                ),
+                listenerHandler
+            );
+        } catch (Exception e) {
+            logger.info("Failed to send ActionListener onFailure request to OpenSearch", e);
+        }
+    }
+
     private Settings getSettings() {
         return settings;
     }
@@ -484,7 +509,7 @@ private Settings getSettings() {
      *
      * @param timeout  The timeout for the listener in milliseconds. A timeout of 0 means no timeout.
      */
-    public void startActionListener(int timeout) {
+    public static void startActionListener(int timeout) {
         final ActionListener actionListener = new ActionListener();
         actionListener.runActionListener(true, timeout);
     }

diff --git a/src/main/java/org/opensearch/sdk/handlers/ActionListenerOnFailureResponseHandler.java b/src/main/java/org/opensearch/sdk/handlers/ActionListenerOnFailureResponseHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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.sdk.handlers;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.extensions.ExtensionBooleanResponse;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.sdk.ExtensionsRunner;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.TransportException;
+import org.opensearch.transport.TransportResponseHandler;
+import org.opensearch.transport.TransportService;
+
+import java.io.IOException;
+
+/**
+ * This class handles the response from OpenSearch to a {@link ExtensionsRunner#sendActionListenerOnFailureRequest(TransportService, Exception)} call.
+ */
+public class ActionListenerOnFailureResponseHandler implements TransportResponseHandler<ExtensionBooleanResponse> {
+    private static final Logger logger = LogManager.getLogger(ActionListenerOnFailureResponseHandler.class);
+
+    @Override
+    public void handleResponse(ExtensionBooleanResponse response) {
+        logger.info("received {}", response);
+    }
+
+    @Override
+    public void handleException(TransportException exp) {
+        logger.info("ActionListenerOnFailureRequest failed", exp);
+    }
+
+    @Override
+    public String executor() {
+        return ThreadPool.Names.GENERIC;
+    }
+
+    @Override
+    public ExtensionBooleanResponse read(StreamInput in) throws IOException {
+        return new ExtensionBooleanResponse(in);
+    }
+}
diff --git a/src/test/java/org/opensearch/sdk/TestExtensionsRunner.java b/src/test/java/org/opensearch/sdk/TestExtensionsRunner.java
@@ -45,6 +45,7 @@
 import org.opensearch.rest.BytesRestResponse;
 import org.opensearch.rest.RestRequest.Method;
 import org.opensearch.rest.RestStatus;
+import org.opensearch.sdk.handlers.ActionListenerOnFailureResponseHandler;
 import org.opensearch.sdk.handlers.ClusterSettingsResponseHandler;
 import org.opensearch.sdk.handlers.ClusterStateResponseHandler;
 import org.opensearch.sdk.handlers.LocalNodeResponseHandler;
@@ -184,6 +185,14 @@ public void testLocalNodeRequest() {
         verify(transportService, times(1)).sendRequest(any(), anyString(), any(), any(LocalNodeResponseHandler.class));
     }
 
+    @Test
+    public void testActionListenerOnFailureRequest() {
+
+        extensionsRunner.sendActionListenerOnFailureRequest(transportService, new Exception("Test failure"));
+
+        verify(transportService, times(1)).sendRequest(any(), anyString(), any(), any(ActionListenerOnFailureResponseHandler.class));
+    }
+
     @Test
     public void testRegisterRestActionsRequest() {