Make RepositoriesService project-aware (#129821)

This PR makes RepositoriesService project aware so that the basic Put,
Get, Delete and Verify repository actions are now project scoped. 

It intentionally leaves the following aspects out of scope for the
current changes: * Repository stats reporting * Repository clean-up,
analysis and integrity verification * Repository usages for searchable
snapshots and CCR

They will be worked on separately. One main reason for leaving them out
is that they are not needed by OBS which is currently blocked by
repository/snapshot changes. They may also have their own complexities,
e.g. stats reporting.

Resolves: ES-10478

Author: ywangd

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

@@ -11,6 +11,7 @@
 
 import org.elasticsearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
@@ -65,7 +66,7 @@ public void testUpdateRepository() {
 
         assertThat(originalRepositoryMetadata.type(), equalTo(FsRepository.TYPE));
 
-        final Repository originalRepository = repositoriesService.repository(repositoryName);
+        final Repository originalRepository = repositoriesService.repository(ProjectId.DEFAULT, repositoryName);
         assertThat(originalRepository, instanceOf(FsRepository.class));
 
         final boolean updated = randomBoolean();
@@ -89,7 +90,7 @@ public void testUpdateRepository() {
 
         assertThat(updatedRepositoryMetadata.type(), equalTo(updatedRepositoryType));
 
-        final Repository updatedRepository = repositoriesService.repository(repositoryName);
+        final Repository updatedRepository = repositoriesService.repository(ProjectId.DEFAULT, repositoryName);
         assertThat(updatedRepository, updated ? not(sameInstance(originalRepository)) : sameInstance(originalRepository));
 
         // check that a noop update does not verify. Since the new data node does not share the same `path.repo`, verification will fail if

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/delete/TransportDeleteRepositoryAction.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -36,14 +37,16 @@ public class TransportDeleteRepositoryAction extends AcknowledgedTransportMaster
 
     public static final ActionType<AcknowledgedResponse> TYPE = new ActionType<>("cluster:admin/repository/delete");
     private final RepositoriesService repositoriesService;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public TransportDeleteRepositoryAction(
         TransportService transportService,
         ClusterService clusterService,
         RepositoriesService repositoriesService,
         ThreadPool threadPool,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver
     ) {
         super(
             TYPE.name(),
@@ -55,11 +58,12 @@ public TransportDeleteRepositoryAction(
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
         this.repositoriesService = repositoriesService;
+        this.projectResolver = projectResolver;
     }
 
     @Override
     protected ClusterBlockException checkBlock(DeleteRepositoryRequest request, ClusterState state) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+        return state.blocks().globalBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_WRITE);
     }
 
     @Override
@@ -69,7 +73,7 @@ protected void masterOperation(
         ClusterState state,
         final ActionListener<AcknowledgedResponse> listener
     ) {
-        repositoriesService.unregisterRepository(request, listener);
+        repositoriesService.unregisterRepository(projectResolver.getProjectId(), request, listener);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/get/TransportGetRepositoriesAction.java

@@ -11,11 +11,12 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
-import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.action.support.master.TransportMasterNodeReadProjectAction;
+import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -28,14 +29,15 @@
 /**
  * Transport action for get repositories operation
  */
-public class TransportGetRepositoriesAction extends TransportMasterNodeReadAction<GetRepositoriesRequest, GetRepositoriesResponse> {
+public class TransportGetRepositoriesAction extends TransportMasterNodeReadProjectAction<GetRepositoriesRequest, GetRepositoriesResponse> {
 
     @Inject
     public TransportGetRepositoriesAction(
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver
     ) {
         super(
             GetRepositoriesAction.NAME,
@@ -44,24 +46,25 @@ public TransportGetRepositoriesAction(
             threadPool,
             actionFilters,
             GetRepositoriesRequest::new,
+            projectResolver,
             GetRepositoriesResponse::new,
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
     }
 
     @Override
-    protected ClusterBlockException checkBlock(GetRepositoriesRequest request, ClusterState state) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+    protected ClusterBlockException checkBlock(GetRepositoriesRequest request, ProjectState state) {
+        return state.blocks().globalBlockedException(state.projectId(), ClusterBlockLevel.METADATA_READ);
     }
 
     @Override
     protected void masterOperation(
         Task task,
         final GetRepositoriesRequest request,
-        ClusterState state,
+        ProjectState state,
         final ActionListener<GetRepositoriesResponse> listener
     ) {
-        final var result = ResolvedRepositories.resolve(state, request.repositories());
+        final var result = ResolvedRepositories.resolve(state.metadata(), request.repositories());
         if (result.hasMissingRepositories()) {
             listener.onFailure(new RepositoryMissingException(String.join(", ", result.missing())));
         } else {

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

@@ -18,6 +18,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -36,14 +37,16 @@ public class TransportPutRepositoryAction extends AcknowledgedTransportMasterNod
 
     public static final ActionType<AcknowledgedResponse> TYPE = new ActionType<>("cluster:admin/repository/put");
     private final RepositoriesService repositoriesService;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public TransportPutRepositoryAction(
         TransportService transportService,
         ClusterService clusterService,
         RepositoriesService repositoriesService,
         ThreadPool threadPool,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver
     ) {
         super(
             TYPE.name(),
@@ -55,11 +58,12 @@ public TransportPutRepositoryAction(
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
         this.repositoriesService = repositoriesService;
+        this.projectResolver = projectResolver;
     }
 
     @Override
     protected ClusterBlockException checkBlock(PutRepositoryRequest request, ClusterState state) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+        return state.blocks().globalBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_WRITE);
     }
 
     @Override
@@ -69,7 +73,7 @@ protected void masterOperation(
         ClusterState state,
         final ActionListener<AcknowledgedResponse> listener
     ) {
-        repositoriesService.registerRepository(request, listener);
+        repositoriesService.registerRepository(projectResolver.getProjectId(), request, listener);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/reservedstate/ReservedRepositoryAction.java

@@ -11,6 +11,8 @@
 
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.reservedstate.ReservedClusterStateHandler;
 import org.elasticsearch.reservedstate.TransformState;
@@ -60,7 +62,9 @@ public Collection<PutRepositoryRequest> prepare(Object input) {
         for (var repositoryRequest : repositories) {
             validate(repositoryRequest);
             RepositoriesService.validateRepositoryName(repositoryRequest.name());
-            repositoriesService.validateRepositoryCanBeCreated(repositoryRequest);
+            @FixForMultiProject(description = "resolve the actual projectId, ES-10479")
+            final var projectId = ProjectId.DEFAULT;
+            repositoriesService.validateRepositoryCanBeCreated(projectId, repositoryRequest);
         }
 
         return repositories;
@@ -72,8 +76,14 @@ public TransformState transform(List<PutRepositoryRequest> source, TransformStat
 
         ClusterState state = prevState.state();
 
+        @FixForMultiProject(description = "resolve the actual projectId, ES-10479")
+        final var projectId = ProjectId.DEFAULT;
         for (var request : requests) {
-            RepositoriesService.RegisterRepositoryTask task = new RepositoriesService.RegisterRepositoryTask(repositoriesService, request);
+            RepositoriesService.RegisterRepositoryTask task = new RepositoriesService.RegisterRepositoryTask(
+                repositoriesService,
+                projectId,
+                request
+            );
             state = task.execute(state);
         }
 
@@ -83,7 +93,11 @@ public TransformState transform(List<PutRepositoryRequest> source, TransformStat
         toDelete.removeAll(entities);
 
         for (var repositoryToDelete : toDelete) {
-            var task = new RepositoriesService.UnregisterRepositoryTask(RESERVED_CLUSTER_STATE_HANDLER_IGNORED_TIMEOUT, repositoryToDelete);
+            var task = new RepositoriesService.UnregisterRepositoryTask(
+                RESERVED_CLUSTER_STATE_HANDLER_IGNORED_TIMEOUT,
+                projectId,
+                repositoryToDelete
+            );
             state = task.execute(state);
         }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/verify/TransportVerifyRepositoryAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -30,14 +31,16 @@
 public class TransportVerifyRepositoryAction extends TransportMasterNodeAction<VerifyRepositoryRequest, VerifyRepositoryResponse> {
 
     private final RepositoriesService repositoriesService;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public TransportVerifyRepositoryAction(
         TransportService transportService,
         ClusterService clusterService,
         RepositoriesService repositoriesService,
         ThreadPool threadPool,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver
     ) {
         super(
             VerifyRepositoryAction.NAME,
@@ -50,11 +53,12 @@ public TransportVerifyRepositoryAction(
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
         this.repositoriesService = repositoriesService;
+        this.projectResolver = projectResolver;
     }
 
     @Override
     protected ClusterBlockException checkBlock(VerifyRepositoryRequest request, ClusterState state) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+        return state.blocks().globalBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_READ);
     }
 
     @Override
@@ -65,6 +69,7 @@ protected void masterOperation(
         final ActionListener<VerifyRepositoryResponse> listener
     ) {
         repositoriesService.verifyRepository(
+            projectResolver.getProjectId(),
             request.name(),
             listener.map(verifyResponse -> new VerifyRepositoryResponse(verifyResponse.toArray(new DiscoveryNode[0])))
         );

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java

@@ -144,7 +144,7 @@ protected void masterOperation(
     ) {
         assert task instanceof CancellableTask : task + " not cancellable";
 
-        final var resolvedRepositories = ResolvedRepositories.resolve(state, request.repositories());
+        final var resolvedRepositories = ResolvedRepositories.resolve(state.metadata().getProject(), request.repositories());
         if (resolvedRepositories.hasMissingRepositories()) {
             throw new RepositoryMissingException(String.join(", ", resolvedRepositories.missing()));
         }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java

@@ -24,11 +24,13 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
 import org.elasticsearch.injection.guice.Inject;
@@ -252,7 +254,7 @@ void buildResponse(
                             // BWC behavior, load the stats directly from the repository.
                             shardStatus = new SnapshotIndexShardStatus(
                                 shardId,
-                                repositoriesService.repository(entry.repository())
+                                repositoriesService.repository(entry.projectId(), entry.repository())
                                     .getShardSnapshotStatus(
                                         entry.snapshot().getSnapshotId(),
                                         entry.indices().get(shardId.getIndexName()),
@@ -297,7 +299,9 @@ private void loadRepositoryData(
     ) {
         final Set<String> requestedSnapshotNames = Sets.newHashSet(request.snapshots());
         final ListenableFuture<RepositoryData> repositoryDataListener = new ListenableFuture<>();
-        repositoriesService.getRepositoryData(repositoryName, repositoryDataListener);
+        @FixForMultiProject(description = "resolve the actual projectId, ES-10166")
+        final var projectId = ProjectId.DEFAULT;
+        repositoriesService.getRepositoryData(projectId, repositoryName, repositoryDataListener);
         final Collection<SnapshotId> snapshotIdsToLoad = new ArrayList<>();
         repositoryDataListener.addListener(listener.delegateFailureAndWrap((delegate, repositoryData) -> {
             task.ensureNotCancelled();

server/src/main/java/org/elasticsearch/health/node/tracker/RepositoriesHealthTracker.java

@@ -42,7 +42,7 @@ protected RepositoriesHealthInfo determineCurrentHealth() {
 
         var unknown = new ArrayList<String>();
         var invalid = new ArrayList<String>();
-        repositories.values().forEach(repository -> {
+        repositories.forEach(repository -> {
             if (repository instanceof UnknownTypeRepository) {
                 unknown.add(repository.getMetadata().name());
             } else if (repository instanceof InvalidRepository) {

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RecoverySource;
 import org.elasticsearch.cluster.routing.RecoverySource.SnapshotRecoverySource;
@@ -154,6 +155,7 @@
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.internal.FieldUsageTrackingDirectoryReader;
 import org.elasticsearch.search.suggest.completion.CompletionStats;
+import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.Transports;
 
@@ -3527,12 +3529,14 @@ public void startRecovery(
                 }
             }
             case SNAPSHOT -> {
-                final String repo = ((SnapshotRecoverySource) recoveryState.getRecoverySource()).snapshot().getRepository();
+                final Snapshot snapshot = ((SnapshotRecoverySource) recoveryState.getRecoverySource()).snapshot();
+                final ProjectId projectId = snapshot.getProjectId();
+                final String repo = snapshot.getRepository();
                 executeRecovery(
                     "from snapshot",
                     recoveryState,
                     recoveryListener,
-                    l -> restoreFromRepository(repositoriesService.repository(repo), l)
+                    l -> restoreFromRepository(repositoriesService.repository(projectId, repo), l)
                 );
             }
             case LOCAL_SHARDS -> {