-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Execute remote actions on another extension (#588)
* Add ProxyAction with TransportAction and handlers Signed-off-by: Daniel Widdis <widdis@gmail.com> * Give SDKActionModule a copy of ExtensionsRunner to use with transport Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add new ProxyActionRequest Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add SDKTransportService wrapper accessible to actions Signed-off-by: Daniel Widdis <widdis@gmail.com> * Implement ProxyTransportAction Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add test case to HelloWorldExtension Signed-off-by: Daniel Widdis <widdis@gmail.com> * Better naming of ExtensionActionResponse and correct action name Signed-off-by: Daniel Widdis <widdis@gmail.com> * Refactoring with TransportService and latest OpenSearch PR updates Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add ExtensionsActionRequestHandler Signed-off-by: Daniel Widdis <widdis@gmail.com> * Instantiate Proxy Action Request Signed-off-by: Daniel Widdis <widdis@gmail.com> * Working test case! Signed-off-by: Daniel Widdis <widdis@gmail.com> * Properly parse returned byte array into a response Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add sequence diagram to DESIGN.md Signed-off-by: Daniel Widdis <widdis@gmail.com> * Typoo fix Signed-off-by: Daniel Widdis <widdis@gmail.com> * Update with latest changes on companion PR Signed-off-by: Daniel Widdis <widdis@gmail.com> * Rename ProxyFoo to RemoteExtensionFoo Signed-off-by: Daniel Widdis <widdis@gmail.com> * Better handling of response bytes Signed-off-by: Daniel Widdis <widdis@gmail.com> * Handle plugin remote action requests Signed-off-by: Daniel Widdis <widdis@gmail.com> * Address code review comments Signed-off-by: Daniel Widdis <widdis@gmail.com> * Update sequence diagram Signed-off-by: Daniel Widdis <widdis@gmail.com> --------- Signed-off-by: Daniel Widdis <widdis@gmail.com> (cherry picked from commit ebc684a) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
- Loading branch information
1 parent
4ce9cc9
commit 72841ee
Showing
22 changed files
with
1,017 additions
and
132 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
132 changes: 132 additions & 0 deletions
132
src/main/java/org/opensearch/sdk/SDKTransportService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * 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; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.concurrent.TimeoutException; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.opensearch.cluster.node.DiscoveryNode; | ||
| import org.opensearch.extensions.ExtensionsManager; | ||
| import org.opensearch.extensions.action.RegisterTransportActionsRequest; | ||
| import org.opensearch.extensions.action.RemoteExtensionActionResponse; | ||
| import org.opensearch.extensions.action.TransportActionRequestFromExtension; | ||
| import org.opensearch.sdk.ActionExtension.ActionHandler; | ||
| import org.opensearch.sdk.action.RemoteExtensionActionRequest; | ||
| import org.opensearch.sdk.action.SDKActionModule; | ||
| import org.opensearch.sdk.handlers.AcknowledgedResponseHandler; | ||
| import org.opensearch.sdk.handlers.ExtensionActionResponseHandler; | ||
| import org.opensearch.transport.TransportService; | ||
|
|
||
| /** | ||
| * Wrapper class for {@link TransportService} and associated methods. | ||
| * | ||
| * TODO: Move all the sendFooRequest() methods here | ||
| * TODO: Replace usages of getExtensionTransportService with this class | ||
| * https://github.com/opensearch-project/opensearch-sdk-java/issues/585 | ||
| */ | ||
| public class SDKTransportService { | ||
| private final Logger logger = LogManager.getLogger(SDKTransportService.class); | ||
|
|
||
| private TransportService transportService; | ||
| private DiscoveryNode opensearchNode; | ||
| private String uniqueId; | ||
|
|
||
| /** | ||
| * Requests that OpenSearch register the Transport Actions for this extension. | ||
| * | ||
| * @param actions The map of registered actions from {@link SDKActionModule#getActions()} | ||
| */ | ||
| public void sendRegisterTransportActionsRequest(Map<String, ActionHandler<?, ?>> actions) { | ||
| logger.info("Sending Register Transport Actions request to OpenSearch"); | ||
| Set<String> actionNameSet = actions.values() | ||
| .stream() | ||
| .filter(h -> !h.getAction().name().startsWith("internal")) | ||
| .map(h -> h.getAction().getClass().getName()) | ||
| .collect(Collectors.toSet()); | ||
| AcknowledgedResponseHandler registerTransportActionsResponseHandler = new AcknowledgedResponseHandler(); | ||
| try { | ||
| transportService.sendRequest( | ||
| opensearchNode, | ||
| ExtensionsManager.REQUEST_EXTENSION_REGISTER_TRANSPORT_ACTIONS, | ||
| new RegisterTransportActionsRequest(uniqueId, actionNameSet), | ||
| registerTransportActionsResponseHandler | ||
| ); | ||
| } catch (Exception e) { | ||
| logger.error("Failed to send Register Transport Actions request to OpenSearch", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Requests that OpenSearch execute a Transport Actions on another extension. | ||
| * | ||
| * @param request The request to send | ||
| * @return A buffer serializing the response from the remote action if successful, otherwise null | ||
| */ | ||
| public RemoteExtensionActionResponse sendRemoteExtensionActionRequest(RemoteExtensionActionRequest request) { | ||
| logger.info("Sending Remote Extension Action request to OpenSearch for [" + request.getAction() + "]"); | ||
| // Combine class name string and request bytes | ||
| byte[] requestClassBytes = request.getRequestClass().getBytes(StandardCharsets.UTF_8); | ||
| byte[] proxyRequestBytes = ByteBuffer.allocate(requestClassBytes.length + 1 + request.getRequestBytes().length) | ||
| .put(requestClassBytes) | ||
| .put(RemoteExtensionActionRequest.UNIT_SEPARATOR) | ||
| .put(request.getRequestBytes()) | ||
| .array(); | ||
| ExtensionActionResponseHandler extensionActionResponseHandler = new ExtensionActionResponseHandler(); | ||
| try { | ||
| transportService.sendRequest( | ||
| opensearchNode, | ||
| ExtensionsManager.TRANSPORT_ACTION_REQUEST_FROM_EXTENSION, | ||
| new TransportActionRequestFromExtension(request.getAction(), proxyRequestBytes, uniqueId), | ||
| extensionActionResponseHandler | ||
| ); | ||
| // Wait on response | ||
| extensionActionResponseHandler.awaitResponse(); | ||
| } catch (TimeoutException e) { | ||
| logger.error("Failed to receive Remote Extension Action response from OpenSearch", e); | ||
| } catch (Exception e) { | ||
| logger.error("Failed to send Remote Extension Action request to OpenSearch", e); | ||
| } | ||
| // At this point, response handler has read in the response bytes | ||
| return new RemoteExtensionActionResponse( | ||
| extensionActionResponseHandler.isSuccess(), | ||
| extensionActionResponseHandler.getResponseBytes() | ||
| ); | ||
| } | ||
|
|
||
| public TransportService getTransportService() { | ||
| return transportService; | ||
| } | ||
|
|
||
| public DiscoveryNode getOpensearchNode() { | ||
| return opensearchNode; | ||
| } | ||
|
|
||
| public String getUniqueId() { | ||
| return uniqueId; | ||
| } | ||
|
|
||
| public void setTransportService(TransportService transportService) { | ||
| this.transportService = transportService; | ||
| } | ||
|
|
||
| public void setOpensearchNode(DiscoveryNode opensearchNode) { | ||
| this.opensearchNode = opensearchNode; | ||
| } | ||
|
|
||
| public void setUniqueId(String uniqueId) { | ||
| this.uniqueId = uniqueId; | ||
| } | ||
| } |
32 changes: 32 additions & 0 deletions
32
src/main/java/org/opensearch/sdk/action/RemoteExtensionAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * 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.action; | ||
|
|
||
| import org.opensearch.action.ActionType; | ||
| import org.opensearch.extensions.action.RemoteExtensionActionResponse; | ||
|
|
||
| /** | ||
| * The {@link ActionType} used as they key for the {@link RemoteExtensionTransportAction}. | ||
| */ | ||
| public class RemoteExtensionAction extends ActionType<RemoteExtensionActionResponse> { | ||
|
|
||
| /** | ||
| * The name to look up this action with | ||
| */ | ||
| public static final String NAME = "internal:remote-extension-action"; | ||
| /** | ||
| * The singleton instance of this class | ||
| */ | ||
| public static final RemoteExtensionAction INSTANCE = new RemoteExtensionAction(); | ||
|
|
||
| private RemoteExtensionAction() { | ||
| super(NAME, RemoteExtensionActionResponse::new); | ||
| } | ||
| } |
Oops, something went wrong.