[Snapshot V2] Support pinned timestamp in delete flow

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

@@ -18,11 +18,13 @@
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeUnit;
 import org.opensearch.index.store.RemoteBufferedOutputDirectory;
 import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.repositories.fs.FsRepository;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.nio.file.Path;
@@ -38,6 +40,8 @@
 import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.comparesEqualTo;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
@@ -370,6 +374,259 @@ public void testRemoteStoreCleanupForDeletedIndex() throws Exception {
         }, 30, TimeUnit.SECONDS);
     }
 
+    public void testDeleteShallowCopyV2() throws Exception {
+        disableRepoConsistencyCheck("Remote store repository is being used in the test");
+
+        final Path remoteStoreRepoPath = randomRepoPath();
+        internalCluster().startClusterManagerOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+
+        internalCluster().startDataOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+        internalCluster().startDataOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+
+        String indexName1 = "testindex1";
+        String indexName2 = "testindex2";
+        String indexName3 = "testindex3";
+        String snapshotRepoName = "test-create-snapshot-repo";
+        String snapshotName1 = "test-create-snapshot1";
+        String snapshotName2 = "test-create-snapshot2";
+        Path absolutePath1 = randomRepoPath().toAbsolutePath();
+        logger.info("Snapshot Path [{}]", absolutePath1);
+
+        Client client = client();
+
+        assertAcked(
+            client.admin()
+                .cluster()
+                .preparePutRepository(snapshotRepoName)
+                .setType(FsRepository.TYPE)
+                .setSettings(
+                    Settings.builder()
+                        .put(FsRepository.LOCATION_SETTING.getKey(), absolutePath1)
+                        .put(FsRepository.COMPRESS_SETTING.getKey(), randomBoolean())
+                        .put(FsRepository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
+                        .put(BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY.getKey(), true)
+                        .put(BlobStoreRepository.SHALLOW_SNAPSHOT_V2.getKey(), true)
+                )
+        );
+
+        createIndex(indexName1, getRemoteStoreBackedIndexSettings());
+        createIndex(indexName2, getRemoteStoreBackedIndexSettings());
+
+        final int numDocsInIndex1 = 10;
+        final int numDocsInIndex2 = 20;
+        indexRandomDocs(indexName1, numDocsInIndex1);
+        indexRandomDocs(indexName2, numDocsInIndex2);
+        ensureGreen(indexName1, indexName2);
+
+        CreateSnapshotResponse createSnapshotResponse = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, snapshotName1)
+            .get();
+        SnapshotInfo snapshotInfo = createSnapshotResponse.getSnapshotInfo();
+        assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
+        assertThat(snapshotInfo.successfulShards(), greaterThan(0));
+        assertThat(snapshotInfo.successfulShards(), equalTo(snapshotInfo.totalShards()));
+        assertThat(snapshotInfo.snapshotId().getName(), equalTo(snapshotName1));
+
+        createIndex(indexName3, getRemoteStoreBackedIndexSettings());
+        indexRandomDocs(indexName3, 10);
+        CreateSnapshotResponse createSnapshotResponse2 = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, snapshotName2)
+            .setWaitForCompletion(true)
+            .get();
+        snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
+        assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
+        assertThat(snapshotInfo.successfulShards(), greaterThan(0));
+        assertThat(snapshotInfo.successfulShards(), equalTo(snapshotInfo.totalShards()));
+        assertThat(snapshotInfo.snapshotId().getName(), equalTo(snapshotName2));
+
+        assertAcked(client().admin().indices().prepareDelete(indexName1));
+        Thread.sleep(100);
+
+        AcknowledgedResponse deleteResponse = client().admin()
+            .cluster()
+            .prepareDeleteSnapshot(snapshotRepoName, snapshotName2)
+            .setSnapshots(snapshotName2)
+            .get();
+        assertTrue(deleteResponse.isAcknowledged());
+
+        // test delete non-existent snapshot
+        assertThrows(
+            SnapshotMissingException.class,
+            () -> client().admin().cluster().prepareDeleteSnapshot(snapshotRepoName, "random-snapshot").setSnapshots(snapshotName2).get()
+        );
+
+    }
+
+    public void testDeleteShallowCopyV2MultipleSnapshots() throws Exception {
+        disableRepoConsistencyCheck("Remote store repository is being used in the test");
+        final Path remoteStoreRepoPath = randomRepoPath();
+
+        internalCluster().startClusterManagerOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+        internalCluster().startDataOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+        internalCluster().startDataOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+
+        String indexName1 = "testindex1";
+        String indexName2 = "testindex2";
+        String indexName3 = "testindex3";
+        String snapshotRepoName = "test-create-snapshot-repo";
+        String snapshotName1 = "test-create-snapshot1";
+        String snapshotName2 = "test-create-snapshot2";
+        Path absolutePath1 = randomRepoPath().toAbsolutePath();
+        logger.info("Snapshot Path [{}]", absolutePath1);
+
+        Client client = client();
+
+        assertAcked(
+            client.admin()
+                .cluster()
+                .preparePutRepository(snapshotRepoName)
+                .setType(FsRepository.TYPE)
+                .setSettings(
+                    Settings.builder()
+                        .put(FsRepository.LOCATION_SETTING.getKey(), absolutePath1)
+                        .put(FsRepository.COMPRESS_SETTING.getKey(), randomBoolean())
+                        .put(FsRepository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
+                        .put(BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY.getKey(), true)
+                        .put(BlobStoreRepository.SHALLOW_SNAPSHOT_V2.getKey(), true)
+                )
+        );
+
+        createIndex(indexName1, getRemoteStoreBackedIndexSettings());
+
+        createIndex(indexName2, getRemoteStoreBackedIndexSettings());
+
+        final int numDocsInIndex1 = 10;
+        final int numDocsInIndex2 = 20;
+        indexRandomDocs(indexName1, numDocsInIndex1);
+        indexRandomDocs(indexName2, numDocsInIndex2);
+        ensureGreen(indexName1, indexName2);
+
+        CreateSnapshotResponse createSnapshotResponse = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, snapshotName1)
+            .get();
+        SnapshotInfo snapshotInfo = createSnapshotResponse.getSnapshotInfo();
+        assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
+        assertThat(snapshotInfo.successfulShards(), greaterThan(0));
+        assertThat(snapshotInfo.successfulShards(), equalTo(snapshotInfo.totalShards()));
+        assertThat(snapshotInfo.snapshotId().getName(), equalTo(snapshotName1));
+
+        createIndex(indexName3, getRemoteStoreBackedIndexSettings());
+        indexRandomDocs(indexName3, 10);
+
+        CreateSnapshotResponse createSnapshotResponse2 = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, snapshotName2)
+            .setWaitForCompletion(true)
+            .get();
+        snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
+        assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
+        assertThat(snapshotInfo.successfulShards(), greaterThan(0));
+        assertThat(snapshotInfo.successfulShards(), equalTo(snapshotInfo.totalShards()));
+        assertThat(snapshotInfo.snapshotId().getName(), equalTo(snapshotName2));
+
+        AcknowledgedResponse deleteResponse = client().admin()
+            .cluster()
+            .prepareDeleteSnapshot(snapshotRepoName, snapshotName1, snapshotName2)
+            .setSnapshots(snapshotName2)
+            .get();
+        assertTrue(deleteResponse.isAcknowledged());
+
+        // test delete non-existent snapshot
+        assertThrows(
+            SnapshotMissingException.class,
+            () -> client().admin().cluster().prepareDeleteSnapshot(snapshotRepoName, "random-snapshot").setSnapshots(snapshotName2).get()
+        );
+
+    }
+
+    public void testRemoteStoreCleanupForDeletedIndexForSnapshotV2() throws Exception {
+        disableRepoConsistencyCheck("Remote store repository is being used in the test");
+        final Path remoteStoreRepoPath = randomRepoPath();
+        Path absolutePath1 = randomRepoPath().toAbsolutePath();
+        Settings settings = remoteStoreClusterSettings(REMOTE_REPO_NAME, remoteStoreRepoPath);
+
+        internalCluster().startClusterManagerOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+        internalCluster().startDataOnlyNode(snapshotV2Settings(remoteStoreRepoPath));
+        final Client clusterManagerClient = internalCluster().clusterManagerClient();
+        ensureStableCluster(2);
+
+        final String snapshotRepoName = "snapshot-repo-name";
+        final Path snapshotRepoPath = randomRepoPath();
+        assertAcked(
+            client().admin()
+                .cluster()
+                .preparePutRepository(snapshotRepoName)
+                .setType(FsRepository.TYPE)
+                .setSettings(
+                    Settings.builder()
+                        .put(FsRepository.LOCATION_SETTING.getKey(), absolutePath1)
+                        .put(FsRepository.COMPRESS_SETTING.getKey(), randomBoolean())
+                        .put(FsRepository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
+                        .put(BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY.getKey(), true)
+                        .put(BlobStoreRepository.SHALLOW_SNAPSHOT_V2.getKey(), true)
+                )
+        );
+
+        final String remoteStoreEnabledIndexName = "remote-index-1";
+        final Settings remoteStoreEnabledIndexSettings = getRemoteStoreBackedIndexSettings();
+        createIndex(remoteStoreEnabledIndexName, remoteStoreEnabledIndexSettings);
+        indexRandomDocs(remoteStoreEnabledIndexName, randomIntBetween(5, 10));
+
+        String indexUUID = client().admin()
+            .indices()
+            .prepareGetSettings(remoteStoreEnabledIndexName)
+            .get()
+            .getSetting(remoteStoreEnabledIndexName, IndexMetadata.SETTING_INDEX_UUID);
+
+        logger.info("--> create two remote index shallow snapshots");
+        CreateSnapshotResponse createSnapshotResponse = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, "snap1")
+            .setWaitForCompletion(true)
+            .get();
+        SnapshotInfo snapshotInfo1 = createSnapshotResponse.getSnapshotInfo();
+
+        CreateSnapshotResponse createSnapshotResponse2 = client().admin()
+            .cluster()
+            .prepareCreateSnapshot(snapshotRepoName, "snap2")
+            .setWaitForCompletion(true)
+            .get();
+        SnapshotInfo snapshotInfo2 = createSnapshotResponse2.getSnapshotInfo();
+        assertThat(snapshotInfo2.state(), equalTo(SnapshotState.SUCCESS));
+        assertThat(snapshotInfo2.successfulShards(), greaterThan(0));
+        assertThat(snapshotInfo2.successfulShards(), equalTo(snapshotInfo2.totalShards()));
+        assertThat(snapshotInfo2.snapshotId().getName(), equalTo("snap2"));
+        final RepositoriesService repositoriesService = internalCluster().getCurrentClusterManagerNodeInstance(RepositoriesService.class);
+
+        // delete remote store index
+        assertAcked(client().admin().indices().prepareDelete(remoteStoreEnabledIndexName));
+
+        logger.info("--> delete snapshot 1");
+        AcknowledgedResponse deleteSnapshotResponse = clusterManagerClient.admin()
+            .cluster()
+            .prepareDeleteSnapshot(snapshotRepoName, snapshotInfo1.snapshotId().getName())
+            .get();
+        assertAcked(deleteSnapshotResponse);
+
+        logger.info("--> delete snapshot 2");
+        deleteSnapshotResponse = clusterManagerClient.admin()
+            .cluster()
+            .prepareDeleteSnapshot(snapshotRepoName, snapshotInfo2.snapshotId().getName())
+            .get();
+        assertAcked(deleteSnapshotResponse);
+
+        Path indexPath = Path.of(String.valueOf(remoteStoreRepoPath), indexUUID);
+        // Delete is async. Give time for it
+        assertBusy(() -> {
+            try {
+                assertThat(RemoteStoreBaseIntegTestCase.getFileCount(indexPath), comparesEqualTo(0));
+            } catch (Exception e) {}
+        }, 30, TimeUnit.SECONDS);
+    }
+
     private List<String> createNSnapshots(String repoName, int count) {
         final List<String> snapshotNames = new ArrayList<>(count);
         final String prefix = "snap-" + UUIDs.randomBase64UUID(random()).toLowerCase(Locale.ROOT) + "-";
@@ -381,4 +638,12 @@ private List<String> createNSnapshots(String repoName, int count) {
         logger.info("--> created {} in [{}]", snapshotNames, repoName);
         return snapshotNames;
     }
+
+    private Settings snapshotV2Settings(Path remoteStoreRepoPath) {
+        Settings settings = Settings.builder()
+            .put(remoteStoreClusterSettings(REMOTE_REPO_NAME, remoteStoreRepoPath))
+            .put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true)
+            .build();
+        return settings;
+    }
 }

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

@@ -52,6 +52,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.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
@@ -96,6 +97,8 @@ public final class TransportCleanupRepositoryAction extends TransportClusterMana
 
     private final RemoteStoreLockManagerFactory remoteStoreLockManagerFactory;
 
+    private final RemoteSegmentStoreDirectoryFactory remoteSegmentStoreDirectoryFactory;
+
     @Override
     protected String executor() {
         return ThreadPool.Names.SAME;
@@ -123,6 +126,7 @@ public TransportCleanupRepositoryAction(
         this.repositoriesService = repositoriesService;
         this.snapshotsService = snapshotsService;
         this.remoteStoreLockManagerFactory = new RemoteStoreLockManagerFactory(() -> repositoriesService);
+        this.remoteSegmentStoreDirectoryFactory = new RemoteSegmentStoreDirectoryFactory(() -> repositoriesService, threadPool);
         // We add a state applier that will remove any dangling repository cleanup actions on cluster-manager failover.
         // This is safe to do since cleanups will increment the repository state id before executing any operations to prevent concurrent
         // operations from corrupting the repository. This is the same safety mechanism used by snapshot deletes.
@@ -272,6 +276,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/common/blobstore/BlobMetadata.java

@@ -32,11 +32,14 @@
 
 package org.opensearch.common.blobstore;
 
+import org.opensearch.common.annotation.ExperimentalApi;
+
 /**
  * An interface for providing basic metadata about a blob.
  *
  * @opensearch.internal
  */
+@ExperimentalApi
 public interface BlobMetadata {
 
     /**

server/src/main/java/org/opensearch/common/blobstore/stream/write/WritePriority.java

@@ -8,11 +8,14 @@
 
 package org.opensearch.common.blobstore.stream.write;
 
+import org.opensearch.common.annotation.ExperimentalApi;
+
 /**
  * WritePriority for upload
  *
  * @opensearch.internal
  */
+@ExperimentalApi
 public enum WritePriority {
     // Used for segment transfers during refresh, flush or merges
     NORMAL,

server/src/main/java/org/opensearch/common/remote/BlobPathParameters.java

@@ -8,12 +8,15 @@
 
 package org.opensearch.common.remote;
 
+import org.opensearch.common.annotation.ExperimentalApi;
+
 import java.util.List;
 
 /**
  * Parameters which can be used to construct a blob path
  *
  */
+@ExperimentalApi
 public class BlobPathParameters {
 
     private final List<String> pathTokens;

server/src/main/java/org/opensearch/common/remote/RemoteWriteableBlobEntity.java

@@ -8,6 +8,7 @@
 
 package org.opensearch.common.remote;
 
+import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.core.compress.Compressor;
 
@@ -17,6 +18,7 @@
  * The abstract class which represents a {@link RemoteWriteableEntity} that can be written to a store
  * @param <T> the entity to be written
  */
+@ExperimentalApi
 public abstract class RemoteWriteableBlobEntity<T> implements RemoteWriteableEntity<T> {
 
     protected String blobFileName;