[WIP] Scale to Zero with Reader/Writer Separation

gradle/run.gradle

@@ -45,6 +45,17 @@ testClusters {
         plugin('plugins:'.concat(p))
       }
     }
+    setting 'opensearch.experimental.feature.read.write.split.enabled', 'true'
+    setting 'path.repo', '["/tmp/my-repo"]'
+    setting 'node.attr.remote_store', 'true'
+    setting 'cluster.remote_store.state.enabled', 'true'
+    setting 'node.attr.remote_store.segment.repository', 'my-repository'
+    setting 'node.attr.remote_store.translog.repository', 'my-repository'
+    setting 'node.attr.remote_store.repository.my-repository.type', 'fs'
+    setting 'node.attr.remote_store.state.repository', 'my-repository'
+    setting 'node.attr.remote_store.repository.my-repository.settings.location', '/tmp/my-repo'
+
+
   }
 }
 

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -184,6 +184,8 @@
 import org.opensearch.action.admin.indices.resolve.ResolveIndexAction;
 import org.opensearch.action.admin.indices.rollover.RolloverAction;
 import org.opensearch.action.admin.indices.rollover.TransportRolloverAction;
+import org.opensearch.action.admin.indices.scaleToZero.PreScaleSyncAction;
+import org.opensearch.action.admin.indices.scaleToZero.TransportPreScaleSyncAction;
 import org.opensearch.action.admin.indices.segments.IndicesSegmentsAction;
 import org.opensearch.action.admin.indices.segments.PitSegmentsAction;
 import org.opensearch.action.admin.indices.segments.TransportIndicesSegmentsAction;
@@ -685,6 +687,9 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(AutoPutMappingAction.INSTANCE, TransportAutoPutMappingAction.class);
         actions.register(IndicesAliasesAction.INSTANCE, TransportIndicesAliasesAction.class);
         actions.register(UpdateSettingsAction.INSTANCE, TransportUpdateSettingsAction.class);
+
+        actions.register(PreScaleSyncAction.INSTANCE, TransportPreScaleSyncAction.class);
+
         actions.register(AnalyzeAction.INSTANCE, TransportAnalyzeAction.class);
         actions.register(PutIndexTemplateAction.INSTANCE, TransportPutIndexTemplateAction.class);
         actions.register(GetIndexTemplatesAction.INSTANCE, TransportGetIndexTemplatesAction.class);

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/NodePreScaleSyncRequest.java

@@ -0,0 +1,48 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.index.shard.ShardId;
+import org.opensearch.transport.TransportRequest;
+
+import java.io.IOException;
+import java.util.List;
+
+public class NodePreScaleSyncRequest extends TransportRequest {
+    private final String index;
+    private final List<ShardId> shardIds;
+
+    public NodePreScaleSyncRequest(String index, List<ShardId> shardIds) {
+        this.index = index;
+        this.shardIds = shardIds;
+    }
+
+    public NodePreScaleSyncRequest(StreamInput in) throws IOException {
+        super(in);
+        this.index = in.readString();
+        this.shardIds = in.readList(ShardId::new);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(index);
+        out.writeList(shardIds);
+    }
+
+    public String getIndex() {
+        return index;
+    }
+
+    public List<ShardId> getShardIds() {
+        return shardIds;
+    }
+}

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/NodePreScaleSyncResponse.java

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.transport.TransportResponse;
+
+import java.io.IOException;
+import java.util.List;
+
+public class NodePreScaleSyncResponse extends TransportResponse {
+    private final DiscoveryNode node;
+    private final List<ShardPreScaleSyncResponse> shardResponses;
+
+    public NodePreScaleSyncResponse(DiscoveryNode node, List<ShardPreScaleSyncResponse> shardResponses) {
+        this.node = node;
+        this.shardResponses = shardResponses;
+    }
+
+    public NodePreScaleSyncResponse(StreamInput in) throws IOException {
+        node = new DiscoveryNode(in);
+        shardResponses = in.readList(ShardPreScaleSyncResponse::new);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        node.writeTo(out);
+        out.writeList(shardResponses);
+    }
+
+    public DiscoveryNode getNode() {
+        return node;
+    }
+
+    public List<ShardPreScaleSyncResponse> getShardResponses() {
+        return shardResponses;
+    }
+}

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/PreScaleSyncAction.java

@@ -0,0 +1,20 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.action.ActionType;
+
+public class PreScaleSyncAction extends ActionType<PreScaleSyncResponse> {
+    public static final PreScaleSyncAction INSTANCE = new PreScaleSyncAction();
+    public static final String NAME = "indices:admin/settings/pre_scale_sync";
+
+    private PreScaleSyncAction() {
+        super(NAME, PreScaleSyncResponse::new);
+    }
+}

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/PreScaleSyncRequest.java

@@ -0,0 +1,44 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.ActionRequestValidationException;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+public class PreScaleSyncRequest extends ActionRequest {
+    private final String index;
+
+    public PreScaleSyncRequest(String index) {
+        this.index = index;
+    }
+
+    public PreScaleSyncRequest(StreamInput in) throws IOException {
+        super(in);
+        this.index = in.readString();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(index);
+    }
+
+    public String getIndex() {
+        return index;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+}

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/PreScaleSyncResponse.java

@@ -0,0 +1,90 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.core.action.ActionResponse;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.xcontent.ToXContent;
+import org.opensearch.core.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+
+public class PreScaleSyncResponse extends ActionResponse implements ToXContent {
+    private final Collection<NodePreScaleSyncResponse> nodeResponses;
+    private final String failureReason;
+    private final boolean hasFailures;
+
+    public PreScaleSyncResponse(Collection<NodePreScaleSyncResponse> responses) {
+        this.nodeResponses = responses;
+        this.hasFailures = responses.stream()
+            .anyMatch(r -> r.getShardResponses().stream().anyMatch(s -> s.hasUncommittedOperations() || s.needsSync()));
+        this.failureReason = buildFailureReason();
+    }
+
+    public PreScaleSyncResponse(StreamInput in) throws IOException {
+        this.nodeResponses = in.readList(NodePreScaleSyncResponse::new);
+        this.hasFailures = in.readBoolean();
+        this.failureReason = in.readOptionalString();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeList(new ArrayList<>(nodeResponses));  // Convert Collection to List
+        out.writeBoolean(hasFailures);
+        out.writeOptionalString(failureReason);
+    }
+
+    public boolean hasFailures() {
+        return hasFailures;
+    }
+
+    public String getFailureReason() {
+        return failureReason;
+    }
+
+    private String buildFailureReason() {
+        if (!hasFailures) {
+            return null;
+        }
+        StringBuilder reason = new StringBuilder();
+        for (NodePreScaleSyncResponse nodeResponse : nodeResponses) {
+            for (ShardPreScaleSyncResponse shardResponse : nodeResponse.getShardResponses()) {
+                if (shardResponse.hasUncommittedOperations() || shardResponse.needsSync()) {
+                    reason.append("Shard ")
+                        .append(shardResponse.getShardId())
+                        .append(" on node ")
+                        .append(nodeResponse.getNode())
+                        .append(": ");
+                    if (shardResponse.hasUncommittedOperations()) {
+                        reason.append("has uncommitted operations ");
+                    }
+                    if (shardResponse.needsSync()) {
+                        reason.append("needs sync ");
+                    }
+                    reason.append("; ");
+                }
+            }
+        }
+        return reason.toString();
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        builder.startObject();
+        builder.field("has_failures", hasFailures);
+        if (failureReason != null) {
+            builder.field("failure_reason", failureReason);
+        }
+        builder.endObject();
+        return builder;
+    }
+}

server/src/main/java/org/opensearch/action/admin/indices/scaleToZero/ShardPreScaleSyncResponse.java

@@ -0,0 +1,53 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.scaleToZero;
+
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.Writeable;
+import org.opensearch.core.index.shard.ShardId;
+
+import java.io.IOException;
+
+public class ShardPreScaleSyncResponse implements Writeable {
+    private final ShardId shardId;
+    private final boolean needsSync;
+    private final int uncommittedOperations;
+
+    public ShardPreScaleSyncResponse(ShardId shardId, boolean needsSync, int uncommittedOperations) {
+        this.shardId = shardId;
+        this.needsSync = needsSync;
+        this.uncommittedOperations = uncommittedOperations;
+    }
+
+    public ShardPreScaleSyncResponse(StreamInput in) throws IOException {
+        this.shardId = new ShardId(in);
+        this.needsSync = in.readBoolean();
+        this.uncommittedOperations = in.readVInt();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        shardId.writeTo(out);
+        out.writeBoolean(needsSync);
+        out.writeVInt(uncommittedOperations);
+    }
+
+    public ShardId getShardId() {
+        return shardId;
+    }
+
+    public boolean needsSync() {
+        return needsSync;
+    }
+
+    public boolean hasUncommittedOperations() {
+        return uncommittedOperations > 0;
+    }
+}