Skip to content

Commit e25c665

Browse files
lawofcycleslawofcycles
committed
Add support for Amazon OpenSearch Serverless
Signed-off-by: lawofcycles <bering1814@gmail.com>
1 parent 525d355 commit e25c665

6 files changed

Lines changed: 123 additions & 33 deletions

File tree

‎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: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -339,9 +339,11 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
339339
try {
340340
mainInfo = bootstrap.mainInfo();
341341
if (log.isDebugEnabled()) {
342+
// Handle serverless mode where UUID might be null
343+
String uuid = mainInfo.getClusterName().getUUID() != null ? mainInfo.getClusterName().getUUID() : "N/A";
342344
log.debug(String.format("Discovered OpenSearch cluster [%s/%s], version [%s]",
343345
mainInfo.getClusterName().getName(),
344-
mainInfo.getClusterName().getUUID(),
346+
uuid,
345347
mainInfo.getMajorVersion()));
346348
}
347349
} catch (OpenSearchHadoopException ex) {
@@ -364,7 +366,8 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
364366
mainInfo.getClusterName().getName(),
365367
clusterName));
366368
}
367-
if (mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
369+
if (mainInfo.getClusterName().getUUID() != null &&
370+
mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
368371
log.warn(String.format(
369372
"Discovered incorrect cluster UUID in settings. Expected [%s] but received [%s]; replacing...",
370373
mainInfo.getClusterName().getUUID(),
@@ -556,4 +559,4 @@ public static boolean setUserProviderIfNotSet(Settings settings, Class<? extends
556559
}
557560
return false;
558561
}
559-
}
562+
}

‎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: 46 additions & 6 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

@@ -295,9 +297,16 @@ public String postDocument(Resource resource, BytesArray document) throws IOExce
295297
return id.toString();
296298
}
297299

298-
public void refresh(Resource resource) {
299-
execute(POST, resource.refresh());
300+
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;
300307
}
308+
execute(POST, resource.refresh());
309+
}
301310

302311
public List<List<Map<String, Object>>> targetShards(String index, String routing) {
303312
List<List<Map<String, Object>>> shardsJson = null;
@@ -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.LATEST);
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: 49 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -277,30 +277,57 @@ public Map<ShardInfo, NodeInfo> getWriteTargetPrimaryShards(boolean clientNodesO
277277
}
278278

279279
protected Map<ShardInfo, NodeInfo> doGetWriteTargetPrimaryShards(boolean clientNodesOnly) {
280-
List<List<Map<String, Object>>> info = client.targetShards(resources.getResourceWrite().index(), SettingsUtils.getFixedRouting(settings));
281-
Map<ShardInfo, NodeInfo> shards = new LinkedHashMap<ShardInfo, NodeInfo>();
282-
List<NodeInfo> nodes = client.getHttpNodes(clientNodesOnly);
283-
Map<String, NodeInfo> nodeMap = new HashMap<String, NodeInfo>(nodes.size());
284-
for (NodeInfo node : nodes) {
285-
nodeMap.put(node.getId(), node);
286-
}
280+
// For serverless mode, we skip shard information retrieval as it's not supported
281+
if (settings.getServerlessMode()) {
282+
Map<ShardInfo, NodeInfo> shards = new LinkedHashMap<ShardInfo, NodeInfo>();
283+
List<NodeInfo> nodes = client.getHttpNodes(clientNodesOnly);
284+
285+
if (nodes.isEmpty()) {
286+
log.warn("Cannot find any nodes (is HTTP enabled?)");
287+
return null;
288+
}
289+
290+
// In serverless mode, we create a single dummy shard assigned to the first available node
291+
NodeInfo firstNode = nodes.get(0);
292+
Map<String, Object> dummyShardData = new HashMap<String, Object>();
293+
dummyShardData.put("index", resources.getResourceWrite().index());
294+
dummyShardData.put("shard", 0);
295+
dummyShardData.put("primary", true);
296+
dummyShardData.put("node", firstNode.getId());
297+
dummyShardData.put("state", "STARTED");
298+
dummyShardData.put("relocating_node", null);
299+
300+
ShardInfo dummyShard = new ShardInfo(dummyShardData);
301+
shards.put(dummyShard, firstNode);
302+
303+
return shards;
304+
} else {
305+
// Original implementation for non-serverless mode
306+
List<List<Map<String, Object>>> info = client.targetShards(resources.getResourceWrite().index(), SettingsUtils.getFixedRouting(settings));
307+
Map<ShardInfo, NodeInfo> shards = new LinkedHashMap<ShardInfo, NodeInfo>();
308+
List<NodeInfo> nodes = client.getHttpNodes(clientNodesOnly);
309+
Map<String, NodeInfo> nodeMap = new HashMap<String, NodeInfo>(nodes.size());
310+
for (NodeInfo node : nodes) {
311+
nodeMap.put(node.getId(), node);
312+
}
287313

288-
for (List<Map<String, Object>> shardGroup : info) {
289-
// consider only primary shards
290-
for (Map<String, Object> shardData : shardGroup) {
291-
ShardInfo shard = new ShardInfo(shardData);
292-
if (shard.isPrimary()) {
293-
NodeInfo node = nodeMap.get(shard.getNode());
294-
if (node == null) {
295-
log.warn(String.format("Cannot find node with id [%s] (is HTTP enabled?) from shard [%s] in nodes [%s]; layout [%s]", shard.getNode(), shard, nodes, info));
296-
return null;
314+
for (List<Map<String, Object>> shardGroup : info) {
315+
// consider only primary shards
316+
for (Map<String, Object> shardData : shardGroup) {
317+
ShardInfo shard = new ShardInfo(shardData);
318+
if (shard.isPrimary()) {
319+
NodeInfo node = nodeMap.get(shard.getNode());
320+
if (node == null) {
321+
log.warn(String.format("Cannot find node with id [%s] (is HTTP enabled?) from shard [%s] in nodes [%s]; layout [%s]", shard.getNode(), shard, nodes, info));
322+
return null;
323+
}
324+
shards.put(shard, node);
325+
break;
297326
}
298-
shards.put(shard, node);
299-
break;
300327
}
301328
}
329+
return shards;
302330
}
303-
return shards;
304331
}
305332

306333
public MappingSet getMappings() {
@@ -475,6 +502,8 @@ public long count(boolean read) {
475502
}
476503

477504
public boolean waitForYellow() {
505+
// For serverless collections, waitForHealth is handled through the RestClient.waitForHealth()
506+
// which will return appropriate result based on serverless mode
478507
return client.waitForHealth(resources.getResourceWrite().index(), RestClient.Health.YELLOW, TimeValue.timeValueSeconds(10));
479508
}
480509

@@ -495,4 +524,4 @@ public Stats stats() {
495524
public Settings getSettings() {
496525
return settings;
497526
}
498-
}
527+
}

0 commit comments

Comments
 (0)