Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update SDKRestClient/OpenSearchAsync client with http.port setting from OpenSearch during intialization #788

Merged
merged 3 commits into from
May 26, 2023
Merged

Update SDKRestClient/OpenSearchAsync client with http.port setting from OpenSearch during intialization #788

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
431e8c8
Modifies extension initialization request handling, updates sdkclient…
joshpalis May 26, 2023
52ba43a
Resetting test TransportAddress port to 9300
joshpalis May 26, 2023
28a8591
pulling http.port settings from environment settings response rather …
joshpalis May 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions src/main/java/org/opensearch/sdk/SDKClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,6 @@
import org.opensearch.client.opensearch.OpenSearchClient;
import org.opensearch.client.transport.OpenSearchTransport;
import org.opensearch.client.transport.rest_client.RestClientTransport;
import org.opensearch.common.transport.TransportAddress;
import org.opensearch.index.reindex.BulkByScrollResponse;
import org.opensearch.index.reindex.DeleteByQueryRequest;

Expand Down Expand Up @@ -132,17 +131,26 @@ public void initialize(Map<ActionType, TransportAction> actions) {

/**
* Update the ExtensionSettings with a new OpenSearch address and port
* @param address the TransportAddress associated with the OpenSearchNode
* @param address the host address associated with the OpenSearchNode
* @param httpPort the http port associated with the OpenSearchNOde
*/
public void updateOpenSearchNodeSettings(TransportAddress address) {
public void updateOpenSearchNodeSettings(String address, String httpPort) {
// Update the settings for future initialization of new clients
this.extensionSettings.setOpensearchAddress(address.getAddress());
this.extensionSettings.setOpensearchPort(Integer.toString(address.getPort()));
this.extensionSettings.setOpensearchAddress(address);
this.extensionSettings.setOpensearchPort(httpPort);
// Update the settings on the already-initialized SDKRestClient (Deprecated -- for migration use)
if (this.sdkRestClient != null) {
this.sdkRestClient.getRestHighLevelClient()
.getLowLevelClient()
.setNodes(List.of(new Node(new HttpHost(address.getAddress(), address.getPort()))));
.setNodes(List.of(new Node(new HttpHost(address, Integer.parseInt(httpPort)))));
}
// Update the settings on the already-initialized OpenSearchAsyncClient
if (this.javaAsyncClient != null) {
OpenSearchTransport javaAsyncClientTransport = this.javaAsyncClient._transport();
if (javaAsyncClientTransport instanceof RestClientTransport) {
RestClientTransport restClientTransport = (RestClientTransport) javaAsyncClientTransport;
restClientTransport.restClient().setNodes(List.of(new Node(new HttpHost(address, Integer.parseInt(httpPort)))));
}
}
}

Expand Down
10 changes: 10 additions & 0 deletions src/main/java/org/opensearch/sdk/handlers/ExtensionsInitRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,12 @@
public class ExtensionsInitRequestHandler {
private static final Logger logger = LogManager.getLogger(ExtensionsInitRequestHandler.class);

// The default http port setting of OpenSearch
private static final String DEFAULT_HTTP_PORT = "9200";

// The configured http port setting of opensearch.yml
private static final String HTTP_PORT_SETTING = "http.port";

private final ExtensionsRunner extensionsRunner;

/**
Expand Down Expand Up @@ -79,6 +85,10 @@ public InitializeExtensionResponse handleExtensionInitRequest(InitializeExtensio
extensionsRunner.setEnvironmentSettings(settings);
extensionsRunner.updateNamedXContentRegistry();
extensionsRunner.updateSdkClusterService();
// Use OpenSearch Settings to update client REST Connections
String openSearchNodeAddress = extensionInitRequest.getSourceNode().getAddress().getAddress();
dbwiddis marked this conversation as resolved.
Show resolved Hide resolved
String openSearchNodeHttpPort = settings.get(HTTP_PORT_SETTING) != null ? settings.get(HTTP_PORT_SETTING) : DEFAULT_HTTP_PORT;
extensionsRunner.getSdkClient().updateOpenSearchNodeSettings(openSearchNodeAddress, openSearchNodeHttpPort);

// Last step of initialization
// TODO: make sure all the other sendX methods have completed
Expand Down
48 changes: 34 additions & 14 deletions src/test/java/org/opensearch/sdk/sample/helloworld/TestHelloWorldExtension.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@

package org.opensearch.sdk.sample.helloworld;

import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.List;
import java.util.Map;
Expand All @@ -27,8 +26,9 @@
import org.opensearch.action.ActionType;
import org.opensearch.action.support.TransportAction;
import org.opensearch.client.Node;
import org.opensearch.client.opensearch.OpenSearchAsyncClient;
import org.opensearch.client.transport.rest_client.RestClientTransport;
import org.opensearch.common.settings.Settings;
import org.opensearch.common.transport.TransportAddress;
import org.opensearch.sdk.api.ActionExtension.ActionHandler;
import org.opensearch.sdk.rest.ExtensionRestHandler;
import org.opensearch.sdk.sample.helloworld.transport.SampleAction;
Expand Down Expand Up @@ -56,6 +56,7 @@ public class TestHelloWorldExtension extends OpenSearchTestCase {
private Injector injector;
private SDKClient sdkClient;
private SDKRestClient sdkRestClient;
private OpenSearchAsyncClient javaAsyncClient;
private final ExtensionSettings extensionSettings = new ExtensionSettings("", "", "", "localhost", "9200");

static class UnregisteredAction extends ActionType<SampleResponse> {
Expand Down Expand Up @@ -85,6 +86,7 @@ public void setUp() throws Exception {
});
initializeSdkClient();
this.sdkRestClient = sdkClient.initializeRestClient("localhost", 9200);
this.javaAsyncClient = sdkClient.initalizeJavaAsyncClient("localhost", 9200);
}

@SuppressWarnings("rawtypes")
Expand All @@ -98,6 +100,7 @@ private void initializeSdkClient() {
public void tearDown() throws Exception {
super.tearDown();
this.sdkRestClient.close();
this.sdkClient.doCloseJavaClients();
this.injector = null;
}

Expand All @@ -113,18 +116,35 @@ public void testExtensionSettings() {

@Test
public void testExtensionSettingsUpdate() {
List<Node> nodes = this.sdkClient.getSdkRestClient().getRestHighLevelClient().getLowLevelClient().getNodes();
assertEquals(1, nodes.size());
HttpHost host = nodes.get(0).getHost();
assertEquals("localhost", host.getHostName());
assertEquals(9200, host.getPort());

this.sdkClient.updateOpenSearchNodeSettings(new TransportAddress(new InetSocketAddress("10.10.10.10", 9300)));
nodes = this.sdkClient.getSdkRestClient().getRestHighLevelClient().getLowLevelClient().getNodes();
assertEquals(1, nodes.size());
host = nodes.get(0).getHost();
assertEquals("10.10.10.10", host.getHostName());
assertEquals(9300, host.getPort());
List<Node> sdkRestClientNodes = this.sdkClient.getSdkRestClient().getRestHighLevelClient().getLowLevelClient().getNodes();
List<Node> javaAsyncClientNodes = ((RestClientTransport) this.javaAsyncClient._transport()).restClient().getNodes();

// Test rest client nodes
assertEquals(1, sdkRestClientNodes.size());
assertEquals(1, javaAsyncClientNodes.size());

// Test client http hosts
HttpHost sdkRestClientHost = sdkRestClientNodes.get(0).getHost();
assertEquals("localhost", sdkRestClientHost.getHostName());
assertEquals(9200, sdkRestClientHost.getPort());
HttpHost javaAsyncClientHost = javaAsyncClientNodes.get(0).getHost();
assertEquals("localhost", javaAsyncClientHost.getHostName());
assertEquals(9200, javaAsyncClientHost.getPort());

// Test updated hosts
this.sdkClient.updateOpenSearchNodeSettings("10.10.10.10", "9204");

sdkRestClientNodes = this.sdkClient.getSdkRestClient().getRestHighLevelClient().getLowLevelClient().getNodes();
assertEquals(1, sdkRestClientNodes.size());
sdkRestClientHost = sdkRestClientNodes.get(0).getHost();
assertEquals("10.10.10.10", sdkRestClientHost.getHostName());
assertEquals(9204, sdkRestClientHost.getPort());

javaAsyncClientNodes = ((RestClientTransport) this.javaAsyncClient._transport()).restClient().getNodes();
assertEquals(1, javaAsyncClientNodes.size());
javaAsyncClientHost = javaAsyncClientNodes.get(0).getHost();
assertEquals("10.10.10.10", javaAsyncClientHost.getHostName());
assertEquals(9204, javaAsyncClientHost.getPort());
}

@Test
Expand Down