[Snapshot Interop] Add Changes in Create Snapshot Flow for remote store interoperability.

Signed-off-by: Harish Bhakuni <hbhakuni@amazon.com>

Author: Harish Bhakuni

CHANGELOG.md

@@ -109,6 +109,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Changed
 - Enable `./gradlew build` on MacOS by disabling bcw tests ([#7303](https://github.com/opensearch-project/OpenSearch/pull/7303))
 - Moved concurrent-search from sandbox plugin to server module behind feature flag ([#7203](https://github.com/opensearch-project/OpenSearch/pull/7203))
+- [Snapshot Interop] Add Changes in Create Snapshot Flow for remote store interoperability. ([#7118](https://github.com/opensearch-project/OpenSearch/pull/7118))
 
 ### Deprecated
 
@@ -120,4 +121,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.7...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.7...2.x

server/src/internalClusterTest/java/org/opensearch/cluster/ClusterStateDiffIT.java

@@ -776,7 +776,8 @@ public ClusterState.Custom randomCreate(String name) {
                                     Map.of(),
                                     null,
                                     SnapshotInfoTests.randomUserMetadata(),
-                                    randomVersion(random())
+                                    randomVersion(random()),
+                                    false
                                 )
                             )
                         );

server/src/main/java/org/opensearch/action/admin/cluster/snapshots/create/CreateSnapshotRequest.java

@@ -34,6 +34,7 @@
 
 import org.opensearch.OpenSearchException;
 import org.opensearch.OpenSearchGenerationException;
+import org.opensearch.Version;
 import org.opensearch.action.ActionRequestValidationException;
 import org.opensearch.action.IndicesRequest;
 import org.opensearch.action.support.IndicesOptions;
@@ -101,6 +102,10 @@ public class CreateSnapshotRequest extends ClusterManagerNodeRequest<CreateSnaps
 
     private Map<String, Object> userMetadata;
 
+    private Boolean remoteStoreIndexShallowCopy;
+
+    private static final String REMOTE_STORE_INDEX_SHALLOW_COPY = "remote_store_index_shallow_copy";
+
     public CreateSnapshotRequest() {}
 
     /**
@@ -125,6 +130,9 @@ public CreateSnapshotRequest(StreamInput in) throws IOException {
         waitForCompletion = in.readBoolean();
         partial = in.readBoolean();
         userMetadata = in.readMap();
+        if (in.getVersion().onOrAfter(Version.V_2_8_0)) {
+            remoteStoreIndexShallowCopy = in.readOptionalBoolean();
+        }
     }
 
     @Override
@@ -139,6 +147,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(waitForCompletion);
         out.writeBoolean(partial);
         out.writeMap(userMetadata);
+        if (out.getVersion().onOrAfter(Version.V_2_8_0)) {
+            out.writeOptionalBoolean(remoteStoreIndexShallowCopy);
+        }
     }
 
     @Override
@@ -328,6 +339,11 @@ public CreateSnapshotRequest waitForCompletion(boolean waitForCompletion) {
         return this;
     }
 
+    public CreateSnapshotRequest remoteStoreIndexShallowCopy(boolean remoteStoreIndexShallowCopy) {
+        this.remoteStoreIndexShallowCopy = remoteStoreIndexShallowCopy;
+        return this;
+    }
+
     /**
      * Returns true if the request should wait for the snapshot completion before returning
      *
@@ -429,6 +445,10 @@ public Map<String, Object> userMetadata() {
         return userMetadata;
     }
 
+    public Boolean remoteStoreIndexShallowCopy() {
+        return remoteStoreIndexShallowCopy;
+    }
+
     public CreateSnapshotRequest userMetadata(Map<String, Object> userMetadata) {
         this.userMetadata = userMetadata;
         return this;
@@ -466,6 +486,8 @@ public CreateSnapshotRequest source(Map<String, Object> source) {
                     throw new IllegalArgumentException("malformed metadata, should be an object");
                 }
                 userMetadata((Map<String, Object>) entry.getValue());
+            } else if (name.equals(REMOTE_STORE_INDEX_SHALLOW_COPY)) {
+                remoteStoreIndexShallowCopy = nodeBooleanValue(entry.getValue(), REMOTE_STORE_INDEX_SHALLOW_COPY);
             }
         }
         indicesOptions(IndicesOptions.fromMap(source, indicesOptions));
@@ -495,6 +517,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             indicesOptions.toXContent(builder, params);
         }
         builder.field("metadata", userMetadata);
+        builder.field(REMOTE_STORE_INDEX_SHALLOW_COPY, remoteStoreIndexShallowCopy);
         builder.endObject();
         return builder;
     }
@@ -518,7 +541,8 @@ public boolean equals(Object o) {
             && Objects.equals(indicesOptions, that.indicesOptions)
             && Objects.equals(settings, that.settings)
             && Objects.equals(clusterManagerNodeTimeout, that.clusterManagerNodeTimeout)
-            && Objects.equals(userMetadata, that.userMetadata);
+            && Objects.equals(userMetadata, that.userMetadata)
+            && Objects.equals(remoteStoreIndexShallowCopy, that.remoteStoreIndexShallowCopy);
     }
 
     @Override
@@ -562,6 +586,8 @@ public String toString() {
             + clusterManagerNodeTimeout
             + ", metadata="
             + userMetadata
+            + ", remoteStoreIndexShallowCopy="
+            + remoteStoreIndexShallowCopy
             + '}';
     }
 }

server/src/main/java/org/opensearch/cluster/SnapshotsInProgress.java

@@ -113,7 +113,8 @@ public static Entry startedEntry(
         long repositoryStateId,
         final Map<ShardId, ShardSnapshotStatus> shards,
         Map<String, Object> userMetadata,
-        Version version
+        Version version,
+        boolean remoteStoreIndexShallowCopy
     ) {
         return new SnapshotsInProgress.Entry(
             snapshot,
@@ -127,7 +128,8 @@ public static Entry startedEntry(
             shards,
             null,
             userMetadata,
-            version
+            version,
+            remoteStoreIndexShallowCopy
         );
     }
 
@@ -164,7 +166,8 @@ public static Entry startClone(
             Collections.emptyMap(),
             version,
             source,
-            Map.of()
+            Map.of(),
+            false // TODO: need to pull this value from the original snapshot, use whatever we set during snapshot create.
         );
     }
 
@@ -177,6 +180,7 @@ public static class Entry implements Writeable, ToXContent, RepositoryOperation
         private final State state;
         private final Snapshot snapshot;
         private final boolean includeGlobalState;
+        private final boolean remoteStoreIndexShallowCopy;
         private final boolean partial;
         /**
          * Map of {@link ShardId} to {@link ShardSnapshotStatus} tracking the state of each shard snapshot operation.
@@ -219,7 +223,8 @@ public Entry(
             final Map<ShardId, ShardSnapshotStatus> shards,
             String failure,
             Map<String, Object> userMetadata,
-            Version version
+            Version version,
+            boolean remoteStoreIndexShallowCopy
         ) {
             this(
                 snapshot,
@@ -235,7 +240,8 @@ public Entry(
                 userMetadata,
                 version,
                 null,
-                Map.of()
+                Map.of(),
+                remoteStoreIndexShallowCopy
             );
         }
 
@@ -253,7 +259,8 @@ private Entry(
             final Map<String, Object> userMetadata,
             Version version,
             @Nullable SnapshotId source,
-            @Nullable final Map<RepositoryShardId, ShardSnapshotStatus> clones
+            @Nullable final Map<RepositoryShardId, ShardSnapshotStatus> clones,
+            boolean remoteStoreIndexShallowCopy
         ) {
             this.state = state;
             this.snapshot = snapshot;
@@ -274,6 +281,7 @@ private Entry(
             } else {
                 this.clones = Collections.unmodifiableMap(clones);
             }
+            this.remoteStoreIndexShallowCopy = remoteStoreIndexShallowCopy;
             assert assertShardsConsistent(this.source, this.state, this.indices, this.shards, this.clones);
         }
 
@@ -292,6 +300,11 @@ private Entry(StreamInput in) throws IOException {
             dataStreams = in.readStringList();
             source = in.readOptionalWriteable(SnapshotId::new);
             clones = in.readMap(RepositoryShardId::new, ShardSnapshotStatus::readFrom);
+            if (in.getVersion().onOrAfter(Version.V_2_8_0)) {
+                remoteStoreIndexShallowCopy = in.readBoolean();
+            } else {
+                remoteStoreIndexShallowCopy = false;
+            }
         }
 
         private static boolean assertShardsConsistent(
@@ -346,7 +359,8 @@ public Entry(
             long repositoryStateId,
             final Map<ShardId, ShardSnapshotStatus> shards,
             Map<String, Object> userMetadata,
-            Version version
+            Version version,
+            boolean remoteStoreIndexShallowCopy
         ) {
             this(
                 snapshot,
@@ -360,7 +374,8 @@ public Entry(
                 shards,
                 null,
                 userMetadata,
-                version
+                version,
+                remoteStoreIndexShallowCopy
             );
         }
 
@@ -385,7 +400,8 @@ public Entry(
                 shards,
                 failure,
                 entry.userMetadata,
-                version
+                version,
+                entry.remoteStoreIndexShallowCopy
             );
         }
 
@@ -409,7 +425,8 @@ public Entry withRepoGen(long newRepoGen) {
                 userMetadata,
                 version,
                 source,
-                clones
+                clones,
+                remoteStoreIndexShallowCopy
             );
         }
 
@@ -431,7 +448,8 @@ public Entry withClones(final Map<RepositoryShardId, ShardSnapshotStatus> update
                 userMetadata,
                 version,
                 source,
-                updatedClones
+                updatedClones,
+                remoteStoreIndexShallowCopy
             );
         }
 
@@ -486,7 +504,8 @@ public Entry fail(final Map<ShardId, ShardSnapshotStatus> shards, State state, S
                 userMetadata,
                 version,
                 source,
-                clones
+                clones,
+                remoteStoreIndexShallowCopy
             );
         }
 
@@ -512,7 +531,8 @@ public Entry withShardStates(final Map<ShardId, ShardSnapshotStatus> shards) {
                     shards,
                     failure,
                     userMetadata,
-                    version
+                    version,
+                    remoteStoreIndexShallowCopy
                 );
             }
             return withStartedShards(shards);
@@ -535,7 +555,8 @@ public Entry withStartedShards(final Map<ShardId, ShardSnapshotStatus> shards) {
                 shards,
                 failure,
                 userMetadata,
-                version
+                version,
+                remoteStoreIndexShallowCopy
             );
             assert updated.state().completed() == false && completed(updated.shards().values()) == false
                 : "Only running snapshots allowed but saw [" + updated + "]";
@@ -567,6 +588,10 @@ public boolean includeGlobalState() {
             return includeGlobalState;
         }
 
+        public boolean remoteStoreIndexShallowCopy() {
+            return remoteStoreIndexShallowCopy;
+        }
+
         public Map<String, Object> userMetadata() {
             return userMetadata;
         }
@@ -630,7 +655,7 @@ public boolean equals(Object o) {
             if (version.equals(entry.version) == false) return false;
             if (Objects.equals(source, ((Entry) o).source) == false) return false;
             if (clones.equals(((Entry) o).clones) == false) return false;
-
+            if (remoteStoreIndexShallowCopy != entry.remoteStoreIndexShallowCopy) return false;
             return true;
         }
 
@@ -647,6 +672,7 @@ public int hashCode() {
             result = 31 * result + version.hashCode();
             result = 31 * result + (source == null ? 0 : source.hashCode());
             result = 31 * result + clones.hashCode();
+            result = 31 * result + (remoteStoreIndexShallowCopy ? 1 : 0);
             return result;
         }
 
@@ -710,6 +736,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeStringCollection(dataStreams);
             out.writeOptionalWriteable(source);
             out.writeMap(clones, (o, v) -> v.writeTo(o), (o, v) -> v.writeTo(o));
+            if (out.getVersion().onOrAfter(Version.V_2_8_0)) {
+                out.writeBoolean(remoteStoreIndexShallowCopy);
+            }
         }
 
         @Override

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -1473,6 +1473,12 @@ public GatedCloseable<IndexCommit> acquireLastIndexCommit(boolean flushFirst) th
         }
     }
 
+    public GatedCloseable<IndexCommit> acquireLastIndexCommitAndRefresh(boolean flushFirst) throws EngineException {
+        GatedCloseable<IndexCommit> indexCommit = acquireLastIndexCommit(flushFirst);
+        getEngine().refresh("Snapshot for Remote Store based Shard");
+        return indexCommit;
+    }
+
     public Optional<NRTReplicationEngine> getReplicationEngine() {
         if (getEngine() instanceof NRTReplicationEngine) {
             return Optional.of((NRTReplicationEngine) getEngine());