From 99e3dddfe69ed40486a47ae81e90347bf08f4721 Mon Sep 17 00:00:00 2001
From: Sachin Kale <kalsac@amazon.com>
Date: Mon, 23 Oct 2023 17:02:44 +0530
Subject: [PATCH] Add support to create empty local translog if remote translog
 is empty

Signed-off-by: Sachin Kale <kalsac@amazon.com>
---
 .../restore/RestoreRemoteStoreRequest.java    | 32 ++++++++++++++++++-
 .../cluster/routing/RecoverySource.java       | 30 +++++++++++++++--
 .../opensearch/gateway/GatewayMetaState.java  |  1 +
 .../recovery/RemoteStoreRestoreService.java   | 17 +++++++---
 .../opensearch/index/shard/StoreRecovery.java | 13 ++++++--
 .../GatewayMetaStatePersistedStateTests.java  | 18 +++++++----
 6 files changed, 95 insertions(+), 16 deletions(-)

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/remotestore/restore/RestoreRemoteStoreRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/remotestore/restore/RestoreRemoteStoreRequest.java
index 3b090415c175b..60928d641c2ab 100644
--- a/server/src/main/java/org/opensearch/action/admin/cluster/remotestore/restore/RestoreRemoteStoreRequest.java
+++ b/server/src/main/java/org/opensearch/action/admin/cluster/remotestore/restore/RestoreRemoteStoreRequest.java
@@ -8,6 +8,7 @@
 
 package org.opensearch.action.admin.cluster.remotestore.restore;
 
+import org.opensearch.Version;
 import org.opensearch.action.ActionRequestValidationException;
 import org.opensearch.action.support.IndicesOptions;
 import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest;
@@ -37,6 +38,7 @@ public class RestoreRemoteStoreRequest extends ClusterManagerNodeRequest<Restore
     private String[] indices = Strings.EMPTY_ARRAY;
     private Boolean waitForCompletion = false;
     private Boolean restoreAllShards = false;
+    private Boolean forceEmptyTranslog = false;
 
     public RestoreRemoteStoreRequest() {}
 
@@ -45,6 +47,9 @@ public RestoreRemoteStoreRequest(StreamInput in) throws IOException {
         indices = in.readStringArray();
         waitForCompletion = in.readOptionalBoolean();
         restoreAllShards = in.readOptionalBoolean();
+        if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+            forceEmptyTranslog = in.readOptionalBoolean();
+        }
     }
 
     @Override
@@ -53,6 +58,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeStringArray(indices);
         out.writeOptionalBoolean(waitForCompletion);
         out.writeOptionalBoolean(restoreAllShards);
+        if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+            out.writeOptionalBoolean(forceEmptyTranslog);
+        }
     }
 
     @Override
@@ -142,6 +150,27 @@ public boolean restoreAllShards() {
         return restoreAllShards;
     }
 
+    /**
+     * Set the value for forceEmptyTranslog, denoting whether to create empty translog if remote translog does not have any data.
+     *
+     * @param forceEmptyTranslog If true and if remote translog does not have any data, the operation will create empty translog on local
+     *                           If false, the operation will always try to fetch data from remote translog and will fail if remote translog is empty.
+     * @return this request
+     */
+    public RestoreRemoteStoreRequest forceEmptyTranslog(boolean forceEmptyTranslog) {
+        this.forceEmptyTranslog = forceEmptyTranslog;
+        return this;
+    }
+
+    /**
+     * Returns forceEmptyTranslog setting
+     *
+     * @return true if the operation will create empty translog on local when remote translog is empty
+     */
+    public boolean forceEmptyTranslog() {
+        return forceEmptyTranslog;
+    }
+
     /**
      * Parses restore definition
      *
@@ -193,12 +222,13 @@ public boolean equals(Object o) {
         RestoreRemoteStoreRequest that = (RestoreRemoteStoreRequest) o;
         return waitForCompletion == that.waitForCompletion
             && restoreAllShards == that.restoreAllShards
+            && forceEmptyTranslog == that.forceEmptyTranslog
             && Arrays.equals(indices, that.indices);
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(waitForCompletion, restoreAllShards);
+        int result = Objects.hash(waitForCompletion, restoreAllShards, forceEmptyTranslog);
         result = 31 * result + Arrays.hashCode(indices);
         return result;
     }
diff --git a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
index b01e074ce40c2..a9a1cf8d9ff90 100644
--- a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
+++ b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
@@ -419,17 +419,28 @@ public static class RemoteStoreRecoverySource extends RecoverySource {
         private final String restoreUUID;
         private final IndexId index;
         private final Version version;
+        private final boolean forceEmptyTranslog;
 
         public RemoteStoreRecoverySource(String restoreUUID, Version version, IndexId indexId) {
+            this(restoreUUID, version, indexId, false);
+        }
+
+        public RemoteStoreRecoverySource(String restoreUUID, Version version, IndexId indexId, boolean forceEmptyTranslog) {
             this.restoreUUID = restoreUUID;
             this.version = Objects.requireNonNull(version);
             this.index = Objects.requireNonNull(indexId);
+            this.forceEmptyTranslog = forceEmptyTranslog;
         }
 
         RemoteStoreRecoverySource(StreamInput in) throws IOException {
             restoreUUID = in.readString();
             version = in.readVersion();
             index = new IndexId(in);
+            if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+                forceEmptyTranslog = in.readBoolean();
+            } else {
+                forceEmptyTranslog = false;
+            }
         }
 
         public String restoreUUID() {
@@ -450,11 +461,18 @@ public Version version() {
             return version;
         }
 
+        public boolean forceEmptyTranslog() {
+            return forceEmptyTranslog;
+        }
+
         @Override
         protected void writeAdditionalFields(StreamOutput out) throws IOException {
             out.writeString(restoreUUID);
             out.writeVersion(version);
             index.writeTo(out);
+            if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+                out.writeBoolean(forceEmptyTranslog);
+            }
         }
 
         @Override
@@ -464,7 +482,10 @@ public Type getType() {
 
         @Override
         public void addAdditionalFields(XContentBuilder builder, ToXContent.Params params) throws IOException {
-            builder.field("version", version.toString()).field("index", index.getName()).field("restoreUUID", restoreUUID);
+            builder.field("version", version.toString())
+                .field("index", index.getName())
+                .field("restoreUUID", restoreUUID)
+                .field("forceEmptyTranslog", forceEmptyTranslog);
         }
 
         @Override
@@ -482,12 +503,15 @@ public boolean equals(Object o) {
             }
 
             RemoteStoreRecoverySource that = (RemoteStoreRecoverySource) o;
-            return restoreUUID.equals(that.restoreUUID) && index.equals(that.index) && version.equals(that.version);
+            return restoreUUID.equals(that.restoreUUID)
+                && index.equals(that.index)
+                && version.equals(that.version)
+                && forceEmptyTranslog == that.forceEmptyTranslog;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(restoreUUID, index, version);
+            return Objects.hash(restoreUUID, index, version, forceEmptyTranslog);
         }
 
         // TODO: This override should be removed/be updated to return "true",
diff --git a/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java b/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java
index f855449c708d2..83b82730fae00 100644
--- a/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java
+++ b/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java
@@ -179,6 +179,7 @@ public void start(
                                     ClusterState.builder(clusterState).metadata(Metadata.EMPTY_METADATA).build(),
                                     lastKnownClusterUUID,
                                     false,
+                                    false,
                                     new String[] {}
                                 );
                                 clusterState = remoteRestoreResult.getClusterState();
diff --git a/server/src/main/java/org/opensearch/index/recovery/RemoteStoreRestoreService.java b/server/src/main/java/org/opensearch/index/recovery/RemoteStoreRestoreService.java
index 9541d13421e27..b1d27949d4518 100644
--- a/server/src/main/java/org/opensearch/index/recovery/RemoteStoreRestoreService.java
+++ b/server/src/main/java/org/opensearch/index/recovery/RemoteStoreRestoreService.java
@@ -99,7 +99,13 @@ public void restore(RestoreRemoteStoreRequest request, final ActionListener<Rest
 
             @Override
             public ClusterState execute(ClusterState currentState) {
-                RemoteRestoreResult remoteRestoreResult = restore(currentState, null, request.restoreAllShards(), request.indices());
+                RemoteRestoreResult remoteRestoreResult = restore(
+                    currentState,
+                    null,
+                    request.restoreAllShards(),
+                    request.forceEmptyTranslog(),
+                    request.indices()
+                );
                 restoreUUID = remoteRestoreResult.getRestoreUUID();
                 restoreInfo = remoteRestoreResult.getRestoreInfo();
                 return remoteRestoreResult.getClusterState();
@@ -135,6 +141,7 @@ public RemoteRestoreResult restore(
         ClusterState currentState,
         @Nullable String restoreClusterUUID,
         boolean restoreAllShards,
+        boolean forceEmptyTranslog,
         String[] indexNames
     ) {
         Map<String, Tuple<Boolean, IndexMetadata>> indexMetadataMap = new HashMap<>();
@@ -176,7 +183,7 @@ public RemoteRestoreResult restore(
                 }
             }
         }
-        return executeRestore(currentState, indexMetadataMap, restoreAllShards, remoteMetadata);
+        return executeRestore(currentState, indexMetadataMap, restoreAllShards, remoteMetadata, forceEmptyTranslog);
     }
 
     /**
@@ -190,7 +197,8 @@ private RemoteRestoreResult executeRestore(
         ClusterState currentState,
         Map<String, Tuple<Boolean, IndexMetadata>> indexMetadataMap,
         boolean restoreAllShards,
-        Metadata remoteMetadata
+        Metadata remoteMetadata,
+        boolean forceEmptyTranslog
     ) {
         final String restoreUUID = UUIDs.randomBase64UUID();
         List<String> indicesToBeRestored = new ArrayList<>();
@@ -228,7 +236,8 @@ private RemoteRestoreResult executeRestore(
                 RecoverySource.RemoteStoreRecoverySource recoverySource = new RecoverySource.RemoteStoreRecoverySource(
                     restoreUUID,
                     updatedIndexMetadata.getCreationVersion(),
-                    indexId
+                    indexId,
+                    forceEmptyTranslog
                 );
 
                 rtBuilder.addAsRemoteStoreRestore(updatedIndexMetadata, recoverySource, indexShardRoutingTableMap, restoreAllShards);
diff --git a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
index c0211e1257c8e..6ff7a2449dd70 100644
--- a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
+++ b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
@@ -76,6 +76,7 @@
 
 import java.io.IOException;
 import java.nio.channels.FileChannel;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Arrays;
@@ -538,16 +539,24 @@ private void recoverFromRemoteStore(IndexShard indexShard) throws IndexShardReco
             indexShard.syncSegmentsFromRemoteSegmentStore(true);
             indexShard.syncTranslogFilesFromRemoteTranslog();
 
+            Path location = indexShard.shardPath().resolveTranslog();
+
             // On index creation, the only segment file that is created is segments_N. We can safely discard this file
             // as there is no data associated with this shard as part of segments.
-            if (store.directory().listAll().length <= 1) {
-                Path location = indexShard.shardPath().resolveTranslog();
+            boolean remoteSegmentEmpty = store.directory().listAll().length <= 1;
+            boolean remoteTranslogEmpty = Files.exists(location.resolve(CHECKPOINT_FILE_NAME)) == false;
+
+            if (remoteSegmentEmpty && remoteTranslogEmpty == false) {
                 Checkpoint checkpoint = Checkpoint.read(location.resolve(CHECKPOINT_FILE_NAME));
                 final Path translogFile = location.resolve(Translog.getFilename(checkpoint.getGeneration()));
                 try (FileChannel channel = FileChannel.open(translogFile, StandardOpenOption.READ)) {
                     TranslogHeader translogHeader = TranslogHeader.read(translogFile, channel);
                     store.createEmpty(indexShard.indexSettings().getIndexVersionCreated().luceneVersion, translogHeader.getTranslogUUID());
                 }
+            } else if (remoteSegmentEmpty == false && remoteTranslogEmpty) {
+                if (((RecoverySource.RemoteStoreRecoverySource) indexShard.shardRouting.recoverySource()).forceEmptyTranslog()) {
+                    bootstrap(indexShard, store);
+                }
             }
 
             assert indexShard.shardRouting.primary() : "only primary shards can recover from store";
diff --git a/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java b/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java
index 1d5c2a0f01b5c..588b91000fb71 100644
--- a/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java
+++ b/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java
@@ -785,7 +785,7 @@ public void testGatewayForRemoteState() throws IOException {
             RemoteClusterStateService remoteClusterStateService = mock(RemoteClusterStateService.class);
             when(remoteClusterStateService.getLastKnownUUIDFromRemote("test-cluster")).thenReturn("test-cluster-uuid");
             RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class);
-            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn(
+            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), anyBoolean(), any())).thenReturn(
                 RemoteRestoreResult.build("test-cluster-uuid", null, ClusterState.EMPTY_STATE)
             );
             gateway = new MockGatewayMetaState(localNode, bigArrays, remoteClusterStateService, remoteStoreRestoreService);
@@ -832,7 +832,7 @@ public void testGatewayForRemoteStateForInitialBootstrap() throws IOException {
             when(remoteClusterStateService.getLastKnownUUIDFromRemote(clusterName.value())).thenReturn(ClusterState.UNKNOWN_UUID);
 
             final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class);
-            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn(
+            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), anyBoolean(), any())).thenReturn(
                 RemoteRestoreResult.build("test-cluster-uuid", null, ClusterState.EMPTY_STATE)
             );
             final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry();
@@ -879,7 +879,7 @@ public void testGatewayForRemoteStateForNodeReplacement() throws IOException {
             );
 
             final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class);
-            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn(
+            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), anyBoolean(), any())).thenReturn(
                 RemoteRestoreResult.build("test-cluster-uuid", null, previousState)
             );
             final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry();
@@ -893,7 +893,7 @@ public void testGatewayForRemoteStateForNodeReplacement() throws IOException {
             final CoordinationState.PersistedState lucenePersistedState = gateway.getPersistedState();
             PersistedState remotePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE);
             verify(remoteClusterStateService).getLastKnownUUIDFromRemote(Mockito.any());
-            verify(remoteStoreRestoreService).restore(any(), any(), anyBoolean(), any());
+            verify(remoteStoreRestoreService).restore(any(), any(), anyBoolean(), anyBoolean(), any());
             assertThat(remotePersistedState.getLastAcceptedState(), nullValue());
             assertThat(lucenePersistedState.getLastAcceptedState().metadata(), equalTo(previousState.metadata()));
         } finally {
@@ -969,7 +969,7 @@ public void testGatewayForRemoteStateForInitialBootstrapBlocksApplied() throws I
             ).nodes(DiscoveryNodes.EMPTY_NODES).build();
 
             final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class);
-            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn(
+            when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), anyBoolean(), any())).thenReturn(
                 RemoteRestoreResult.build("test-cluster-uuid", null, clusterState)
             );
             final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry();
@@ -983,7 +983,13 @@ public void testGatewayForRemoteStateForInitialBootstrapBlocksApplied() throws I
             PersistedState remotePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE);
             PersistedState lucenePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.LOCAL);
             verify(remoteClusterStateService).getLastKnownUUIDFromRemote(clusterName.value()); // change this
-            verify(remoteStoreRestoreService).restore(any(ClusterState.class), any(String.class), anyBoolean(), any(String[].class));
+            verify(remoteStoreRestoreService).restore(
+                any(ClusterState.class),
+                any(String.class),
+                anyBoolean(),
+                anyBoolean(),
+                any(String[].class)
+            );
             assertThat(remotePersistedState.getLastAcceptedState(), nullValue());
             assertThat(
                 Metadata.isGlobalStateEquals(lucenePersistedState.getLastAcceptedState().metadata(), clusterState.metadata()),