Enable Safe Bidirectional CCR via Alias policy on Restore

CHANGELOG.md

@@ -63,6 +63,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix QueryPhaseResultConsumer incomplete callback loops ([#19231](https://github.com/opensearch-project/OpenSearch/pull/19231))
 - Fix the `scaled_float` precision issue ([#19188](https://github.com/opensearch-project/OpenSearch/pull/19188))
 - Fix Using an excessively large reindex slice can lead to a JVM OutOfMemoryError on coordinator.([#18964](https://github.com/opensearch-project/OpenSearch/pull/18964))
+- Add alias write index policy to control writeIndex during restore([#1511](https://github.com/opensearch-project/OpenSearch/pull/19368))
 - [Flaky Test] Fix flaky test in SecureReactorNetty4HttpServerTransportTests with reproducible seed ([#19327](https://github.com/opensearch-project/OpenSearch/pull/19327))
 - Remove unnecessary looping in field data cache clear ([#19116](https://github.com/opensearch-project/OpenSearch/pull/19116))
 - [Flaky Test] Fix flaky test IngestFromKinesisIT.testAllActiveIngestion ([#19380](https://github.com/opensearch-project/OpenSearch/pull/19380))

server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java

@@ -52,6 +52,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 
@@ -129,6 +130,29 @@ private static StorageType fromString(String string) {
     @Nullable // if any snapshot UUID will do
     private String snapshotUuid;
 
+    /**
+     * Alias write index policy for controlling how writeIndex attribute is handled during restore
+     *
+     * @opensearch.api
+     */
+    @PublicApi(since = "3.3.0")
+    public enum AliasWriteIndexPolicy {
+        PRESERVE,
+        STRIP_WRITE_INDEX;
+
+        public static AliasWriteIndexPolicy fromString(String value) {
+            try {
+                return valueOf(value.toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException(
+                    "Unknown alias_write_index_policy [" + value + "]. Valid values are: " + Arrays.toString(values())
+                );
+            }
+        }
+    }
+
+    private AliasWriteIndexPolicy aliasWriteIndexPolicy = AliasWriteIndexPolicy.PRESERVE;
+
     public RestoreSnapshotRequest() {}
 
     /**
@@ -172,6 +196,9 @@ public RestoreSnapshotRequest(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_2_18_0)) {
             renameAliasReplacement = in.readOptionalString();
         }
+        if (in.getVersion().onOrAfter(Version.V_3_3_0)) {
+            aliasWriteIndexPolicy = in.readEnum(AliasWriteIndexPolicy.class);
+        }
     }
 
     @Override
@@ -205,6 +232,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_18_0)) {
             out.writeOptionalString(renameAliasReplacement);
         }
+        if (out.getVersion().onOrAfter(Version.V_3_3_0)) {
+            out.writeEnum(aliasWriteIndexPolicy);
+        }
     }
 
     @Override
@@ -640,6 +670,26 @@ public String getSourceRemoteTranslogRepository() {
         return sourceRemoteTranslogRepository;
     }
 
+    /**
+     * Sets alias write index policy for controlling how writeIndex attribute is handled during restore
+     *
+     * @param policy the policy to apply
+     * @return this request
+     */
+    public RestoreSnapshotRequest aliasWriteIndexPolicy(AliasWriteIndexPolicy policy) {
+        this.aliasWriteIndexPolicy = Objects.requireNonNull(policy);
+        return this;
+    }
+
+    /**
+     * Returns alias write index policy
+     *
+     * @return alias write index policy
+     */
+    public AliasWriteIndexPolicy aliasWriteIndexPolicy() {
+        return aliasWriteIndexPolicy;
+    }
+
     /**
      * Parses restore definition
      *
@@ -729,6 +779,8 @@ public RestoreSnapshotRequest source(Map<String, Object> source) {
                 } else {
                     throw new IllegalArgumentException("malformed source_remote_translog_repository");
                 }
+            } else if ("alias_write_index_policy".equals(name)) {
+                aliasWriteIndexPolicy(AliasWriteIndexPolicy.fromString((String) entry.getValue()));
             } else {
                 if (IndicesOptions.isIndicesOptions(name) == false) {
                     throw new IllegalArgumentException("Unknown parameter " + name);
@@ -786,6 +838,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (sourceRemoteTranslogRepository != null) {
             builder.field("source_remote_translog_repository", sourceRemoteTranslogRepository);
         }
+        builder.field("alias_write_index_policy", aliasWriteIndexPolicy.name().toLowerCase(Locale.ROOT));
         builder.endObject();
         return builder;
     }
@@ -817,7 +870,8 @@ public boolean equals(Object o) {
             && Objects.equals(snapshotUuid, that.snapshotUuid)
             && Objects.equals(storageType, that.storageType)
             && Objects.equals(sourceRemoteStoreRepository, that.sourceRemoteStoreRepository)
-            && Objects.equals(sourceRemoteTranslogRepository, that.sourceRemoteTranslogRepository);
+            && Objects.equals(sourceRemoteTranslogRepository, that.sourceRemoteTranslogRepository)
+            && aliasWriteIndexPolicy == that.aliasWriteIndexPolicy;
         return equals;
     }
 
@@ -840,7 +894,8 @@ public int hashCode() {
             snapshotUuid,
             storageType,
             sourceRemoteStoreRepository,
-            sourceRemoteTranslogRepository
+            sourceRemoteTranslogRepository,
+            aliasWriteIndexPolicy
         );
         result = 31 * result + Arrays.hashCode(indices);
         result = 31 * result + Arrays.hashCode(ignoreIndexSettings);

server/src/main/java/org/opensearch/snapshots/RestoreService.java

@@ -479,7 +479,7 @@ public ClusterState execute(ClusterState currentState) {
                                         // Remove all aliases - they shouldn't be restored
                                         indexMdBuilder.removeAllAliases();
                                     } else {
-                                        applyAliasesWithRename(snapshotIndexMetadata, indexMdBuilder, aliases);
+                                        applyAliasesWithRename(snapshotIndexMetadata.getAliases(), request, indexMdBuilder, aliases);
                                     }
                                     IndexMetadata updatedIndexMetadata = indexMdBuilder.build();
                                     if (partial) {
@@ -524,7 +524,7 @@ public ClusterState execute(ClusterState currentState) {
                                             indexMdBuilder.putAlias(alias);
                                         }
                                     } else {
-                                        applyAliasesWithRename(snapshotIndexMetadata, indexMdBuilder, aliases);
+                                        applyAliasesWithRename(snapshotIndexMetadata.getAliases(), request, indexMdBuilder, aliases);
                                     }
                                     final Settings.Builder indexSettingsBuilder = Settings.builder()
                                         .put(snapshotIndexMetadata.getSettings())
@@ -655,22 +655,32 @@ private void checkAliasNameConflicts(Map<String, String> renamedIndices, Set<Str
                     }
 
                     private void applyAliasesWithRename(
-                        IndexMetadata snapshotIndexMetadata,
+                        Map<String, AliasMetadata> snapshotAliases,
+                        RestoreSnapshotRequest request,
                         IndexMetadata.Builder indexMdBuilder,
                         Set<String> aliases
                     ) {
                         if (request.renameAliasPattern() == null || request.renameAliasReplacement() == null) {
-                            aliases.addAll(snapshotIndexMetadata.getAliases().keySet());
+                            for (final Map.Entry<String, AliasMetadata> alias : snapshotAliases.entrySet()) {
+                                AliasMetadata transformedAlias = applyAliasWriteIndexPolicy(
+                                    alias.getValue(),
+                                    request.aliasWriteIndexPolicy()
+                                );
+                                indexMdBuilder.removeAlias(alias.getKey());
+                                indexMdBuilder.putAlias(transformedAlias);
+                                aliases.add(transformedAlias.alias());
+                            }
                         } else {
                             Pattern renameAliasPattern = Pattern.compile(request.renameAliasPattern());
-                            for (final Map.Entry<String, AliasMetadata> alias : snapshotIndexMetadata.getAliases().entrySet()) {
+                            for (final Map.Entry<String, AliasMetadata> alias : snapshotAliases.entrySet()) {
                                 String currentAliasName = alias.getKey();
                                 indexMdBuilder.removeAlias(currentAliasName);
                                 String newAliasName = renameAliasPattern.matcher(currentAliasName)
                                     .replaceAll(request.renameAliasReplacement());
-                                AliasMetadata newAlias = AliasMetadata.newAliasMetadata(alias.getValue(), newAliasName);
-                                indexMdBuilder.putAlias(newAlias);
-                                aliases.add(newAliasName);
+                                AliasMetadata renamedAlias = AliasMetadata.newAliasMetadata(alias.getValue(), newAliasName);
+                                AliasMetadata transformedAlias = applyAliasWriteIndexPolicy(renamedAlias, request.aliasWriteIndexPolicy());
+                                indexMdBuilder.putAlias(transformedAlias);
+                                aliases.add(transformedAlias.alias());
                             }
                         }
                     }
@@ -1369,6 +1379,23 @@ public void applyClusterState(ClusterChangedEvent event) {
         }
     }
 
+    /**
+     * Apply alias write index policy to transform alias metadata during restore.
+     * Package-private for testing.
+     */
+    static AliasMetadata applyAliasWriteIndexPolicy(AliasMetadata aliasMd, RestoreSnapshotRequest.AliasWriteIndexPolicy policy) {
+        if (policy == RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX && Boolean.TRUE.equals(aliasMd.writeIndex())) {
+            return AliasMetadata.builder(aliasMd.alias())
+                .filter(aliasMd.filter())
+                .indexRouting(aliasMd.indexRouting())
+                .searchRouting(aliasMd.searchRouting())
+                .isHidden(aliasMd.isHidden())
+                .writeIndex(false)
+                .build();
+        }
+        return aliasMd;
+    }
+
     private static IndexMetadata addSnapshotToIndexSettings(IndexMetadata metadata, Snapshot snapshot, IndexId indexId) {
         final Settings newSettings = Settings.builder()
             .put(metadata.getSettings())

server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/AliasWriteIndexPolicyRequestTests.java

@@ -0,0 +1,106 @@
+/*
+ * 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.cluster.snapshots.restore;
+
+import org.opensearch.common.io.stream.BytesStreamOutput;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.xcontent.ToXContent;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.equalTo;
+
+public class AliasWriteIndexPolicyRequestTests extends OpenSearchTestCase {
+
+    public void testEqualsAndHashCode() {
+        RestoreSnapshotRequest request1 = new RestoreSnapshotRequest("repo", "snapshot").indices("index1", "index2")
+            .aliasWriteIndexPolicy(RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX);
+
+        RestoreSnapshotRequest request2 = new RestoreSnapshotRequest("repo", "snapshot").indices("index1", "index2")
+            .aliasWriteIndexPolicy(RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX);
+
+        assertEquals(request1, request2);
+        assertEquals(request1.hashCode(), request2.hashCode());
+
+        // Test with different policy
+        RestoreSnapshotRequest request3 = new RestoreSnapshotRequest("repo", "snapshot").indices("index1", "index2")
+            .aliasWriteIndexPolicy(RestoreSnapshotRequest.AliasWriteIndexPolicy.PRESERVE);
+
+        assertNotEquals(request1, request3);
+    }
+
+    public void testSerialization() throws IOException {
+        RestoreSnapshotRequest request = new RestoreSnapshotRequest("test-repo", "test-snapshot").indices("index1", "index2")
+            .renamePattern("(.+)")
+            .renameReplacement("restored-$1")
+            .aliasWriteIndexPolicy(RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX)
+            .includeAliases(true)
+            .partial(true)
+            .waitForCompletion(false);
+
+        BytesStreamOutput out = new BytesStreamOutput();
+        request.writeTo(out);
+
+        StreamInput in = out.bytes().streamInput();
+        RestoreSnapshotRequest deserialized = new RestoreSnapshotRequest(in);
+
+        assertEquals(request.repository(), deserialized.repository());
+        assertEquals(request.snapshot(), deserialized.snapshot());
+        assertArrayEquals(request.indices(), deserialized.indices());
+        assertEquals(request.aliasWriteIndexPolicy(), deserialized.aliasWriteIndexPolicy());
+        assertEquals(request.includeAliases(), deserialized.includeAliases());
+    }
+
+    public void testXContentRoundTrip() throws IOException {
+        RestoreSnapshotRequest request = new RestoreSnapshotRequest("test-repo", "test-snapshot").indices("index1", "index2")
+            .aliasWriteIndexPolicy(RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX)
+            .includeAliases(true)
+            .includeGlobalState(false)
+            .partial(true);
+
+        // Convert to XContent
+        XContentBuilder builder = jsonBuilder();
+        request.toXContent(builder, ToXContent.EMPTY_PARAMS);
+
+        // Parse from XContent
+        XContentParser parser = createParser(builder);
+        Map<String, Object> source = parser.map();
+        RestoreSnapshotRequest parsed = new RestoreSnapshotRequest().source(source);
+
+        // Verify key fields
+        assertArrayEquals(request.indices(), parsed.indices());
+        assertEquals(request.aliasWriteIndexPolicy(), parsed.aliasWriteIndexPolicy());
+        assertEquals(request.includeAliases(), parsed.includeAliases());
+        assertEquals(request.includeGlobalState(), parsed.includeGlobalState());
+        assertEquals(request.partial(), parsed.partial());
+    }
+
+    public void testPolicyFromString() {
+        assertEquals(
+            RestoreSnapshotRequest.AliasWriteIndexPolicy.PRESERVE,
+            RestoreSnapshotRequest.AliasWriteIndexPolicy.fromString("preserve")
+        );
+        assertEquals(
+            RestoreSnapshotRequest.AliasWriteIndexPolicy.STRIP_WRITE_INDEX,
+            RestoreSnapshotRequest.AliasWriteIndexPolicy.fromString("strip_write_index")
+        );
+
+        expectThrows(IllegalArgumentException.class, () -> RestoreSnapshotRequest.AliasWriteIndexPolicy.fromString("invalid"));
+    }
+
+    public void testDefaultValues() {
+        RestoreSnapshotRequest request = new RestoreSnapshotRequest();
+        assertThat(request.aliasWriteIndexPolicy(), equalTo(RestoreSnapshotRequest.AliasWriteIndexPolicy.PRESERVE));
+    }
+}