Add support for Amazon OpenSearch Serverless

CHANGELOG.md

@@ -3,6 +3,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ## [Unreleased]
 ### Added
+- Add support for Amazon OpenSearch Serverless ([#586](https://github.com/opensearch-project/opensearch-hadoop/pull/586))
 - Add Apache Spark 4.0 support ([#684](https://github.com/opensearch-project/opensearch-hadoop/pull/684))
 
 ### Changed

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

@@ -75,6 +75,10 @@ public interface ConfigurationOptions {
     String OPENSEARCH_NODES_WAN_ONLY = "opensearch.nodes.wan.only";
     String OPENSEARCH_NODES_WAN_ONLY_DEFAULT = "false";
 
+    /** Serverless mode */
+    String OPENSEARCH_SERVERLESS = "opensearch.serverless";
+    String OPENSEARCH_SERVERLESS_DEFAULT = "false";
+
     String OPENSEARCH_NODES_RESOLVE_HOST_NAME = "opensearch.nodes.resolve.hostname";
 
     /** Secure Settings Keystore */
@@ -126,6 +130,10 @@ public interface ConfigurationOptions {
     String OPENSEARCH_SCROLL_KEEPALIVE = "opensearch.scroll.keepalive";
     String OPENSEARCH_SCROLL_KEEPALIVE_DEFAULT = "5m";
 
+    /** PIT keep-alive (serverless mode) */
+    String OPENSEARCH_PIT_KEEPALIVE = "opensearch.pit.keepalive";
+    String OPENSEARCH_PIT_KEEPALIVE_DEFAULT = "5m";
+
     /** Scroll size */
     String OPENSEARCH_SCROLL_SIZE = "opensearch.scroll.size";
     String OPENSEARCH_SCROLL_SIZE_DEFAULT = "1000";
@@ -345,4 +353,5 @@ public interface ConfigurationOptions {
 
     String OPENSEARCH_AWS_SIGV4_SERVICE_NAME = "opensearch.aws.sigv4.service.name";
     String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT = "es";
-}
+    String OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS = "aoss";
+}

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

@@ -168,6 +168,10 @@ public boolean getNodesWANOnly() {
         return Booleans.parseBoolean(getProperty(OPENSEARCH_NODES_WAN_ONLY, OPENSEARCH_NODES_WAN_ONLY_DEFAULT));
     }
 
+    public boolean getServerlessMode() {
+        return Booleans.parseBoolean(getProperty(OPENSEARCH_SERVERLESS, OPENSEARCH_SERVERLESS_DEFAULT));
+    }
+
     public long getHttpTimeout() {
         return TimeValue.parseTimeValue(getProperty(OPENSEARCH_HTTP_TIMEOUT, OPENSEARCH_HTTP_TIMEOUT_DEFAULT)).getMillis();
     }
@@ -216,6 +220,10 @@ public long getScrollKeepAlive() {
         return TimeValue.parseTimeValue(getProperty(OPENSEARCH_SCROLL_KEEPALIVE, OPENSEARCH_SCROLL_KEEPALIVE_DEFAULT)).getMillis();
     }
 
+    public String getPitKeepAlive() {
+        return getProperty(OPENSEARCH_PIT_KEEPALIVE, OPENSEARCH_PIT_KEEPALIVE_DEFAULT);
+    }
+
     public long getScrollSize() {
         return Long.valueOf(getProperty(OPENSEARCH_SCROLL_SIZE, OPENSEARCH_SCROLL_SIZE_DEFAULT));
     }
@@ -616,6 +624,11 @@ public Settings setPort(int port) {
         return this;
     }
 
+    public Settings setServerlessMode(boolean serverless) {
+        setProperty(OPENSEARCH_SERVERLESS, Boolean.toString(serverless));
+        return this;
+    }
+
     public Settings setResourceRead(String index) {
         setProperty(OPENSEARCH_RESOURCE_READ, index);
         return this;
@@ -814,6 +827,10 @@ public String getAwsSigV4Region() {
     }
 
     public String getAwsSigV4ServiceName() {
-        return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
+        if (getServerlessMode()) {
+            return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_SERVERLESS);
+        } else {
+            return getProperty(OPENSEARCH_AWS_SIGV4_SERVICE_NAME, OPENSEARCH_AWS_SIGV4_SERVICE_NAME_DEFAULT);
+        }
     }
-}
+}

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

@@ -85,13 +85,19 @@ public static void checkIndexStatus(Settings settings) {
 
             try {
                 if (bootstrap.indexExists(readResource.index())) {
-                    RestClient.Health status = bootstrap.getHealth(readResource.index());
-                    if (status == RestClient.Health.RED) {
-                        throw new OpenSearchHadoopIllegalStateException("Index specified [" + readResource.index()
-                                + "] is either red or " +
-                                "includes an index that is red, and thus all requested data cannot be safely and fully loaded. "
-                                +
-                                "Bailing out...");
+                    if (settings.getServerlessMode()) {
+                        if (LOG.isDebugEnabled()) {
+                            LOG.debug("Serverless mode - skipping health check (not supported in serverless)");
+                        }
+                   } else {
+                        RestClient.Health status = bootstrap.getHealth(readResource.index());
+                        if (status == RestClient.Health.RED) {
+                            throw new OpenSearchHadoopIllegalStateException("Index specified [" + readResource.index()
+                                    + "] is either red or " +
+                                    "includes an index that is red, and thus all requested data cannot be safely and fully loaded. "
+                                    +
+                                    "Bailing out...");
+                        }
                     }
                 }
             } finally {
@@ -339,9 +345,11 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
         try {
             mainInfo = bootstrap.mainInfo();
             if (log.isDebugEnabled()) {
+                // Handle serverless mode where UUID might be null
+                String uuid = mainInfo.getClusterName().getUUID() != null ? mainInfo.getClusterName().getUUID() : "N/A";
                 log.debug(String.format("Discovered OpenSearch cluster [%s/%s], version [%s]",
                         mainInfo.getClusterName().getName(),
-                        mainInfo.getClusterName().getUUID(),
+                        uuid,
                         mainInfo.getMajorVersion()));
             }
         } catch (OpenSearchHadoopException ex) {
@@ -364,7 +372,8 @@ public static ClusterInfo discoverAndValidateClusterInfo(Settings settings, Log
                         mainInfo.getClusterName().getName(),
                         clusterName));
             }
-            if (mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
+            if (mainInfo.getClusterName().getUUID() != null && 
+                mainInfo.getClusterName().getUUID().equals(clusterUUID) == false) {
                 log.warn(String.format(
                         "Discovered incorrect cluster UUID in settings. Expected [%s] but received [%s]; replacing...",
                         mainInfo.getClusterName().getUUID(),
@@ -556,4 +565,4 @@ public static boolean setUserProviderIfNotSet(Settings settings, Class<? extends
         }
         return false;
     }
-}
+}

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

@@ -209,4 +209,4 @@ public String currentNode() {
     public String toString() {
         return settings.toString();
     }
-}
+}

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

@@ -95,6 +95,7 @@ public class RestClient implements Closeable, StatsAware {
     private final HttpRetryPolicy retryPolicy;
     final ClusterInfo clusterInfo;
     private final ErrorExtractor errorExtractor;
+    private final Settings settings;
 
     {
         mapper = new ObjectMapper();
@@ -116,6 +117,7 @@ public RestClient(Settings settings) {
         this.network = networkClient;
         this.scrollKeepAlive = TimeValue.timeValueMillis(settings.getScrollKeepAlive());
         this.indexReadMissingAsEmpty = settings.getIndexReadMissingAsEmpty();
+        this.settings = settings;
 
         String retryPolicyName = settings.getBatchWriteRetryPolicy();
 
@@ -296,6 +298,13 @@ public String postDocument(Resource resource, BytesArray document) throws IOExce
     }
 
     public void refresh(Resource resource) {
+        // Skip refresh operation for serverless mode as _refresh endpoint is not supported
+        if (settings.getServerlessMode()) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Serverless mode - skipping refresh operation (not supported in serverless)");
+            }
+            return;
+        }
         execute(POST, resource.refresh());
     }
 
@@ -526,6 +535,19 @@ public boolean deleteScroll(String scrollId) {
         return (res.status() == HttpStatus.OK ? true : false);
     }
 
+    public String createPit(String index, String keepAlive) {
+        Response res = execute(POST, index + "/_search/point_in_time?keep_alive=" + keepAlive, true);
+        String pitId = parseContent(res.body(), "pit_id");
+        return pitId;
+    }
+
+    public boolean deletePit(String pitId) {
+        BytesArray body = new BytesArray(("{\"pit_id\":[\"" + pitId + "\"]}").getBytes(StringUtils.UTF_8));
+        Request req = new SimpleRequest(DELETE, null, "_search/point_in_time", body);
+        Response res = executeNotFoundAllowed(req);
+        return (res.status() == HttpStatus.OK ? true : false);
+    }
+
     public boolean documentExists(String index, String type, String id) {
         return exists(index + "/" + type + "/" + id);
     }
@@ -632,6 +654,35 @@ static BytesArray searchRequest(QueryBuilder query) {
         return out.bytes();
     }
 
+    /**
+     * Build a search_after request body by appending search_after to the original query body.
+     */
+    public BytesArray buildSearchAfterBody(Object[] searchAfter) {
+        FastByteArrayOutputStream out = new FastByteArrayOutputStream(256);
+        JacksonJsonGenerator generator = new JacksonJsonGenerator(out);
+        try {
+            generator.writeBeginObject();
+            generator.writeFieldName("search_after");
+            generator.writeBeginArray();
+            for (Object value : searchAfter) {
+                if (value instanceof Long) {
+                    generator.writeNumber((Long) value);
+                } else if (value instanceof Double) {
+                    generator.writeNumber((Double) value);
+                } else if (value instanceof Integer) {
+                    generator.writeNumber((Integer) value);
+                } else {
+                    generator.writeString(String.valueOf(value));
+                }
+            }
+            generator.writeEndArray();
+            generator.writeEndObject();
+        } finally {
+            generator.close();
+        }
+        return out.bytes();
+    }
+
     public boolean isAlias(String query) {
         Map<String, Object> aliases = (Map<String, Object>) get(query, null);
         return (aliases.size() > 1);
@@ -717,11 +768,26 @@ public boolean cancelToken(OpenSearchToken tokenToCancel) {
     }
 
     public ClusterInfo mainInfo() {
-        Response response = execute(GET, "", true);
-        Map<String, Object> result = parseContent(response.body(), null);
-        if (result == null) {
-            throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
+        // For serverless mode, return dummy info without making root request
+        if (this.settings.getServerlessMode()) {
+            // Use a dummy UUID instead of null to avoid NPE in validation
+            ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
+            return new ClusterInfo(clusterName, OpenSearchMajorVersion.V_2_X);
         }
+
+        // Check for cached serverless cluster info
+        if (clusterInfo != null && clusterInfo.getMajorVersion() != null && "serverless-collection".equals(clusterInfo.getClusterName().getName())) {
+            // already detected as serverless
+            return clusterInfo;
+        }
+
+        // Standard mode - retrieve information from the cluster
+        try {
+            Response response = execute(GET, "", true);
+            Map<String, Object> result = parseContent(response.body(), null);
+            if (result == null) {
+                throw new OpenSearchHadoopIllegalStateException("Unable to retrieve OpenSearch main cluster info.");
+            }
         String clusterName = result.get("cluster_name").toString();
         String clusterUUID = (String) result.get("cluster_uuid");
         @SuppressWarnings("unchecked")
@@ -736,6 +802,13 @@ public ClusterInfo mainInfo() {
                     "Version is lower than minimum required version [" + OpenSearchMajorVersion.V_1_X + "].");
         }
         return new ClusterInfo(new ClusterName(clusterName, clusterUUID), OpenSearchMajorVersion.parse(versionNumber));
+        } catch (Exception e) {
+            // If unable to get cluster info in normal way, fallback to serverless mode
+            LOG.debug("Error getting cluster info, falling back to serverless mode", e);
+            // Use a dummy UUID instead of null to avoid NPE in validation
+            ClusterName clusterName = new ClusterName("serverless-collection", "serverless-uuid");
+            return new ClusterInfo(clusterName, OpenSearchMajorVersion.LATEST);
+        }
     }
 
     /**
@@ -758,6 +831,15 @@ public Health getHealth(String index) {
     }
 
     public boolean waitForHealth(String index, Health health, TimeValue timeout) {
+        // Skip health check for serverless mode as _cluster/health endpoint is not supported
+        if (settings.getServerlessMode()) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Serverless mode - skipping health check (not supported in serverless)");
+            }
+            // Return false to indicate no timeout
+            return false;
+        }
+
         StringBuilder sb = new StringBuilder("/_cluster/health/");
         sb.append(index);
         sb.append("?wait_for_status=");

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

@@ -348,6 +348,45 @@ Scroll scroll(String scrollId, ScrollReader reader) throws IOException {
         }
     }
 
+    Scroll searchAfter(String queryUri, BytesArray baseBody, Object[] searchAfter, String pitId, String keepAlive, ScrollReader reader) throws IOException {
+        BytesArray body = mergeSearchAfterIntoBody(baseBody, searchAfter);
+        if (StringUtils.hasText(pitId)) {
+            body = injectPit(body, pitId, keepAlive);
+        }
+        InputStream response = client.execute(Request.Method.POST, queryUri, body).body();
+        try {
+            return reader.read(response);
+        } finally {
+            if (response instanceof StatsAware) {
+                stats.aggregate(((StatsAware) response).stats());
+            }
+        }
+    }
+
+    private static BytesArray injectPit(BytesArray body, String pitId, String keepAlive) {
+        String base = body.toString().trim();
+        String pitFragment = "\"pit\":{\"id\":\"" + pitId + "\",\"keep_alive\":\"" + keepAlive + "\"}";
+        return new BytesArray(base.substring(0, base.length() - 1) + "," + pitFragment + "}");
+    }
+
+    private BytesArray mergeSearchAfterIntoBody(BytesArray baseBody, Object[] searchAfter) {
+        BytesArray searchAfterJson = client.buildSearchAfterBody(searchAfter);
+        String base = baseBody.toString().trim();
+        String saFragment = searchAfterJson.toString().trim();
+        String saContent = saFragment.substring(1, saFragment.length() - 1);
+        String merged = base.substring(0, base.length() - 1) + "," + saContent + "}";
+        return new BytesArray(merged);
+    }
+
+    ScrollQuery scanLimitSearchAfter(String query, BytesArray body, long limit, ScrollReader reader, String index, String keepAlive) {
+        String pitId = client.createPit(index, keepAlive);
+        return new ScrollQuery(this, query, body, limit, reader, true, pitId, keepAlive);
+    }
+
+    public void deletePit(String pitId) {
+        client.deletePit(pitId);
+    }
+
     public boolean resourceExists(boolean read) {
         Resource res = (read ? resources.getResourceRead() : resources.getResourceWrite());
         // cheap hit - works for exact index names, index patterns, the `_all` resource, and alias names
@@ -377,18 +416,20 @@ public boolean touch() {
     }
 
     public void delete() {
-        // try first a blind delete by query
-        try {
-            Resource res = resources.getResourceWrite();
-            client.deleteByQuery(
-                    res.isTyped()
-                            ? res.index() + "/" + res.type()
-                            : res.index(),
-                    MatchAllQueryBuilder.MATCH_ALL);
-        } catch (OpenSearchHadoopInvalidRequest ehir) {
-            log.error("Delete by query was not successful...", ehir);
+        if (!this.settings.getServerlessMode()) {
+            // try first a blind delete by query
+            try {
+                Resource res = resources.getResourceWrite();
+                client.deleteByQuery(
+                        res.isTyped()
+                                ? res.index() + "/" + res.type()
+                                : res.index(),
+                        MatchAllQueryBuilder.MATCH_ALL);
+            } catch (OpenSearchHadoopInvalidRequest ehir) {
+                log.error("Delete by query was not successful...", ehir);
+            } 
         }
-
+        
         // in ES 2.0 and higher this means scrolling and deleting the docs by hand...
         // do a scroll-scan without source
 
@@ -475,6 +516,8 @@ public long count(boolean read) {
     }
 
     public boolean waitForYellow() {
+        // For serverless collections, waitForHealth is handled through the RestClient.waitForHealth()
+        // which will return appropriate result based on serverless mode
         return client.waitForHealth(resources.getResourceWrite().index(), RestClient.Health.YELLOW, TimeValue.timeValueSeconds(10));
     }
 
@@ -495,4 +538,4 @@ public Stats stats() {
     public Settings getSettings() {
         return settings;
     }
-}
+}