Add support to skip pinned timestamp in remote segment garbage collector

Signed-off-by: Sachin Kale <[email protected]>

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

@@ -217,10 +217,7 @@ public void testStaleCommitDeletionWithInvokeFlush() throws Exception {
             } else {
                 // As delete is async its possible that the file gets created before the deletion or after
                 // deletion.
-                MatcherAssert.assertThat(
-                    actualFileCount,
-                    is(oneOf(lastNMetadataFilesToKeep - 1, lastNMetadataFilesToKeep, lastNMetadataFilesToKeep + 1))
-                );
+                assertTrue(actualFileCount >= lastNMetadataFilesToKeep);
             }
         }, 30, TimeUnit.SECONDS);
     }
@@ -249,7 +246,7 @@ public void testStaleCommitDeletionWithMinSegmentFiles_3() throws Exception {
         Path indexPath = Path.of(segmentRepoPath + "/" + shardPath);
         int actualFileCount = getFileCount(indexPath);
         // We also allow (numberOfIterations + 1) as index creation also triggers refresh.
-        MatcherAssert.assertThat(actualFileCount, is(oneOf(4)));
+        assertTrue(actualFileCount >= 4);
     }
 
     public void testStaleCommitDeletionWithMinSegmentFiles_Disabled() throws Exception {

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

@@ -40,6 +40,7 @@
 import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata;
 import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler;
 import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint;
+import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService;
 import org.opensearch.threadpool.ThreadPool;
 
 import java.io.FileNotFoundException;
@@ -91,6 +92,8 @@ public final class RemoteSegmentStoreDirectory extends FilterDirectory implement
 
     private final RemoteStoreLockManager mdLockManager;
 
+    private final Map<Long, String> metadataFilePinnedTimestampMap;
+
     private final ThreadPool threadPool;
 
     /**
@@ -132,6 +135,7 @@ public RemoteSegmentStoreDirectory(
         this.remoteMetadataDirectory = remoteMetadataDirectory;
         this.mdLockManager = mdLockManager;
         this.threadPool = threadPool;
+        this.metadataFilePinnedTimestampMap = new HashMap<>();
         this.logger = Loggers.getLogger(getClass(), shardId);
         init();
     }
@@ -176,6 +180,38 @@ public RemoteSegmentMetadata initializeToSpecificCommit(long primaryTerm, long c
         return remoteSegmentMetadata;
     }
 
+    /**
+     * Initializes the remote segment metadata to a specific timestamp.
+     *
+     * @param timestamp The timestamp to initialize the remote segment metadata to.
+     * @return The RemoteSegmentMetadata object corresponding to the specified timestamp, or null if no metadata file is found for that timestamp.
+     * @throws IOException If an I/O error occurs while reading the metadata file.
+     */
+    public RemoteSegmentMetadata initializeToSpecificTimestamp(long timestamp) throws IOException {
+        List<String> metadataFiles = remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder(
+            MetadataFilenameUtils.METADATA_PREFIX,
+            Integer.MAX_VALUE
+        );
+        Set<String> lockedMetadataFiles = RemoteStoreUtils.getPinnedTimestampLockedFiles(
+            metadataFiles,
+            Set.of(timestamp),
+            MetadataFilenameUtils::getTimestamp,
+            MetadataFilenameUtils::getNodeIdByPrimaryTermAndGen
+        );
+        if (lockedMetadataFiles.isEmpty()) {
+            return null;
+        }
+        assert lockedMetadataFiles.size() == 1 : "Expected exactly one metadata file but got " + lockedMetadataFiles;
+        String metadataFile = lockedMetadataFiles.iterator().next();
+        RemoteSegmentMetadata remoteSegmentMetadata = readMetadataFile(metadataFile);
+        if (remoteSegmentMetadata != null) {
+            this.segmentsUploadedToRemoteStore = new ConcurrentHashMap<>(remoteSegmentMetadata.getMetadata());
+        } else {
+            this.segmentsUploadedToRemoteStore = new ConcurrentHashMap<>();
+        }
+        return remoteSegmentMetadata;
+    }
+
     /**
      * Read the latest metadata file to get the list of segments uploaded to the remote segment store.
      * We upload a metadata file per refresh, but it is not unique per refresh. Refresh metadata file is unique for a given commit.
@@ -324,7 +360,8 @@ public static String getMetadataFilename(
             long translogGeneration,
             long uploadCounter,
             int metadataVersion,
-            String nodeId
+            String nodeId,
+            long creationTimestamp
         ) {
             return String.join(
                 SEPARATOR,
@@ -334,11 +371,30 @@ public static String getMetadataFilename(
                 RemoteStoreUtils.invertLong(translogGeneration),
                 RemoteStoreUtils.invertLong(uploadCounter),
                 String.valueOf(Objects.hash(nodeId)),
-                RemoteStoreUtils.invertLong(System.currentTimeMillis()),
+                RemoteStoreUtils.invertLong(creationTimestamp),
                 String.valueOf(metadataVersion)
             );
         }
 
+        public static String getMetadataFilename(
+            long primaryTerm,
+            long generation,
+            long translogGeneration,
+            long uploadCounter,
+            int metadataVersion,
+            String nodeId
+        ) {
+            return getMetadataFilename(
+                primaryTerm,
+                generation,
+                translogGeneration,
+                uploadCounter,
+                metadataVersion,
+                nodeId,
+                System.currentTimeMillis()
+            );
+        }
+
         // Visible for testing
         static long getPrimaryTerm(String[] filenameTokens) {
             return RemoteStoreUtils.invertLong(filenameTokens[1]);
@@ -778,6 +834,7 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException
             );
             return;
         }
+
         List<String> sortedMetadataFileList = remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder(
             MetadataFilenameUtils.METADATA_PREFIX,
             Integer.MAX_VALUE
@@ -791,16 +848,44 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException
             return;
         }
 
-        List<String> metadataFilesEligibleToDelete = new ArrayList<>(
-            sortedMetadataFileList.subList(lastNMetadataFilesToKeep, sortedMetadataFileList.size())
+        // Check last fetch status of pinned timestamps. If stale, return.
+        if (RemoteStoreUtils.isPinnedTimestampStateStale()) {
+            logger.warn("Skipping remote segment store garbage collection as last fetch of pinned timestamp is stale");
+            return;
+        }
+
+        Tuple<Long, Set<Long>> pinnedTimestampsState = RemoteStorePinnedTimestampService.getPinnedTimestamps();
+
+        Set<String> implicitLockedFiles = RemoteStoreUtils.getPinnedTimestampLockedFiles(
+            sortedMetadataFileList,
+            pinnedTimestampsState.v2(),
+            metadataFilePinnedTimestampMap,
+            MetadataFilenameUtils::getTimestamp,
+            MetadataFilenameUtils::getNodeIdByPrimaryTermAndGen
         );
-        Set<String> allLockFiles;
+        final Set<String> allLockFiles = new HashSet<>(implicitLockedFiles);
+
         try {
-            allLockFiles = ((RemoteStoreMetadataLockManager) mdLockManager).fetchLockedMetadataFiles(MetadataFilenameUtils.METADATA_PREFIX);
+            allLockFiles.addAll(
+                ((RemoteStoreMetadataLockManager) mdLockManager).fetchLockedMetadataFiles(MetadataFilenameUtils.METADATA_PREFIX)
+            );
         } catch (Exception e) {
             logger.error("Exception while fetching segment metadata lock files, skipping deleteStaleSegments", e);
             return;
         }
+
+        List<String> metadataFilesEligibleToDelete = new ArrayList<>(
+            sortedMetadataFileList.subList(lastNMetadataFilesToKeep, sortedMetadataFileList.size())
+        );
+
+        // Along with last N files, we need to keep files since last successful run of scheduler
+        long lastSuccessfulFetchOfPinnedTimestamps = pinnedTimestampsState.v1();
+        metadataFilesEligibleToDelete = RemoteStoreUtils.filterOutMetadataFilesBasedOnAge(
+            metadataFilesEligibleToDelete,
+            MetadataFilenameUtils::getTimestamp,
+            lastSuccessfulFetchOfPinnedTimestamps
+        );
+
         List<String> metadataFilesToBeDeleted = metadataFilesEligibleToDelete.stream()
             .filter(metadataFile -> allLockFiles.contains(metadataFile) == false)
             .collect(Collectors.toList());

server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java

@@ -245,6 +245,14 @@ public static Tuple<Long, Set<Long>> getPinnedTimestamps() {
         return pinnedTimestampsSet;
     }
 
+    public RemoteStorePinnedTimestampsBlobStore pinnedTimestampsBlobStore() {
+        return pinnedTimestampsBlobStore;
+    }
+
+    public BlobStoreTransferService blobStoreTransferService() {
+        return blobStoreTransferService;
+    }
+
     /**
      * Inner class for asynchronously updating the pinned timestamp set.
      */
@@ -266,11 +274,12 @@ protected void runInternal() {
                 clusterService.state().metadata().clusterUUID(),
                 blobStoreRepository.getCompressor()
             );
-            BlobPath path = pinnedTimestampsBlobStore.getBlobPathForUpload(remotePinnedTimestamps);
-            blobStoreTransferService.listAllInSortedOrder(path, remotePinnedTimestamps.getType(), 1, new ActionListener<>() {
+            BlobPath path = pinnedTimestampsBlobStore().getBlobPathForUpload(remotePinnedTimestamps);
+            blobStoreTransferService().listAllInSortedOrder(path, remotePinnedTimestamps.getType(), 1, new ActionListener<>() {
                 @Override
                 public void onResponse(List<BlobMetadata> blobMetadata) {
                     if (blobMetadata.isEmpty()) {
+                        pinnedTimestampsSet = new Tuple<>(triggerTimestamp, Set.of());
                         return;
                     }
                     PinnedTimestamps pinnedTimestamps = readExistingPinnedTimestamps(blobMetadata.get(0).name(), remotePinnedTimestamps);

server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java

@@ -1081,5 +1081,4 @@ public void testIsPinnedTimestampStateStaleFeatureEnabled() {
         setupRemotePinnedTimestampFeature(true);
         assertTrue(RemoteStoreUtils.isPinnedTimestampStateStale());
     }
-
 }

server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java

@@ -49,7 +49,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase {
         34,
         1,
         1,
-        "node-1"
+        "node-1",
+        System.currentTimeMillis() - 300000
     );
 
     protected final String metadataFilenameDup = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(
@@ -58,31 +59,35 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase {
         34,
         2,
         1,
-        "node-2"
+        "node-2",
+        System.currentTimeMillis() - 400000
     );
     protected final String metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(
         12,
         13,
         34,
         1,
         1,
-        "node-1"
+        "node-1",
+        System.currentTimeMillis() - 500000
     );
     protected final String metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(
         10,
         38,
         34,
         1,
         1,
-        "node-1"
+        "node-1",
+        System.currentTimeMillis() - 600000
     );
     protected final String metadataFilename4 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(
         10,
         36,
         34,
         1,
         1,
-        "node-1"
+        "node-1",
+        System.currentTimeMillis() - 700000
     );
 
     public void setupRemoteSegmentStoreDirectory() throws IOException {