Make restore support multi-project (#131661)

This PR makes the restore process project aware and unmute relevant
tests. The later requires TransportRecoveryAction to be project aware
which is done in this PR as well.

Relates: #130000 Resolves: ES-10228

Author: ywangd

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamsSnapshotsIT.java

@@ -47,6 +47,7 @@
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
 import org.elasticsearch.snapshots.RestoreInfo;
+import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInProgressException;
 import org.elasticsearch.snapshots.SnapshotInfo;
@@ -1394,7 +1395,7 @@ public void testWarningHeaderOnRestoreWithoutTemplates() throws Exception {
             .get();
 
         RestStatus status = createSnapshotResponse.getSnapshotInfo().status();
-        SnapshotId snapshotId = createSnapshotResponse.getSnapshotInfo().snapshotId();
+        Snapshot snapshot = createSnapshotResponse.getSnapshotInfo().snapshot();
         assertEquals(RestStatus.OK, status);
 
         assertEquals(Collections.singletonList(dsBackingIndexName), getSnapshot(REPO, SNAPSHOT).indices());
@@ -1423,7 +1424,7 @@ public void testWarningHeaderOnRestoreWithoutTemplates() throws Exception {
             client,
             request,
             "Snapshot ["
-                + snapshotId
+                + snapshot
                 + "] contains data stream ["
                 + datastreamName
                 + "] but custer does not have a matching index "

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/restore/TransportRestoreSnapshotAction.java

@@ -16,6 +16,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.injection.guice.Inject;
 import org.elasticsearch.snapshots.RestoreService;
@@ -29,14 +30,16 @@
 public class TransportRestoreSnapshotAction extends TransportMasterNodeAction<RestoreSnapshotRequest, RestoreSnapshotResponse> {
     public static final ActionType<RestoreSnapshotResponse> TYPE = new ActionType<>("cluster:admin/snapshot/restore");
     private final RestoreService restoreService;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public TransportRestoreSnapshotAction(
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
         RestoreService restoreService,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        ProjectResolver projectResolver
     ) {
         super(
             TYPE.name(),
@@ -49,13 +52,15 @@ public TransportRestoreSnapshotAction(
             threadPool.executor(ThreadPool.Names.SNAPSHOT_META)
         );
         this.restoreService = restoreService;
+        this.projectResolver = projectResolver;
     }
 
     @Override
     protected ClusterBlockException checkBlock(RestoreSnapshotRequest request, ClusterState state) {
         // Restoring a snapshot might change the global state and create/change an index,
         // so we need to check for METADATA_WRITE and WRITE blocks
-        ClusterBlockException blockException = state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+        ClusterBlockException blockException = state.blocks()
+            .globalBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_WRITE);
         if (blockException != null) {
             return blockException;
         }
@@ -70,17 +75,21 @@ protected void masterOperation(
         final ClusterState state,
         final ActionListener<RestoreSnapshotResponse> listener
     ) {
-        restoreService.restoreSnapshot(request, listener.delegateFailure((delegatedListener, restoreCompletionResponse) -> {
-            if (restoreCompletionResponse.restoreInfo() == null && request.waitForCompletion()) {
-                RestoreClusterStateListener.createAndRegisterListener(
-                    clusterService,
-                    restoreCompletionResponse,
-                    delegatedListener,
-                    threadPool.getThreadContext()
-                );
-            } else {
-                delegatedListener.onResponse(new RestoreSnapshotResponse(restoreCompletionResponse.restoreInfo()));
-            }
-        }));
+        restoreService.restoreSnapshot(
+            projectResolver.getProjectId(),
+            request,
+            listener.delegateFailure((delegatedListener, restoreCompletionResponse) -> {
+                if (restoreCompletionResponse.restoreInfo() == null && request.waitForCompletion()) {
+                    RestoreClusterStateListener.createAndRegisterListener(
+                        clusterService,
+                        restoreCompletionResponse,
+                        delegatedListener,
+                        threadPool.getThreadContext()
+                    );
+                } else {
+                    delegatedListener.onResponse(new RestoreSnapshotResponse(restoreCompletionResponse.restoreInfo()));
+                }
+            })
+        );
     }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/recovery/TransportRecoveryAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardsIterator;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -43,14 +44,16 @@
 public class TransportRecoveryAction extends TransportBroadcastByNodeAction<RecoveryRequest, RecoveryResponse, RecoveryState> {
 
     private final IndicesService indicesService;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public TransportRecoveryAction(
         ClusterService clusterService,
         TransportService transportService,
         IndicesService indicesService,
         ActionFilters actionFilters,
-        IndexNameExpressionResolver indexNameExpressionResolver
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        ProjectResolver projectResolver
     ) {
         super(
             RecoveryAction.NAME,
@@ -62,6 +65,7 @@ public TransportRecoveryAction(
             transportService.getThreadPool().executor(ThreadPool.Names.MANAGEMENT)
         );
         this.indicesService = indicesService;
+        this.projectResolver = projectResolver;
     }
 
     @Override
@@ -110,16 +114,16 @@ protected void shardOperation(RecoveryRequest request, ShardRouting shardRouting
 
     @Override
     protected ShardsIterator shards(ClusterState state, RecoveryRequest request, String[] concreteIndices) {
-        return state.routingTable().allShardsIncludingRelocationTargets(concreteIndices);
+        return state.routingTable(projectResolver.getProjectId()).allShardsIncludingRelocationTargets(concreteIndices);
     }
 
     @Override
     protected ClusterBlockException checkGlobalBlock(ClusterState state, RecoveryRequest request) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+        return state.blocks().globalBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_READ);
     }
 
     @Override
     protected ClusterBlockException checkRequestBlock(ClusterState state, RecoveryRequest request, String[] concreteIndices) {
-        return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_READ, concreteIndices);
+        return state.blocks().indicesBlockedException(projectResolver.getProjectId(), ClusterBlockLevel.METADATA_READ, concreteIndices);
     }
 }

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -1159,7 +1159,8 @@ public Map<String, String> queryFields() {
             systemIndices,
             indicesService,
             fileSettingsService,
-            threadPool
+            threadPool,
+            projectResolver.supportsMultipleProjects()
         );
 
         DiscoveryModule discoveryModule = createDiscoveryModule(

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -134,6 +134,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.LeakTracker;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
@@ -377,7 +378,10 @@ public static String getRepositoryDataBlobName(long repositoryGeneration) {
         METADATA_NAME_FORMAT,
         (repoName, parser) -> ProjectMetadata.Builder.fromXContent(parser),
         projectMetadata -> ChunkedToXContent.wrapAsToXContent(
-            params -> Iterators.concat(Iterators.single((builder, ignored) -> builder.field("id", projectMetadata.id())))
+            params -> Iterators.concat(
+                Iterators.single((ToXContent) (builder, p) -> builder.field("id", projectMetadata.id())),
+                projectMetadata.toXContentChunked(params)
+            )
         )
     );
 

server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -126,8 +127,9 @@ public Path watchedFile() {
      * file based settings from the cluster state.
      * @param clusterState the cluster state before snapshot restore
      * @param mdBuilder the current metadata builder for the new cluster state
+     * @param projectId the project associated with the restore
      */
-    public void handleSnapshotRestore(ClusterState clusterState, Metadata.Builder mdBuilder) {
+    public void handleSnapshotRestore(ClusterState clusterState, Metadata.Builder mdBuilder, ProjectId projectId) {
         assert clusterState.nodes().isLocalNodeElectedMaster();
 
         ReservedStateMetadata fileSettingsMetadata = clusterState.metadata().reservedStateMetadata().get(NAMESPACE);

server/src/main/java/org/elasticsearch/snapshots/InternalSnapshotsInfoService.java

@@ -211,7 +211,10 @@ private class FetchingSnapshotShardSizeRunnable extends AbstractRunnable {
 
         @Override
         protected void doRun() throws Exception {
-            final Repository repository = repositoriesService.repository(snapshotShard.snapshot.getRepository());
+            final Repository repository = repositoriesService.repository(
+                snapshotShard.snapshot().getProjectId(),
+                snapshotShard.snapshot.getRepository()
+            );
 
             logger.debug("fetching snapshot shard size for {}", snapshotShard);
             final long snapshotShardSize = repository.getShardSnapshotStatus(