Skip to content

Commit 9807b25

Browse files
lawofcyclesdimorportheca47harshavamsi
authored
Add support for Amazon OpenSearch Serverless (#586)
* Add support for Amazon OpenSearch Serverless Co-authored-by: fukamishuhei <dimorportheca.0407@gmail.com> Signed-off-by: Sotaro Hikita <bering1814@gmail.com> * Replace Scroll API with Search After API for read pagination in serverless mode Signed-off-by: Sotaro Hikita <bering1814@gmail.com> * Fix testNoScrollIdFromFrozenIndex to expect Scroll object instead of null for serverless compatibility Signed-off-by: Sotaro Hikita <bering1814@gmail.com> * Add PIT support and _id sort tiebreaker to fix data consistency and missing documents in serverless search_after reads Signed-off-by: Sotaro Hikita <bering1814@gmail.com> --------- Signed-off-by: Sotaro Hikita <bering1814@gmail.com> Signed-off-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com> Co-authored-by: fukamishuhei <dimorportheca.0407@gmail.com> Co-authored-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com>
1 parent 1121d59 commit 9807b25

16 files changed

Lines changed: 805 additions & 77 deletions

‎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
- Add Apache Spark 4.0 support ([#684](https://github.com/opensearch-project/opensearch-hadoop/pull/684))
78

89
### Changed

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

Lines changed: 10 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 */
@@ -126,6 +130,10 @@ public interface ConfigurationOptions {
126130
String OPENSEARCH_SCROLL_KEEPALIVE = "opensearch.scroll.keepalive";
127131
String OPENSEARCH_SCROLL_KEEPALIVE_DEFAULT = "5m";
128132

133+
/** PIT keep-alive (serverless mode) */
134+
String OPENSEARCH_PIT_KEEPALIVE = "opensearch.pit.keepalive";
135+
String OPENSEARCH_PIT_KEEPALIVE_DEFAULT = "5m";
136+
129137
/** Scroll size */
130138
String OPENSEARCH_SCROLL_SIZE = "opensearch.scroll.size";
131139
String OPENSEARCH_SCROLL_SIZE_DEFAULT = "1000";
@@ -345,4 +353,5 @@ public interface ConfigurationOptions {
345353

346354
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME = "opensearch.aws.sigv4.service.name";
347355
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT = "es";
348-
}
356+
String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS = "aoss";
357+
}

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

Lines changed: 19 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
}
@@ -216,6 +220,10 @@ public long getScrollKeepAlive() {
216220
return TimeValue.parseTimeValue(getProperty(OPENSEARCH_SCROLL_KEEPALIVE, OPENSEARCH_SCROLL_KEEPALIVE_DEFAULT)).getMillis();
217221
}
218222

223+
public String getPitKeepAlive() {
224+
return getProperty(OPENSEARCH_PIT_KEEPALIVE, OPENSEARCH_PIT_KEEPALIVE_DEFAULT);
225+
}
226+
219227
public long getScrollSize() {
220228
return Long.valueOf(getProperty(OPENSEARCH_SCROLL_SIZE, OPENSEARCH_SCROLL_SIZE_DEFAULT));
221229
}
@@ -616,6 +624,11 @@ public Settings setPort(int port) {
616624
return this;
617625
}
618626

627+
public Settings setServerlessMode(boolean serverless) {
628+
setProperty(OPENSEARCH_SERVERLESS, Boolean.toString(serverless));
629+
return this;
630+
}
631+
619632
public Settings setResourceRead(String index) {
620633
setProperty(OPENSEARCH_RESOURCE_READ, index);
621634
return this;
@@ -814,6 +827,10 @@ public String getAwsSigV4Region() {
814827
}
815828

816829
public String getAwsSigV4ServiceName() {
817-
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
830+
if (getServerlessMode()) {
831+
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS);
832+
} else {
833+
return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
834+
}
818835
}
819-
}
836+
}

‎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: 86 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

@@ -530,6 +539,19 @@ public boolean deleteScroll(String scrollId) {
530539
return (res.status() == HttpStatus.OK ? true : false);
531540
}
532541

542+
public String createPit(String index, String keepAlive) {
543+
Response res = execute(POST, index + "/_search/point_in_time?keep_alive=" + keepAlive, true);
544+
String pitId = parseContent(res.body(), "pit_id");
545+
return pitId;
546+
}
547+
548+
public boolean deletePit(String pitId) {
549+
BytesArray body = new BytesArray(("{\"pit_id\":[\"" + pitId + "\"]}").getBytes(StringUtils.UTF_8));
550+
Request req = new SimpleRequest(DELETE, null, "_search/point_in_time", body);
551+
Response res = executeNotFoundAllowed(req);
552+
return (res.status() == HttpStatus.OK ? true : false);
553+
}
554+
533555
public boolean documentExists(String index, String type, String id) {
534556
return exists(index + "/" + type + "/" + id);
535557
}
@@ -636,6 +658,35 @@ static BytesArray searchRequest(QueryBuilder query) {
636658
return out.bytes();
637659
}
638660

661+
/**
662+
* Build a search_after request body by appending search_after to the original query body.
663+
*/
664+
public BytesArray buildSearchAfterBody(Object[] searchAfter) {
665+
FastByteArrayOutputStream out = new FastByteArrayOutputStream(256);
666+
JacksonJsonGenerator generator = new JacksonJsonGenerator(out);
667+
try {
668+
generator.writeBeginObject();
669+
generator.writeFieldName("search_after");
670+
generator.writeBeginArray();
671+
for (Object value : searchAfter) {
672+
if (value instanceof Long) {
673+
generator.writeNumber((Long) value);
674+
} else if (value instanceof Double) {
675+
generator.writeNumber((Double) value);
676+
} else if (value instanceof Integer) {
677+
generator.writeNumber((Integer) value);
678+
} else {
679+
generator.writeString(String.valueOf(value));
680+
}
681+
}
682+
generator.writeEndArray();
683+
generator.writeEndObject();
684+
} finally {
685+
generator.close();
686+
}
687+
return out.bytes();
688+
}
689+
639690
public boolean isAlias(String query) {
640691
Map<String, Object> aliases = (Map<String, Object>) get(query, null);
641692
return (aliases.size() > 1);
@@ -721,11 +772,26 @@ public boolean cancelToken(OpenSearchToken tokenToCancel) {
721772
}
722773

723774
public ClusterInfo mainInfo() {
724-
Response response = execute(GET, "", true);
725-
Map<String, Object> result = parseContent(response.body(), null);
726-
if (result == null) {
727-
throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
775+
// For serverless mode, return dummy info without making root request
776+
if (this.settings.getServerlessMode()) {
777+
// Use a dummy UUID instead of null to avoid NPE in validation
778+
ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
779+
return new ClusterInfo(clusterName, OpenSearchMajorVersion.V_2_X);
728780
}
781+
782+
// Check for cached serverless cluster info
783+
if (clusterInfo != null && clusterInfo.getMajorVersion() != null && "serverless-collection".equals(clusterInfo.getClusterName().getName())) {
784+
// already detected as serverless
785+
return clusterInfo;
786+
}
787+
788+
// Standard mode - retrieve information from the cluster
789+
try {
790+
Response response = execute(GET, "", true);
791+
Map<String, Object> result = parseContent(response.body(), null);
792+
if (result == null) {
793+
throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
794+
}
729795
String clusterName = result.get("cluster_name").toString();
730796
String clusterUUID = (String) result.get("cluster_uuid");
731797
@SuppressWarnings("unchecked")
@@ -740,6 +806,13 @@ public ClusterInfo mainInfo() {
740806
"Version is lower than minimum required version [" + OpenSearchMajorVersion.V_1_X + "].");
741807
}
742808
return new ClusterInfo(new ClusterName(clusterName, clusterUUID), OpenSearchMajorVersion.parse(versionNumber));
809+
} catch (Exception e) {
810+
// If unable to get cluster info in normal way, fallback to serverless mode
811+
LOG.debug("Error getting cluster info, falling back to serverless mode", e);
812+
// Use a dummy UUID instead of null to avoid NPE in validation
813+
ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
814+
return new ClusterInfo(clusterName, OpenSearchMajorVersion.LATEST);
815+
}
743816
}
744817

745818
/**
@@ -762,6 +835,15 @@ public Health getHealth(String index) {
762835
}
763836

764837
public boolean waitForHealth(String index, Health health, TimeValue timeout) {
838+
// Skip health check for serverless mode as _cluster/health endpoint is not supported
839+
if (settings.getServerlessMode()) {
840+
if (LOG.isDebugEnabled()) {
841+
LOG.debug("Serverless mode - skipping health check (not supported in serverless)");
842+
}
843+
// Return false to indicate no timeout
844+
return false;
845+
}
846+
765847
StringBuilder sb = new StringBuilder("/_cluster/health/");
766848
sb.append(index);
767849
sb.append("?wait_for_status=");

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

Lines changed: 55 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -348,6 +348,45 @@ Scroll scroll(String scrollId, ScrollReader reader) throws IOException {
348348
}
349349
}
350350

351+
Scroll searchAfter(String queryUri, BytesArray baseBody, Object[] searchAfter, String pitId, String keepAlive, ScrollReader reader) throws IOException {
352+
BytesArray body = mergeSearchAfterIntoBody(baseBody, searchAfter);
353+
if (StringUtils.hasText(pitId)) {
354+
body = injectPit(body, pitId, keepAlive);
355+
}
356+
InputStream response = client.execute(Request.Method.POST, queryUri, body).body();
357+
try {
358+
return reader.read(response);
359+
} finally {
360+
if (response instanceof StatsAware) {
361+
stats.aggregate(((StatsAware) response).stats());
362+
}
363+
}
364+
}
365+
366+
private static BytesArray injectPit(BytesArray body, String pitId, String keepAlive) {
367+
String base = body.toString().trim();
368+
String pitFragment = "\"pit\":{\"id\":\"" + pitId + "\",\"keep_alive\":\"" + keepAlive + "\"}";
369+
return new BytesArray(base.substring(0, base.length() - 1) + "," + pitFragment + "}");
370+
}
371+
372+
private BytesArray mergeSearchAfterIntoBody(BytesArray baseBody, Object[] searchAfter) {
373+
BytesArray searchAfterJson = client.buildSearchAfterBody(searchAfter);
374+
String base = baseBody.toString().trim();
375+
String saFragment = searchAfterJson.toString().trim();
376+
String saContent = saFragment.substring(1, saFragment.length() - 1);
377+
String merged = base.substring(0, base.length() - 1) + "," + saContent + "}";
378+
return new BytesArray(merged);
379+
}
380+
381+
ScrollQuery scanLimitSearchAfter(String query, BytesArray body, long limit, ScrollReader reader, String index, String keepAlive) {
382+
String pitId = client.createPit(index, keepAlive);
383+
return new ScrollQuery(this, query, body, limit, reader, true, pitId, keepAlive);
384+
}
385+
386+
public void deletePit(String pitId) {
387+
client.deletePit(pitId);
388+
}
389+
351390
public boolean resourceExists(boolean read) {
352391
Resource res = (read ? resources.getResourceRead() : resources.getResourceWrite());
353392
// cheap hit - works for exact index names, index patterns, the `_all` resource, and alias names
@@ -377,18 +416,20 @@ public boolean touch() {
377416
}
378417

379418
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);
419+
if (!this.settings.getServerlessMode()) {
420+
// try first a blind delete by query
421+
try {
422+
Resource res = resources.getResourceWrite();
423+
client.deleteByQuery(
424+
res.isTyped()
425+
? res.index() + "/" + res.type()
426+
: res.index(),
427+
MatchAllQueryBuilder.MATCH_ALL);
428+
} catch (OpenSearchHadoopInvalidRequest ehir) {
429+
log.error("Delete by query was not successful...", ehir);
430+
}
390431
}
391-
432+
392433
// in ES 2.0 and higher this means scrolling and deleting the docs by hand...
393434
// do a scroll-scan without source
394435

@@ -475,6 +516,8 @@ public long count(boolean read) {
475516
}
476517

477518
public boolean waitForYellow() {
519+
// For serverless collections, waitForHealth is handled through the RestClient.waitForHealth()
520+
// which will return appropriate result based on serverless mode
478521
return client.waitForHealth(resources.getResourceWrite().index(), RestClient.Health.YELLOW, TimeValue.timeValueSeconds(10));
479522
}
480523

@@ -495,4 +538,4 @@ public Stats stats() {
495538
public Settings getSettings() {
496539
return settings;
497540
}
498-
}
541+
}

0 commit comments

Comments
 (0)