Add restriction to have a single repository with shallow snapshot v2 …

…setting

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

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

@@ -30,6 +30,8 @@
 import java.io.ByteArrayInputStream;
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -75,25 +77,42 @@ public RemoteStorePinnedTimestampService(
      * and starts the asynchronous update task.
      */
     public void start() {
-        validateRemoteStoreConfiguration();
+        blobContainer = validateAndCreateBlobContainer(settings, repositoriesService.get());
         startAsyncUpdateTask(RemoteStoreSettings.getPinnedTimestampsSchedulerInterval());
     }
 
-    private void validateRemoteStoreConfiguration() {
+    private static BlobContainer validateAndCreateBlobContainer(Settings settings, RepositoriesService repositoriesService) {
         final String remoteStoreRepo = settings.get(
             Node.NODE_ATTRIBUTES.getKey() + RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY
         );
         assert remoteStoreRepo != null : "Remote Segment Store repository is not configured";
-        final Repository repository = repositoriesService.get().repository(remoteStoreRepo);
+        final Repository repository = repositoriesService.repository(remoteStoreRepo);
         assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository";
         BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository;
-        blobContainer = blobStoreRepository.blobStore().blobContainer(blobStoreRepository.basePath().add(PINNED_TIMESTAMPS_PATH_TOKEN));
+        return blobStoreRepository.blobStore().blobContainer(blobStoreRepository.basePath().add(PINNED_TIMESTAMPS_PATH_TOKEN));
     }
 
     private void startAsyncUpdateTask(TimeValue pinnedTimestampsSchedulerInterval) {
         asyncUpdatePinnedTimestampTask = new AsyncUpdatePinnedTimestampTask(logger, threadPool, pinnedTimestampsSchedulerInterval, true);
     }
 
+    public static Map<String, Set<Long>> fetchPinnedTimestamps(Settings settings, RepositoriesService repositoriesService)
+        throws IOException {
+        BlobContainer blobContainer = validateAndCreateBlobContainer(settings, repositoriesService);
+        Set<String> pinnedTimestamps = blobContainer.listBlobs().keySet();
+        Map<String, Set<Long>> pinningEntityTimestampMap = new HashMap<>();
+        for (String pinnedTimestamp : pinnedTimestamps) {
+            String[] tokens = pinnedTimestamp.split(PINNED_TIMESTAMPS_FILENAME_SEPARATOR);
+            Long timestamp = Long.parseLong(tokens[tokens.length - 1]);
+            String pinningEntity = pinnedTimestamp.substring(pinnedTimestamp.lastIndexOf(PINNED_TIMESTAMPS_FILENAME_SEPARATOR));
+            if (pinningEntityTimestampMap.containsKey(pinningEntity) == false) {
+                pinningEntityTimestampMap.put(pinningEntity, new HashSet<>());
+            }
+            pinningEntityTimestampMap.get(pinningEntity).add(timestamp);
+        }
+        return pinningEntityTimestampMap;
+    }
+
     /**
      * Pins a timestamp in the remote store.
      *

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

@@ -68,7 +68,9 @@
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.Strings;
+import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService;
 import org.opensearch.repositories.blobstore.MeteredBlobStoreRepository;
+import org.opensearch.snapshots.SnapshotsService;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
@@ -84,6 +86,7 @@
 import java.util.stream.Stream;
 
 import static org.opensearch.repositories.blobstore.BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY;
+import static org.opensearch.repositories.blobstore.BlobStoreRepository.SHALLOW_SNAPSHOT_V2;
 import static org.opensearch.repositories.blobstore.BlobStoreRepository.SYSTEM_REPOSITORY_SETTING;
 
 /**
@@ -123,6 +126,7 @@ public class RepositoriesService extends AbstractLifecycleComponent implements C
     private final RepositoriesStatsArchive repositoriesStatsArchive;
     private final ClusterManagerTaskThrottler.ThrottlingKey putRepositoryTaskKey;
     private final ClusterManagerTaskThrottler.ThrottlingKey deleteRepositoryTaskKey;
+    private final Settings settings;
 
     public RepositoriesService(
         Settings settings,
@@ -132,6 +136,7 @@ public RepositoriesService(
         Map<String, Repository.Factory> internalTypesRegistry,
         ThreadPool threadPool
     ) {
+        this.settings = settings;
         this.typesRegistry = typesRegistry;
         this.internalTypesRegistry = internalTypesRegistry;
         this.clusterService = clusterService;
@@ -173,7 +178,7 @@ public void registerOrUpdateRepository(final PutRepositoryRequest request, final
             CryptoMetadata.fromRequest(request.cryptoSettings())
         );
         validate(request.name());
-        validateRepositoryMetadataSettings(clusterService, request.name(), request.settings());
+        validateRepositoryMetadataSettings(clusterService, request.name(), request.settings(), repositories, settings, this);
         if (newRepositoryMetadata.cryptoMetadata() != null) {
             validate(newRepositoryMetadata.cryptoMetadata().keyProviderName());
         }
@@ -684,7 +689,10 @@ public static void validate(final String identifier) {
     public static void validateRepositoryMetadataSettings(
         ClusterService clusterService,
         final String repositoryName,
-        final Settings repositoryMetadataSettings
+        final Settings repositoryMetadataSettings,
+        Map<String, Repository> repositories,
+        Settings settings,
+        RepositoriesService repositoriesService
     ) {
         // We can add more validations here for repository settings in the future.
         Version minVersionInCluster = clusterService.state().getNodes().getMinNodeVersion();
@@ -699,6 +707,41 @@ public static void validateRepositoryMetadataSettings(
                     + minVersionInCluster
             );
         }
+        if (SHALLOW_SNAPSHOT_V2.get(repositoryMetadataSettings)) {
+            if (minVersionInCluster.onOrAfter(Version.V_2_17_0) == false) {
+                throw new RepositoryException(
+                    repositoryName,
+                    "setting "
+                        + SHALLOW_SNAPSHOT_V2.getKey()
+                        + " cannot be enabled as some of the nodes in cluster are on version older than "
+                        + Version.V_2_17_0
+                        + ". Minimum node version in cluster is: "
+                        + minVersionInCluster
+                );
+            }
+            if (repositoryWithShallowV2Exists(repositories)) {
+                throw new RepositoryException(
+                    repositoryName,
+                    "setting "
+                        + SHALLOW_SNAPSHOT_V2.getKey()
+                        + " cannot be enabled as this setting can be enabled only on one repository "
+                        + " and one or more repositories in the cluster have the setting as enabled"
+                );
+            }
+            try {
+                if (pinnedTimestampExistsWithDifferentRepository(repositoryName, settings, repositoriesService)) {
+                    throw new RepositoryException(
+                        repositoryName,
+                        "setting "
+                            + SHALLOW_SNAPSHOT_V2.getKey()
+                            + " cannot be enabled if there are existing snapshots created with shallow V2 "
+                            + "setting using different repository."
+                    );
+                }
+            } catch (Exception e) {
+                throw new RepositoryException(repositoryName, "Exception while fetching pinned timestamp details");
+            }
+        }
         // Validation to not allow users to create system repository via put repository call.
         if (isSystemRepositorySettingPresent(repositoryMetadataSettings)) {
             throw new RepositoryException(
@@ -710,6 +753,36 @@ public static void validateRepositoryMetadataSettings(
         }
     }
 
+    private static boolean repositoryWithShallowV2Exists(Map<String, Repository> repositories) {
+        return repositories.values().stream().anyMatch(repo -> SHALLOW_SNAPSHOT_V2.get(repo.getMetadata().settings()));
+    }
+
+    private static boolean pinnedTimestampExistsWithDifferentRepository(
+        String newRepoName,
+        Settings settings,
+        RepositoriesService repositoriesService
+    ) throws IOException {
+        Map<String, Set<Long>> pinningEntityTimestampMap = RemoteStorePinnedTimestampService.fetchPinnedTimestamps(
+            settings,
+            repositoriesService
+        );
+        for (String pinningEntiy : pinningEntityTimestampMap.keySet()) {
+            String[] tokens = pinningEntiy.split(SnapshotsService.SNAPSHOT_PINNED_TIMESTAMP_DELIMITER);
+            if (tokens.length > 2) {
+                logger.warn(
+                    "With more than one {} in the pinning entity = {}, not able to determine if there is a pinned timestamp created with different repository",
+                    SnapshotsService.SNAPSHOT_PINNED_TIMESTAMP_DELIMITER,
+                    pinningEntiy
+                );
+                return true;
+            }
+            if (tokens[0].equals(newRepoName) == false) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private static void ensureRepositoryNotInUse(ClusterState clusterState, String repository) {
         if (isRepositoryInUse(clusterState, repository)) {
             throw new IllegalStateException("trying to modify or unregister repository that is currently used");

server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryRemoteIndexTests.java

@@ -33,6 +33,8 @@
 package org.opensearch.repositories.blobstore;
 
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.action.admin.cluster.repositories.verify.VerifyRepositoryResponse;
+import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.common.settings.Settings;
@@ -41,10 +43,12 @@
 import org.opensearch.gateway.remote.RemoteClusterStateService;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.snapshots.blobstore.RemoteStoreShardShallowCopySnapshot;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.repositories.IndexId;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.RepositoryData;
+import org.opensearch.repositories.RepositoryException;
 import org.opensearch.repositories.fs.FsRepository;
 import org.opensearch.snapshots.SnapshotId;
 import org.opensearch.snapshots.SnapshotInfo;
@@ -64,6 +68,9 @@
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
+import static org.opensearch.repositories.blobstore.BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY;
+import static org.opensearch.repositories.blobstore.BlobStoreRepository.SHALLOW_SNAPSHOT_V2;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
 
 /**
@@ -81,6 +88,7 @@ protected Settings nodeSettings() {
             .put(Environment.PATH_HOME_SETTING.getKey(), tempDir)
             .put(Environment.PATH_REPO_SETTING.getKey(), tempDir.resolve("repo"))
             .put(Environment.PATH_SHARED_DATA_SETTING.getKey(), tempDir.getParent())
+            .put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true)
             .build();
     }
 
@@ -373,4 +381,78 @@ public void testRetrieveShallowCopySnapshotCase2() throws IOException {
         assertThat(snapshotIds, equalTo(originalSnapshots));
     }
 
+    public void testRepositoryCreationShallowV2() throws Exception {
+        Client client = client();
+
+        Settings snapshotRepoSettings1 = Settings.builder()
+            .put(node().settings())
+            .put("location", OpenSearchIntegTestCase.randomRepoPath(node().settings()))
+            .put(REMOTE_STORE_INDEX_SHALLOW_COPY.getKey(), true)
+            .put(SHALLOW_SNAPSHOT_V2.getKey(), true)
+            .build();
+
+        // Create repo with shallow snapshot V2 enabled
+        createRepository(client, "test-repo-1", snapshotRepoSettings1);
+
+        logger.info("--> verify the repository");
+        VerifyRepositoryResponse verifyRepositoryResponse = client.admin().cluster().prepareVerifyRepository("test-repo-1").get();
+        assertNotNull(verifyRepositoryResponse.getNodes());
+
+        GetRepositoriesResponse getRepositoriesResponse = client.admin().cluster().prepareGetRepositories("test-repo-1").get();
+        assertTrue(SHALLOW_SNAPSHOT_V2.get(getRepositoriesResponse.repositories().get(0).settings()));
+
+        Settings snapshotRepoSettings2 = Settings.builder()
+            .put(node().settings())
+            .put("location", OpenSearchIntegTestCase.randomRepoPath(node().settings()))
+            .put(REMOTE_STORE_INDEX_SHALLOW_COPY.getKey(), true)
+            .put(SHALLOW_SNAPSHOT_V2.getKey(), true)
+            .build();
+
+        // Create another repo with shallow snapshot V2 enabled, this should fail.
+        assertThrows(RepositoryException.class, () -> createRepository(client, "test-repo-2", snapshotRepoSettings2));
+
+        // Disable shallow snapshot V2 setting on test-repo-1
+        updateRepository(
+            client,
+            "test-repo-1",
+            Settings.builder().put(snapshotRepoSettings1).put(SHALLOW_SNAPSHOT_V2.getKey(), false).build()
+        );
+        getRepositoriesResponse = client.admin().cluster().prepareGetRepositories("test-repo-1").get();
+        assertFalse(SHALLOW_SNAPSHOT_V2.get(getRepositoriesResponse.repositories().get(0).settings()));
+
+        // Create test-repo-2 with shallow snapshot V2 enabled, this should pass now.
+        createRepository(client, "test-repo-2", snapshotRepoSettings2);
+        getRepositoriesResponse = client.admin().cluster().prepareGetRepositories("test-repo-2").get();
+        assertTrue(SHALLOW_SNAPSHOT_V2.get(getRepositoriesResponse.repositories().get(0).settings()));
+
+        final String indexName = "test-idx";
+        createIndex(indexName);
+        ensureGreen();
+        indexDocuments(client, indexName);
+
+        // Create pinned timestamp snapshot in test-repo-2
+        SnapshotInfo snapshotInfo = createSnapshot("test-repo-2", "test-snap-1", new ArrayList<>());
+        assertNotNull(snapshotInfo.snapshotId());
+
+        // As snapshot is present, even after disabling shallow snapshot setting in test-repo-2, we will not be able to
+        // enable shallow snapshot v2 setting in test-repo-1
+        updateRepository(
+            client,
+            "test-repo-2",
+            Settings.builder().put(snapshotRepoSettings2).put(SHALLOW_SNAPSHOT_V2.getKey(), false).build()
+        );
+        getRepositoriesResponse = client.admin().cluster().prepareGetRepositories("test-repo-2").get();
+        assertFalse(SHALLOW_SNAPSHOT_V2.get(getRepositoriesResponse.repositories().get(0).settings()));
+
+        assertThrows(RepositoryException.class, () -> updateRepository(client, "test-repo-1", snapshotRepoSettings1));
+
+        // After deleting the snapshot, we will be able to enable shallow snapshot v2 setting in test-repo-1
+        AcknowledgedResponse deleteSnapshotResponse = client().admin().cluster().prepareDeleteSnapshot("test-repo-2", "test-snap-1").get();
+
+        assertAcked(deleteSnapshotResponse);
+
+        updateRepository(client, "test-repo-1", snapshotRepoSettings1);
+        getRepositoriesResponse = client.admin().cluster().prepareGetRepositories("test-repo-1").get();
+        assertTrue(SHALLOW_SNAPSHOT_V2.get(getRepositoriesResponse.repositories().get(0).settings()));
+    }
 }