Skip to content

Commit daacaee

Browse files
lawofcyclesdimorportheca47
lawofcycles
and
dimorportheca47
committed
Add support for Amazon OpenSearch Serverless
Co-authored-by: fukamishuhei <dimorportheca.0407@gmail.com> Signed-off-by: Sotaro Hikita <bering1814@gmail.com>
1 parent 1fc312e commit daacaee

10 files changed

Lines changed: 399 additions & 62 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
33

44
## [Unreleased]
55
### Added
6+
- Add support for Amazon OpenSearch Serverless ([#586](https://github.com/opensearch-project/opensearch-hadoop/pull/586))
67

78
### Changed
89
- Switched to more reliable OpenSearch Lucene snapshot location ([#597](https://github.com/opensearch-project/opensearch-hadoop/pull/597))

‎mr/src/main/java/org/opensearch/hadoop/cfg/ConfigurationOptions.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,10 @@ public interface ConfigurationOptions {
7575
String OPENSEARCH_NODES_WAN_ONLY = "opensearch.nodes.wan.only";
7676
String OPENSEARCH_NODES_WAN_ONLY_DEFAULT = "false";
7777

78+
/** Serverless mode */
79+
String OPENSEARCH_SERVERLESS = "opensearch.serverless";
80+
String OPENSEARCH_SERVERLESS_DEFAULT = "false";
81+
7882
String OPENSEARCH_NODES_RESOLVE_HOST_NAME = "opensearch.nodes.resolve.hostname";
7983

8084
/** Secure Settings Keystore */
@@ -345,4 +349,5 @@ public interface ConfigurationOptions {
345349

346350
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME = "opensearch.aws.sigv4.service.name";
347351
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT = "es";
348-
}
352+
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS = "aoss";
353+
}

‎mr/src/main/java/org/opensearch/hadoop/cfg/Settings.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,10 @@ public boolean getNodesWANOnly() {
168168
return Booleans.parseBoolean(getProperty(OPENSEARCH_NODES_WAN_ONLY, OPENSEARCH_NODES_WAN_ONLY_DEFAULT));
169169
}
170170

171+
public boolean getServerlessMode() {
172+
return Booleans.parseBoolean(getProperty(OPENSEARCH_SERVERLESS, OPENSEARCH_SERVERLESS_DEFAULT));
173+
}
174+
171175
public long getHttpTimeout() {
172176
return TimeValue.parseTimeValue(getProperty(OPENSEARCH_HTTP_TIMEOUT, OPENSEARCH_HTTP_TIMEOUT_DEFAULT)).getMillis();
173177
}
@@ -616,6 +620,11 @@ public Settings setPort(int port) {
616620
return this;
617621
}
618622

623+
public Settings setServerlessMode(boolean serverless) {
624+
setProperty(OPENSEARCH_SERVERLESS, Boolean.toString(serverless));
625+
return this;
626+
}
627+
619628
public Settings setResourceRead(String index) {
620629
setProperty(OPENSEARCH_RESOURCE_READ, index);
621630
return this;
@@ -814,6 +823,10 @@ public String getAwsSigV4Region() {
814823
}
815824

816825
public String getAwsSigV4ServiceName() {
817-
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
826+
if (getServerlessMode()) {
827+
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS);
828+
} else {
829+
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
830+
}
818831
}
819-
}
832+
}

‎mr/src/main/java/org/opensearch/hadoop/rest/InitializationUtils.java‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -85,13 +85,19 @@ public static void checkIndexStatus(Settings settings) {
8585

8686
try {
8787
if (bootstrap.indexExists(readResource.index())) {
88-
RestClient.Health status = bootstrap.getHealth(readResource.index());
89-
if (status == RestClient.Health.RED) {
90-
throw new OpenSearchHadoopIllegalStateException("Index specified [" + readResource.index()
91-
+ "] is either red or " +
92-
"includes an index that is red, and thus all requested data cannot be safely and fully loaded. "
93-
+
94-
"Bailing out...");
88+
if (settings.getServerlessMode()) {
89+
if (LOG.isDebugEnabled()) {
90+
LOG.debug("Serverless mode - skipping health check (not supported in serverless)");
91+
}
92+
} else {
93+
RestClient.Health status = bootstrap.getHealth(readResource.index());
94+
if (status == RestClient.Health.RED) {
95+
throw new OpenSearchHadoopIllegalStateException("Index specified [" + readResource.index()
96+
+ "] is either red or " +
97+
"includes an index that is red, and thus all requested data cannot be safely and fully loaded. "
98+
+
99+
"Bailing out...");
100+
}
95101
}
96102
}
97103
} finally {
@@ -339,9 +345,11 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
339345
try {
340346
mainInfo = bootstrap.mainInfo();
341347
if (log.isDebugEnabled()) {
348+
// Handle serverless mode where UUID might be null
349+
String uuid = mainInfo.getClusterName().getUUID() != null ? mainInfo.getClusterName().getUUID() : "N/A";
342350
log.debug(String.format("Discovered OpenSearch cluster [%s/%s], version [%s]",
343351
mainInfo.getClusterName().getName(),
344-
mainInfo.getClusterName().getUUID(),
352+
uuid,
345353
mainInfo.getMajorVersion()));
346354
}
347355
} catch (OpenSearchHadoopException ex) {
@@ -364,7 +372,8 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
364372
mainInfo.getClusterName().getName(),
365373
clusterName));
366374
}
367-
if (mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
375+
if (mainInfo.getClusterName().getUUID() != null &&
376+
mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
368377
log.warn(String.format(
369378
"Discovered incorrect cluster UUID in settings. Expected [%s] but received [%s]; replacing...",
370379
mainInfo.getClusterName().getUUID(),
@@ -556,4 +565,4 @@ public static boolean setUserProviderIfNotSet(Settings settings, Class<? extends
556565
}
557566
return false;
558567
}
559-
}
568+
}

‎mr/src/main/java/org/opensearch/hadoop/rest/NetworkClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,4 +209,4 @@ public String currentNode() {
209209
public String toString() {
210210
return settings.toString();
211211
}
212-
}
212+
}

‎mr/src/main/java/org/opensearch/hadoop/rest/RestClient.java‎

Lines changed: 44 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ public class RestClient implements Closeable, StatsAware {
9595
private final HttpRetryPolicy retryPolicy;
9696
final ClusterInfo clusterInfo;
9797
private final ErrorExtractor errorExtractor;
98+
private final Settings settings;
9899

99100
{
100101
mapper = new ObjectMapper();
@@ -116,6 +117,7 @@ public RestClient(Settings settings) {
116117
this.network = networkClient;
117118
this.scrollKeepAlive = TimeValue.timeValueMillis(settings.getScrollKeepAlive());
118119
this.indexReadMissingAsEmpty = settings.getIndexReadMissingAsEmpty();
120+
this.settings = settings;
119121

120122
String retryPolicyName = settings.getBatchWriteRetryPolicy();
121123

@@ -296,6 +298,13 @@ public String postDocument(Resource resource, BytesArray document) throws IOExce
296298
}
297299

298300
public void refresh(Resource resource) {
301+
// Skip refresh operation for serverless mode as _refresh endpoint is not supported
302+
if (settings.getServerlessMode()) {
303+
if (LOG.isDebugEnabled()) {
304+
LOG.debug("Serverless mode - skipping refresh operation (not supported in serverless)");
305+
}
306+
return;
307+
}
299308
execute(POST, resource.refresh());
300309
}
301310

@@ -717,11 +726,26 @@ public boolean cancelToken(OpenSearchToken tokenToCancel) {
717726
}
718727

719728
public ClusterInfo mainInfo() {
720-
Response response = execute(GET, "", true);
721-
Map<String, Object> result = parseContent(response.body(), null);
722-
if (result == null) {
723-
throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
729+
// For serverless mode, return dummy info without making root request
730+
if (this.settings.getServerlessMode()) {
731+
// Use a dummy UUID instead of null to avoid NPE in validation
732+
ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
733+
return new ClusterInfo(clusterName, OpenSearchMajorVersion.V_2_X);
734+
}
735+
736+
// Check for cached serverless cluster info
737+
if (clusterInfo != null && clusterInfo.getMajorVersion() != null && "serverless-collection".equals(clusterInfo.getClusterName().getName())) {
738+
// already detected as serverless
739+
return clusterInfo;
724740
}
741+
742+
// Standard mode - retrieve information from the cluster
743+
try {
744+
Response response = execute(GET, "", true);
745+
Map<String, Object> result = parseContent(response.body(), null);
746+
if (result == null) {
747+
throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
748+
}
725749
String clusterName = result.get("cluster_name").toString();
726750
String clusterUUID = (String) result.get("cluster_uuid");
727751
@SuppressWarnings("unchecked")
@@ -736,6 +760,13 @@ public ClusterInfo mainInfo() {
736760
"Version is lower than minimum required version [" + OpenSearchMajorVersion.V_1_X + "].");
737761
}
738762
return new ClusterInfo(new ClusterName(clusterName, clusterUUID), OpenSearchMajorVersion.parse(versionNumber));
763+
} catch (Exception e) {
764+
// If unable to get cluster info in normal way, fallback to serverless mode
765+
LOG.debug("Error getting cluster info, falling back to serverless mode", e);
766+
// Use a dummy UUID instead of null to avoid NPE in validation
767+
ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
768+
return new ClusterInfo(clusterName, OpenSearchMajorVersion.LATEST);
769+
}
739770
}
740771

741772
/**
@@ -758,6 +789,15 @@ public Health getHealth(String index) {
758789
}
759790

760791
public boolean waitForHealth(String index, Health health, TimeValue timeout) {
792+
// Skip health check for serverless mode as _cluster/health endpoint is not supported
793+
if (settings.getServerlessMode()) {
794+
if (LOG.isDebugEnabled()) {
795+
LOG.debug("Serverless mode - skipping health check (not supported in serverless)");
796+
}
797+
// Return false to indicate no timeout
798+
return false;
799+
}
800+
761801
StringBuilder sb = new StringBuilder("/_cluster/health/");
762802
sb.append(index);
763803
sb.append("?wait_for_status=");

‎mr/src/main/java/org/opensearch/hadoop/rest/RestRepository.java‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -377,18 +377,20 @@ public boolean touch() {
377377
}
378378

379379
public void delete() {
380-
// try first a blind delete by query
381-
try {
382-
Resource res = resources.getResourceWrite();
383-
client.deleteByQuery(
384-
res.isTyped()
385-
? res.index() + "/" + res.type()
386-
: res.index(),
387-
MatchAllQueryBuilder.MATCH_ALL);
388-
} catch (OpenSearchHadoopInvalidRequest ehir) {
389-
log.error("Delete by query was not successful...", ehir);
380+
if (!this.settings.getServerlessMode()) {
381+
// try first a blind delete by query
382+
try {
383+
Resource res = resources.getResourceWrite();
384+
client.deleteByQuery(
385+
res.isTyped()
386+
? res.index() + "/" + res.type()
387+
: res.index(),
388+
MatchAllQueryBuilder.MATCH_ALL);
389+
} catch (OpenSearchHadoopInvalidRequest ehir) {
390+
log.error("Delete by query was not successful...", ehir);
391+
}
390392
}
391-
393+
392394
// in ES 2.0 and higher this means scrolling and deleting the docs by hand...
393395
// do a scroll-scan without source
394396

@@ -475,6 +477,8 @@ public long count(boolean read) {
475477
}
476478

477479
public boolean waitForYellow() {
480+
// For serverless collections, waitForHealth is handled through the RestClient.waitForHealth()
481+
// which will return appropriate result based on serverless mode
478482
return client.waitForHealth(resources.getResourceWrite().index(), RestClient.Health.YELLOW, TimeValue.timeValueSeconds(10));
479483
}
480484

@@ -495,4 +499,4 @@ public Stats stats() {
495499
public Settings getSettings() {
496500
return settings;
497501
}
498-
}
502+
}

0 commit comments

Comments
 (0)