Introduce System Repository

CHANGELOG.md

@@ -51,6 +51,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Replace the deprecated IndexReader APIs with new storedFields() & termVectors() ([#7792](https://github.com/opensearch-project/OpenSearch/pull/7792))
 - [Remote Store] Restrict user override for remote store index level settings ([#8812](https://github.com/opensearch-project/OpenSearch/pull/8812))
 - Removed blocking wait in TransportGetSnapshotsAction which was exhausting generic threadpool ([#8377](https://github.com/opensearch-project/OpenSearch/pull/8377))
+- [Remote Store] Introduce System Repository for Remote Store ([#9088](https://github.com/opensearch-project/OpenSearch/pull/9088/))
 
 ### Deprecated
 

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3Repository.java

@@ -62,7 +62,10 @@
 import org.opensearch.snapshots.SnapshotInfo;
 import org.opensearch.threadpool.Scheduler;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -388,6 +391,15 @@ protected ByteSizeValue chunkSize() {
         return chunkSize;
     }
 
+    @Override
+    public List<Setting<?>> restrictedSystemRepositorySettings() {
+        List<Setting<?>> settings = new ArrayList<>();
+        List<Setting<?>> restrictedSystemRepositorySettings = super.restrictedSystemRepositorySettings();
+        settings.addAll(restrictedSystemRepositorySettings);
+        settings.addAll(new ArrayList<>(List.of(BUCKET_SETTING, BASE_PATH_SETTING)));
+        return Collections.unmodifiableList(settings);
+    }
+
     @Override
     protected void doClose() {
         final Scheduler.Cancellable cancellable = finalizationFuture.getAndSet(null);

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/S3RepositoryTests.java

@@ -49,6 +49,7 @@
 import java.util.Map;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
@@ -132,6 +133,22 @@ public void testDefaultBufferSize() {
         }
     }
 
+    public void testCreateS3SystemRepository() {
+        final RepositoryMetadata metadata = new RepositoryMetadata(
+            "dummy-repo",
+            "mock",
+            Settings.builder()
+                .put(S3Repository.BASE_PATH_SETTING.getKey(), "foo/bar")
+                .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
+                .put(S3Repository.SYSTEM_REPOSITORY_SETTING.getKey(), true)
+                .build()
+        );
+        try (S3Repository s3repo = createS3Repo(metadata)) {
+            assertTrue(s3repo.isSystemRepository());
+            assertThat(s3repo.restrictedSystemRepositorySettings(), hasSize(4));
+        }
+    }
+
     private S3Repository createS3Repo(RepositoryMetadata metadata) {
         return new S3Repository(
             metadata,

server/src/internalClusterTest/java/org/opensearch/repositories/RepositoriesServiceIT.java

@@ -35,6 +35,7 @@
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.fs.FsRepository;
@@ -44,6 +45,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
@@ -108,4 +110,67 @@ public void testUpdateRepository() {
         final Repository updatedRepository = repositoriesService.repository(repositoryName);
         assertThat(updatedRepository, updated ? not(sameInstance(originalRepository)) : sameInstance(originalRepository));
     }
+
+    public void testExternalSystemRepositoryCreationFails() {
+        final String repositoryName = "test-system-repo";
+        final Client client = client();
+
+        final Settings.Builder repoSettings = Settings.builder().put("location", randomRepoPath()).put("system_repository", true);
+
+        assertThrows(
+            "[" + repositoryName + "] cannot register a system repository externally",
+            RepositoryException.class,
+            () -> client.admin().cluster().preparePutRepository(repositoryName).setType(FsRepository.TYPE).setSettings(repoSettings).get()
+        );
+
+        assertThrows(RepositoryException.class, () -> client.admin().cluster().prepareGetRepositories(repositoryName).get());
+    }
+
+    public void testSystemRepositorySettingsUpdationFails() {
+        final InternalTestCluster cluster = internalCluster();
+        final String repositoryName = "test-repo";
+        final Client client = client();
+
+        final RepositoriesService repositoriesService = cluster.getDataOrClusterManagerNodeInstances(RepositoriesService.class)
+            .iterator()
+            .next();
+        final Settings.Builder repoSettings = Settings.builder().put("location", randomRepoPath());
+
+        assertAcked(
+            client.admin().cluster().preparePutRepository(repositoryName).setType(FsRepository.TYPE).setSettings(repoSettings).get()
+        );
+
+        final Settings.Builder updatedRepoSettings = Settings.builder().put("location", randomRepoPath()).put("system_repository", true);
+
+        assertThrows(
+            "[" + repositoryName + "] trying to modify system repository attribute for a repository",
+            RepositoryException.class,
+            () -> client.admin()
+                .cluster()
+                .preparePutRepository(repositoryName)
+                .setType(FsRepository.TYPE)
+                .setSettings(updatedRepoSettings)
+                .get()
+        );
+        assertThrows(RepositoryException.class, () -> repositoriesService.getSystemRepository(repositoryName));
+        assertThat(repositoriesService.repository(repositoryName), instanceOf(FsRepository.class));
+    }
+
+    public void testRestrictedSystemSettings() {
+        final InternalTestCluster cluster = internalCluster();
+        final String repositoryName = "test-repo";
+        final Client client = client();
+
+        final RepositoriesService repositoriesService = cluster.getDataOrClusterManagerNodeInstances(RepositoriesService.class)
+            .iterator()
+            .next();
+        final Settings.Builder repoSettings = Settings.builder().put("location", randomRepoPath());
+
+        assertAcked(
+            client.admin().cluster().preparePutRepository(repositoryName).setType(FsRepository.TYPE).setSettings(repoSettings).get()
+        );
+        List<Setting<?>> restrictedSettings = repositoriesService.repository(repositoryName).restrictedSystemRepositorySettings();
+        assertThat(restrictedSettings, hasSize(2));
+
+    }
 }

server/src/main/java/org/opensearch/action/admin/cluster/repositories/put/TransportPutRepositoryAction.java

@@ -100,7 +100,7 @@ protected void clusterManagerOperation(
         ClusterState state,
         final ActionListener<AcknowledgedResponse> listener
     ) {
-        repositoriesService.registerRepository(
+        repositoriesService.registerOrUpdateRepository(
             request,
             ActionListener.delegateFailure(
                 listener,

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

@@ -52,6 +52,7 @@ public Directory newDirectory(IndexSettings indexSettings, ShardPath path) throw
     }
 
     public Directory newDirectory(String repositoryName, String indexUUID, String shardId) throws IOException {
+        // TODO switch to using {@RepositoriesService#getSystemRespository} once repository auto-bootstrap changes are in
         try (Repository repository = repositoriesService.get().repository(repositoryName)) {
             assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
             BlobPath commonBlobPath = ((BlobStoreRepository) repository).basePath();

server/src/main/java/org/opensearch/index/translog/RemoteBlobStoreInternalTranslogFactory.java

@@ -38,6 +38,7 @@ public RemoteBlobStoreInternalTranslogFactory(
     ) {
         Repository repository;
         try {
+            // TODO switch to using {@RepositoriesService#getSystemRespository} once repository auto-bootstrap changes are in
             repository = repositoriesServiceSupplier.get().repository(repositoryName);
         } catch (RepositoryMissingException ex) {
             throw new IllegalArgumentException("Repository should be created before creating index with remote_store enabled setting", ex);

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

@@ -157,6 +157,11 @@
         return in.isReadOnly();
     }
 
+    @Override
+    public boolean isSystemRepository() {
+        return in.isSystemRepository();
+    }
+
     @Override
     public void snapshotShard(
         Store store,

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

@@ -82,6 +82,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.SYSTEM_REPOSITORY_SETTING;
 
 /**
  * Service responsible for maintaining and providing access to snapshot repositories on nodes.
@@ -159,7 +160,7 @@
      * @param request  register repository request
      * @param listener register repository listener
      */
-    public void registerRepository(final PutRepositoryRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
+    public void registerOrUpdateRepository(final PutRepositoryRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
         assert lifecycle.started() : "Trying to register new repository but service is in state [" + lifecycle.state() + "]";
 
         final RepositoryMetadata newRepositoryMetadata = new RepositoryMetadata(request.name(), request.type(), request.settings());
@@ -210,30 +211,41 @@
                     RepositoriesMetadata repositories = metadata.custom(RepositoriesMetadata.TYPE);
                     if (repositories == null) {
                         logger.info("put repository [{}]", request.name());
+                        ensureNonSystemRepository(request.settings(), request.name());
                         repositories = new RepositoriesMetadata(
                             Collections.singletonList(new RepositoryMetadata(request.name(), request.type(), request.settings()))
                         );
                     } else {
                         boolean found = false;
                         List<RepositoryMetadata> repositoriesMetadata = new ArrayList<>(repositories.repositories().size() + 1);
-
+                        RepositoryMetadata currentRepositoryMetadata = null;
                         for (RepositoryMetadata repositoryMetadata : repositories.repositories()) {
                             if (repositoryMetadata.name().equals(newRepositoryMetadata.name())) {
                                 if (newRepositoryMetadata.equalsIgnoreGenerations(repositoryMetadata)) {
                                     // Previous version is the same as this one no update is needed.
                                     return currentState;
+                                } else if (SYSTEM_REPOSITORY_SETTING.get(repositoryMetadata.settings()) != SYSTEM_REPOSITORY_SETTING.get(
+                                    newRepositoryMetadata.settings()
+                                )) {
+                                    throw new RepositoryException(
+                                        repositoryMetadata.name(),
+                                        "trying to modify system repository attribute for a repository"
+                                    );
                                 }
                                 found = true;
+                                currentRepositoryMetadata = repositoryMetadata;
                                 repositoriesMetadata.add(newRepositoryMetadata);
                             } else {
                                 repositoriesMetadata.add(repositoryMetadata);
                             }
                         }
                         if (!found) {
                             logger.info("put repository [{}]", request.name());
+                            ensureNonSystemRepository(newRepositoryMetadata.settings(), newRepositoryMetadata.name());
                             repositoriesMetadata.add(new RepositoryMetadata(request.name(), request.type(), request.settings()));
                         } else {
                             logger.info("update repository [{}]", request.name());
+                            validateRestrictedSystemRepositorySettingsAndType(currentRepositoryMetadata, newRepositoryMetadata);
                         }
                         repositories = new RepositoriesMetadata(repositoriesMetadata);
                     }
@@ -261,6 +273,29 @@
         );
     }
 
+    private void validateRestrictedSystemRepositorySettingsAndType(
+        RepositoryMetadata currentRepositoryMetadata,
+        RepositoryMetadata newRepositoryMetadata
+    ) {
+        if (SYSTEM_REPOSITORY_SETTING.get(currentRepositoryMetadata.settings())) {
+            if (newRepositoryMetadata.type() != currentRepositoryMetadata.type()) {
+                throw new RepositoryException(
+                    currentRepositoryMetadata.name(),
+                    "trying to modify system repository type " + currentRepositoryMetadata.type()
+                );
+            }
+            Repository repository = repositories.get(currentRepositoryMetadata.name());
+            for (Setting<?> setting : repository.restrictedSystemRepositorySettings()) {
+                if (newRepositoryMetadata.settings().get(setting.getKey()) != null) {
+                    throw new RepositoryException(
+                        repository.getMetadata().name(),
+                        "trying to modify restricted system repository settings " + setting.getKey()
+                    );
+                }
+            }
+        }
+    }
+
     /**
      * Unregisters repository in the cluster
      * <p>
@@ -288,6 +323,7 @@
                         boolean changed = false;
                         for (RepositoryMetadata repositoryMetadata : repositories.repositories()) {
                             if (Regex.simpleMatch(request.name(), repositoryMetadata.name())) {
+                                ensureNonSystemRepository(repositoryMetadata.settings(), request.name());
                                 ensureRepositoryNotInUse(currentState, repositoryMetadata.name());
                                 logger.info("delete repository [{}]", repositoryMetadata.name());
                                 changed = true;
@@ -499,6 +535,21 @@
         throw new RepositoryMissingException(repositoryName);
     }
 
+    /**
+     * Returns registered system repository
+     * <p>*
+     * @param repositoryName repository name
+     * @return registered system repository
+     * @throws RepositoryMissingException if repository with such name isn't registered
+     */
+    public Repository getSystemRepository(String repositoryName) {
+        Repository repository = repositories.get(repositoryName);
+        if (repository != null && SYSTEM_REPOSITORY_SETTING.get(repository.getMetadata().settings())) {
+            return repository;
+        }
+        throw new RepositoryMissingException(repositoryName);
+    }
+
     public List<RepositoryStatsSnapshot> repositoriesStats() {
         List<RepositoryStatsSnapshot> archivedRepoStats = repositoriesStatsArchive.getArchivedStats();
         List<RepositoryStatsSnapshot> activeRepoStats = getRepositoryStatsForActiveRepositories();
@@ -610,6 +661,12 @@
         }
     }
 
+    public static void ensureNonSystemRepository(Settings repoSettings, String repository) {
+        if (SYSTEM_REPOSITORY_SETTING.get(repoSettings)) {
+            throw new RepositoryException(repository, "cannot register a system repository externally");
+        }
+    }
+
     public static void validateRepositoryMetadataSettings(
         ClusterService clusterService,
         final String repositoryName,

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

@@ -43,6 +43,7 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.lifecycle.LifecycleComponent;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.snapshots.IndexShardSnapshotStatus;
@@ -55,6 +56,8 @@
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -237,6 +240,13 @@
      */
     boolean isReadOnly();
 
+    /**
+     * Returns true if the repository is managed by the system directly and doesn't allow managing the lifetime of the
+     * repository through external APIs
+     * @return true if the repository is system managed
+     */
+    boolean isSystemRepository();
+
     /**
      * Creates a snapshot of the shard based on the index commit point.
      * <p>
@@ -341,6 +351,14 @@
         throw new UnsupportedOperationException();
     }
 
+    /**
+     * Returns the list of restricted system repository settings that cannot be mutated once repository is created*
+     * @return the list of settings
+     */
+    default List<Setting<?>> restrictedSystemRepositorySettings() {
+        return Collections.emptyList();
+    }
+
     /**
      * Retrieve shard snapshot status for the stored snapshot
      *