Writable warm replica relocation/recovery

Signed-off-by: Sandeep Kumawat <skumwt@amazon.com>

Author: Sandeep Kumawat

server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationBaseIT.java

@@ -42,15 +42,14 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static java.util.Arrays.asList;
-import static org.opensearch.test.OpenSearchIntegTestCase.client;
-import static org.opensearch.test.OpenSearchTestCase.assertBusy;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
 
 public class SegmentReplicationBaseIT extends OpenSearchIntegTestCase {
@@ -245,4 +244,12 @@ protected SegmentInfos getLatestSegmentInfos(IndexShard shard) throws IOExceptio
             return closeable.get();
         }
     }
+
+    protected boolean warmIndexSegmentReplicationEnabled() {
+        return Objects.equals(
+            IndexModule.INDEX_STORE_LOCALITY_SETTING.get(indexSettings()).toString(),
+            IndexModule.DataLocalityType.PARTIAL.name()
+        );
+    }
+
 }

server/src/internalClusterTest/java/org/opensearch/indices/replication/WarmIndexRemoteStoreSegmentReplicationIT.java

@@ -16,8 +16,6 @@
 
 import java.nio.file.Path;
 
-import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
-
 /**
  * This class runs Segment Replication Integ test suite with remote store enabled.
  */

server/src/internalClusterTest/java/org/opensearch/remotestore/SegmentReplicationUsingRemoteStoreIT.java

@@ -242,7 +242,8 @@ private IndexMetadata.Builder updateInSyncAllocations(
                     allocationId = RecoverySource.ExistingStoreRecoverySource.FORCED_ALLOCATION_ID;
                 } else {
                     assert (recoverySource instanceof RecoverySource.SnapshotRecoverySource
-                        || recoverySource instanceof RecoverySource.RemoteStoreRecoverySource) : recoverySource;
+                        || recoverySource instanceof RecoverySource.RemoteStoreRecoverySource
+                        || recoverySource instanceof RecoverySource.ExistingStoreRecoverySource) : recoverySource;
                     allocationId = updates.initializedPrimary.allocationId().getId();
                 }
                 // forcing a stale primary resets the in-sync allocations to the singleton set with the stale id

server/src/main/java/org/opensearch/cluster/routing/allocation/IndexMetadataUpdater.java

@@ -33,6 +33,8 @@
 import java.util.Queue;
 import java.util.Set;
 
+import static org.opensearch.action.admin.indices.tiering.TieringUtils.isPartialIndex;
+
 /**
  * A {@link RemoteShardsBalancer} used by the {@link BalancedShardsAllocator} to perform allocation operations
  * for remote shards within the cluster.
@@ -345,7 +347,8 @@ private void unassignIgnoredRemoteShards(RoutingAllocation routingAllocation) {
                 // Remote shards do not have an existing store to recover from and can be recovered from an empty source
                 // to re-fetch any shard blocks from the repository.
                 if (shard.primary()) {
-                    if (RecoverySource.Type.SNAPSHOT.equals(shard.recoverySource().getType()) == false) {
+                    if (RecoverySource.Type.SNAPSHOT.equals(shard.recoverySource().getType()) == false
+                        && !isPartialIndex(allocation.metadata().getIndexSafe(shard.index()))) {
                         unassignedShard = shard.updateUnassigned(shard.unassignedInfo(), RecoverySource.EmptyStoreRecoverySource.INSTANCE);
                     }
                 }

server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/RemoteShardsBalancer.java

@@ -368,6 +368,11 @@ public boolean shouldPeriodicallyFlush() {
     @Override
     public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
         ensureOpen();
+        // Skip flushing for indices with partial locality (warm indices)
+        // For these indices, we don't need to commit as we will sync from the remote store on re-open
+        if (engineConfig.getIndexSettings().isStoreLocalityPartial() == false) {
+            return;
+        }
         // readLock is held here to wait/block any concurrent close that acquires the writeLock.
         try (final ReleasableLock lock = readLock.acquire()) {
             ensureOpen();

server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java

@@ -5110,6 +5110,8 @@ public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal) throws IOE
      */
     public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal, final Runnable onFileSync) throws IOException {
         boolean syncSegmentSuccess = false;
+        // For warm indices, it doesn't make sense to check for local existence of files.
+        overrideLocal = overrideLocal && indexSettings.isStoreLocalityPartial() == false;
         long startTimeMs = System.currentTimeMillis();
         assert indexSettings.isRemoteStoreEnabled() || this.isRemoteSeeded();
         logger.trace("Downloading segments from remote segment store");
@@ -5141,7 +5143,9 @@ public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal, final Runn
             } else {
                 storeDirectory = store.directory();
             }
-            copySegmentFiles(storeDirectory, remoteDirectory, null, uploadedSegments, overrideLocal, onFileSync);
+            if (indexSettings.isStoreLocalityPartial() == false) {
+                copySegmentFiles(storeDirectory, remoteDirectory, null, uploadedSegments, overrideLocal, onFileSync);
+            }
 
             if (remoteSegmentMetadata != null) {
                 final SegmentInfos infosSnapshot = store.buildSegmentInfos(

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

@@ -86,10 +86,10 @@ public String[] listAll() throws IOException {
         logger.trace("Composite Directory[{}]: listAll() called", this::toString);
         String[] localFiles = localDirectory.listAll();
         Set<String> allFiles = new HashSet<>(Arrays.asList(localFiles));
-        String[] remoteFiles = getRemoteFiles();
-        allFiles.addAll(Arrays.asList(remoteFiles));
+        // String[] remoteFiles = getRemoteFiles();
+        // allFiles.addAll(Arrays.asList(remoteFiles));
         logger.trace("Composite Directory[{}]: Local Directory files - {}", this::toString, () -> Arrays.toString(localFiles));
-        logger.trace("Composite Directory[{}]: Remote Directory files - {}", this::toString, () -> Arrays.toString(remoteFiles));
+        // logger.trace("Composite Directory[{}]: Remote Directory files - {}", this::toString, () -> Arrays.toString(remoteFiles));
         Set<String> nonBlockLuceneFiles = allFiles.stream()
             .filter(file -> !FileTypeUtils.isBlockFile(file))
             .collect(Collectors.toUnmodifiableSet());
@@ -113,8 +113,11 @@ public void deleteFile(String name) throws IOException {
         if (FileTypeUtils.isTempFile(name)) {
             localDirectory.deleteFile(name);
         } else if (Arrays.asList(listAll()).contains(name) == false) {
-            throw new NoSuchFileException("File " + name + " not found in directory");
+            logger.debug("The file [{}] does not exist", name);
+            // we should not fail here as localDirectory might not contain this file.
+            // throw new NoSuchFileException("File " + name + " not found in directory");
         } else {
+            localDirectory.deleteFile(name);
             fileCache.remove(getFilePath(name));
         }
     }

server/src/main/java/org/opensearch/index/store/CompositeDirectory.java

@@ -316,7 +316,9 @@ public void logCurrentState() {
                 String file = path.substring(path.lastIndexOf('/'));
                 allFiles += file + " [RefCount: " + entry.getValue().refCount + " , Weight: " + entry.getValue().weight + " ]\n";
             }
-            logger.trace("Cache entries : " + allFiles);
+            if (allFiles.equals("\n") == false) {
+                logger.debug("Cache entries : {}", allFiles);
+            }
         } finally {
             lock.unlock();
         }

server/src/main/java/org/opensearch/index/store/remote/utils/cache/LRUCache.java

@@ -190,8 +190,10 @@ public CacheStats stats() {
     public void logCurrentState() {
         int i = 0;
         for (RefCountedCache<K, V> cache : table) {
-            logger.trace("SegmentedCache " + i);
-            ((LRUCache<K, V>) cache).logCurrentState();
+            if (cache.size() > 0) {
+                logger.trace("SegmentedCache " + i);
+                ((LRUCache<K, V>) cache).logCurrentState();
+            }
             i++;
         }
     }

server/src/main/java/org/opensearch/index/store/remote/utils/cache/SegmentedCache.java

@@ -36,6 +36,7 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
@@ -199,6 +200,12 @@ public void startReplication(ActionListener<Void> listener) {
     private List<StoreFileMetadata> getFiles(CheckpointInfoResponse checkpointInfo) throws IOException {
         cancellableThreads.checkForCancel();
         state.setStage(SegmentReplicationState.Stage.FILE_DIFF);
+
+        // Return an empty list for warm indices, In this case, replica shards don't require downloading files from remote storage
+        // as they can will sync all files from remote in case failure.
+        if (indexShard.indexSettings().isStoreLocalityPartial() == true) {
+            return Collections.emptyList();
+        }
         final Store.RecoveryDiff diff = Store.segmentReplicationDiff(checkpointInfo.getMetadataMap(), indexShard.getSegmentMetadataMap());
         // local files
         final Set<String> localFiles = Set.of(indexShard.store().directory().listAll());