[Snapshot V2] Support pinned timestamp in delete flow (#15256) (#15730)

(cherry picked from commit 5bf34d2)

Signed-off-by: Anshu Agarwal <anshukag@amazon.com>
Co-authored-by: Anshu Agarwal <anshukag@amazon.com>

Author: anshu1106

server/src/internalClusterTest/java/org/opensearch/snapshots/DeleteSnapshotITV2.java

@@ -54,6 +54,7 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.index.store.RemoteSegmentStoreDirectoryFactory;
 import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
 import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.repositories.RepositoriesService;
@@ -101,6 +102,8 @@ public final class TransportCleanupRepositoryAction extends TransportClusterMana
 
     private final RemoteStoreLockManagerFactory remoteStoreLockManagerFactory;
 
+    private final RemoteSegmentStoreDirectoryFactory remoteSegmentStoreDirectoryFactory;
+
     @Override
     protected String executor() {
         return ThreadPool.Names.SAME;
@@ -128,6 +131,11 @@ public TransportCleanupRepositoryAction(
         );
         this.repositoriesService = repositoriesService;
         this.snapshotsService = snapshotsService;
+        this.remoteSegmentStoreDirectoryFactory = new RemoteSegmentStoreDirectoryFactory(
+            () -> repositoriesService,
+            threadPool,
+            remoteStoreSettings.getSegmentsPathFixedPrefix()
+        );
         this.remoteStoreLockManagerFactory = new RemoteStoreLockManagerFactory(
             () -> repositoriesService,
             remoteStoreSettings.getSegmentsPathFixedPrefix()
@@ -291,6 +299,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
                                         repositoryStateId,
                                         snapshotsService.minCompatibleVersion(newState.nodes().getMinNodeVersion(), repositoryData, null),
                                         remoteStoreLockManagerFactory,
+                                        remoteSegmentStoreDirectoryFactory,
                                         ActionListener.wrap(result -> after(null, result), e -> after(e, null))
                                     )
                                 )

server/src/main/java/org/opensearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java

@@ -12,6 +12,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.blobstore.BlobContainer;
 import org.opensearch.common.blobstore.BlobMetadata;
 import org.opensearch.common.collect.Tuple;
@@ -42,6 +43,7 @@
  *
  * @opensearch.internal
  */
+@ExperimentalApi
 public class RemoteStorePinnedTimestampService implements Closeable {
     private static final Logger logger = LogManager.getLogger(RemoteStorePinnedTimestampService.class);
     private static Tuple<Long, Set<Long>> pinnedTimestampsSet = new Tuple<>(-1L, Set.of());

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

@@ -50,9 +50,11 @@
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.snapshots.IndexShardSnapshotStatus;
 import org.opensearch.index.snapshots.blobstore.RemoteStoreShardShallowCopySnapshot;
+import org.opensearch.index.store.RemoteSegmentStoreDirectoryFactory;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
 import org.opensearch.indices.recovery.RecoveryState;
+import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService;
 import org.opensearch.snapshots.SnapshotId;
 import org.opensearch.snapshots.SnapshotInfo;
 
@@ -220,11 +222,59 @@ void deleteSnapshots(
     /**
      * Deletes snapshots and releases respective lock files from remote store repository.
      *
-     * @param snapshotIds                   snapshot ids
-     * @param repositoryStateId             the unique id identifying the state of the repository when the snapshot deletion began
-     * @param repositoryMetaVersion         version of the updated repository metadata to write
-     * @param remoteStoreLockManagerFactory RemoteStoreLockManagerFactory to be used for cleaning up remote store lock files
-     * @param listener                      completion listener
+     * @param snapshotIds                           snapshot ids
+     * @param repositoryStateId                     the unique id identifying the state of the repository when the snapshot deletion began
+     * @param repositoryMetaVersion                 version of the updated repository metadata to write
+     * @param remoteStoreLockManagerFactory         RemoteStoreLockManagerFactory to be used for cleaning up remote store lock files
+     * @param remoteSegmentStoreDirectoryFactory    RemoteSegmentStoreDirectoryFactory to be used for cleaning up remote store segment files
+     * @param remoteStorePinnedTimestampService     service for pinning and unpinning of the timestamp
+     * @param snapshotIdsPinnedTimestampMap         map of snapshots ids and the pinned timestamp
+     * @param isShallowSnapshotV2                   true for shallow snapshots v2
+     * @param listener                              completion listener
+     */
+    default void deleteSnapshotsInternal(
+        Collection<SnapshotId> snapshotIds,
+        long repositoryStateId,
+        Version repositoryMetaVersion,
+        RemoteStoreLockManagerFactory remoteStoreLockManagerFactory,
+        RemoteSegmentStoreDirectoryFactory remoteSegmentStoreDirectoryFactory,
+        RemoteStorePinnedTimestampService remoteStorePinnedTimestampService,
+        Map<SnapshotId, Long> snapshotIdsPinnedTimestampMap,
+        boolean isShallowSnapshotV2,
+        ActionListener<RepositoryData> listener
+    ) {
+        throw new UnsupportedOperationException();
+    }
+
+    /**
+     * Deletes snapshots and unpin the snapshot timestamp using remoteStorePinnedTimestampService
+     *
+     * @param snapshotsWithPinnedTimestamp          map of snapshot ids and the pinned timestamps
+     * @param repositoryStateId                     the unique id identifying the state of the repository when the snapshot deletion began
+     * @param repositoryMetaVersion                 version of the updated repository metadata to write
+     * @param remoteSegmentStoreDirectoryFactory    RemoteSegmentStoreDirectoryFactory to be used for cleaning up remote store segment files
+     * @param remoteStorePinnedTimestampService     service for pinning and unpinning of the timestamp
+     * @param listener                              completion listener
+     */
+    default void deleteSnapshotsWithPinnedTimestamp(
+        Map<SnapshotId, Long> snapshotsWithPinnedTimestamp,
+        long repositoryStateId,
+        Version repositoryMetaVersion,
+        RemoteSegmentStoreDirectoryFactory remoteSegmentStoreDirectoryFactory,
+        RemoteStorePinnedTimestampService remoteStorePinnedTimestampService,
+        ActionListener<RepositoryData> listener
+    ) {
+        throw new UnsupportedOperationException();
+    }
+
+    /**
+     * Deletes snapshots and releases respective lock files from remote store repository
+     *
+     * @param snapshotIds
+     * @param repositoryStateId
+     * @param repositoryMetaVersion
+     * @param remoteStoreLockManagerFactory
+     * @param listener
      */
     default void deleteSnapshotsAndReleaseLockFiles(
         Collection<SnapshotId> snapshotIds,

server/src/main/java/org/opensearch/repositories/Repository.java

@@ -112,6 +112,11 @@ public final class RepositoryData {
      * The indices found in the repository across all snapshots, as a name to {@link IndexId} mapping
      */
     private final Map<String, IndexId> indices;
+
+    public Map<IndexId, List<SnapshotId>> getIndexSnapshots() {
+        return indexSnapshots;
+    }
+
     /**
      * The snapshots that each index belongs to.
      */