Introduce projectClient method on Client (#129174)

We originally defined the `projectClient` method on `ProjectResolver` as
a convenience method to execute API calls for specific projects. That
method requires a reference to both a `ProjectResolver` and a `Client`.

We now introduce the same method directly on the `Client` interface and
inject a `ProjectResolver` there, removing the need for a
`ProjectResolver` reference in places that just want to execute API
requests on a specific project.

To reduce the number of changes, this change solely focuses on
introducing the new method. Future changes will migrate the uses of the
original method to the new one and remove the original altogether.

Author: nielsbauman

modules/analysis-common/src/test/java/org/elasticsearch/analysis/common/PredicateTokenScriptFilterTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.support.AbstractClient;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
@@ -96,7 +97,7 @@ public <FactoryType> FactoryType compile(Script script, ScriptContext<FactoryTyp
 
     private static class MockClient extends AbstractClient {
         MockClient(Settings settings, ThreadPool threadPool) {
-            super(settings, threadPool);
+            super(settings, threadPool, TestProjectResolvers.alwaysThrow());
         }
 
         @Override

modules/analysis-common/src/test/java/org/elasticsearch/analysis/common/ScriptedConditionTokenFilterTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.support.AbstractClient;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
@@ -96,7 +97,7 @@ public <FactoryType> FactoryType compile(Script script, ScriptContext<FactoryTyp
 
     private class MockClient extends AbstractClient {
         MockClient(Settings settings, ThreadPool threadPool) {
-            super(settings, threadPool);
+            super(settings, threadPool, TestProjectResolvers.alwaysThrow());
         }
 
         @Override

modules/rank-eval/src/test/java/org/elasticsearch/index/rankeval/TransportRankEvalActionTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
@@ -68,7 +69,7 @@ public void testTransferRequestParameters() throws Exception {
         );
         rankEvalRequest.indicesOptions(expectedIndicesOptions);
 
-        NodeClient client = new NodeClient(settings, null) {
+        NodeClient client = new NodeClient(settings, null, TestProjectResolvers.alwaysThrow()) {
             @Override
             public void multiSearch(MultiSearchRequest request, ActionListener<MultiSearchResponse> listener) {
                 assertEquals(1, request.requests().size());

modules/reindex/src/test/java/org/elasticsearch/reindex/ClientScrollableHitSourceTests.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.client.internal.ParentTaskAssigningClient;
 import org.elasticsearch.client.internal.support.AbstractClient;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
@@ -214,7 +215,7 @@ private static class MockClient extends AbstractClient {
         private ExecuteRequest<?, ?> executeRequest;
 
         MockClient(ThreadPool threadPool) {
-            super(Settings.EMPTY, threadPool);
+            super(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow());
         }
 
         @Override

server/src/main/java/org/elasticsearch/client/internal/Client.java

@@ -52,6 +52,8 @@
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.action.update.UpdateResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.transport.RemoteClusterService;
@@ -399,6 +401,16 @@ public interface Client extends ElasticsearchClient {
      */
     Client filterWithHeader(Map<String, String> headers);
 
+    /**
+     * Returns a client that executes every request in the context of the given project.
+     */
+    Client projectClient(ProjectId projectId);
+
+    /**
+     * Returns this client's project resolver.
+     */
+    ProjectResolver projectResolver();
+
     /**
      * Returns a client to a remote cluster with the given cluster alias.
      *

server/src/main/java/org/elasticsearch/client/internal/FilterClient.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.client.internal.support.AbstractClient;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterService;
@@ -35,15 +36,15 @@ public abstract class FilterClient extends AbstractClient {
      * @see #in()
      */
     public FilterClient(Client in) {
-        this(in.settings(), in.threadPool(), in);
+        this(in.settings(), in.threadPool(), in.projectResolver(), in);
     }
 
     /**
      * A Constructor that allows to pass settings and threadpool separately. This is useful if the
      * client is a proxy and not yet fully constructed ie. both dependencies are not available yet.
      */
-    protected FilterClient(Settings settings, ThreadPool threadPool, Client in) {
-        super(settings, threadPool);
+    protected FilterClient(Settings settings, ThreadPool threadPool, ProjectResolver projectResolver, Client in) {
+        super(settings, threadPool, projectResolver);
         this.in = in;
     }
 

server/src/main/java/org/elasticsearch/client/internal/node/NodeClient.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.client.internal.RemoteClusterClient;
 import org.elasticsearch.client.internal.support.AbstractClient;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskCancelledException;
@@ -48,8 +49,8 @@ public class NodeClient extends AbstractClient {
     private Transport.Connection localConnection;
     private RemoteClusterService remoteClusterService;
 
-    public NodeClient(Settings settings, ThreadPool threadPool) {
-        super(settings, threadPool);
+    public NodeClient(Settings settings, ThreadPool threadPool, ProjectResolver projectResolver) {
+        super(settings, threadPool, projectResolver);
     }
 
     public void initialize(

server/src/main/java/org/elasticsearch/client/internal/support/AbstractClient.java

@@ -76,6 +76,8 @@
 import org.elasticsearch.client.internal.AdminClient;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.FilterClient;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Nullable;
@@ -92,12 +94,14 @@ public abstract class AbstractClient implements Client {
 
     protected final Settings settings;
     private final ThreadPool threadPool;
+    private final ProjectResolver projectResolver;
     private final AdminClient admin;
 
     @SuppressWarnings("this-escape")
-    public AbstractClient(Settings settings, ThreadPool threadPool) {
+    public AbstractClient(Settings settings, ThreadPool threadPool, ProjectResolver projectResolver) {
         this.settings = settings;
         this.threadPool = threadPool;
+        this.projectResolver = projectResolver;
         this.admin = new AdminClient(this);
         this.logger = LogManager.getLogger(this.getClass());
     }
@@ -112,6 +116,11 @@ public final ThreadPool threadPool() {
         return this.threadPool;
     }
 
+    @Override
+    public ProjectResolver projectResolver() {
+        return projectResolver;
+    }
+
     @Override
     public final AdminClient admin() {
         return admin;
@@ -407,6 +416,32 @@ protected <Request extends ActionRequest, Response extends ActionResponse> void
         };
     }
 
+    @Override
+    public Client projectClient(ProjectId projectId) {
+        // We only take the shortcut when the given project ID matches the "current" project ID. If it doesn't, we'll let #executeOnProject
+        // take care of error handling.
+        if (projectResolver.supportsMultipleProjects() == false && projectId.equals(projectResolver.getProjectId())) {
+            return this;
+        }
+        return new FilterClient(this) {
+            @Override
+            protected <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
+                ActionType<Response> action,
+                Request request,
+                ActionListener<Response> listener
+            ) {
+                projectResolver.executeOnProject(projectId, () -> super.doExecute(action, request, listener));
+            }
+
+            @Override
+            public Client projectClient(ProjectId projectId) {
+                throw new IllegalStateException(
+                    "Unable to create a project client for project [" + projectId + "], nested project client creation is not supported"
+                );
+            }
+        };
+    }
+
     /**
      * Same as {@link PlainActionFuture} but for use with {@link RefCounted} result types. Unlike {@code PlainActionFuture} this future
      * acquires a reference to its result. This means that the result reference must be released by a call to {@link RefCounted#decRef()}

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

@@ -287,9 +287,15 @@ static NodeConstruction prepareConstruction(
                 // places they shouldn't. Best to explicitly drop them now to protect against such leakage.
                 settingsModule = constructor.validateSettings(initialEnvironment.settings(), settings, threadPool);
             }
+            // serverless deployments plug-in the multi-project resolver factory
+            ProjectResolver projectResolver = constructor.pluginsService.loadSingletonServiceProvider(
+                ProjectResolverFactory.class,
+                () -> ProjectResolverFactory.DEFAULT
+            ).create();
+            constructor.modules.bindToInstance(ProjectResolver.class, projectResolver);
 
             SearchModule searchModule = constructor.createSearchModule(settingsModule.getSettings(), threadPool, telemetryProvider);
-            constructor.createClientAndRegistries(settingsModule.getSettings(), threadPool, searchModule);
+            constructor.createClientAndRegistries(settingsModule.getSettings(), threadPool, searchModule, projectResolver);
             DocumentParsingProvider documentParsingProvider = constructor.getDocumentParsingProvider();
 
             ScriptService scriptService = constructor.createScriptService(settingsModule, threadPool, serviceProvider);
@@ -305,7 +311,8 @@ static NodeConstruction prepareConstruction(
                 serviceProvider,
                 forbidPrivateIndexSettings,
                 telemetryProvider,
-                documentParsingProvider
+                documentParsingProvider,
+                projectResolver
             );
 
             return constructor;
@@ -562,8 +569,13 @@ private SearchModule createSearchModule(Settings settings, ThreadPool threadPool
     /**
      * Create various objects that are stored as member variables. This is so they are accessible as soon as possible.
      */
-    private void createClientAndRegistries(Settings settings, ThreadPool threadPool, SearchModule searchModule) {
-        client = new NodeClient(settings, threadPool);
+    private void createClientAndRegistries(
+        Settings settings,
+        ThreadPool threadPool,
+        SearchModule searchModule,
+        ProjectResolver projectResolver
+    ) {
+        client = new NodeClient(settings, threadPool, projectResolver);
         modules.add(b -> {
             b.bind(Client.class).toInstance(client);
             b.bind(NodeClient.class).toInstance(client);
@@ -664,7 +676,8 @@ private void construct(
         NodeServiceProvider serviceProvider,
         boolean forbidPrivateIndexSettings,
         TelemetryProvider telemetryProvider,
-        DocumentParsingProvider documentParsingProvider
+        DocumentParsingProvider documentParsingProvider,
+        ProjectResolver projectResolver
     ) throws IOException {
 
         Settings settings = settingsModule.getSettings();
@@ -681,12 +694,6 @@ private void construct(
             telemetryProvider.getTracer()
         );
 
-        // serverless deployments plug-in the multi-project resolver factory
-        ProjectResolver projectResolver = pluginsService.loadSingletonServiceProvider(
-            ProjectResolverFactory.class,
-            () -> ProjectResolverFactory.DEFAULT
-        ).create();
-        modules.bindToInstance(ProjectResolver.class, projectResolver);
         ClusterService clusterService = createClusterService(settingsModule, threadPool, taskManager);
         clusterService.addStateApplier(scriptService);
 

server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportGetTaskActionTests.java

@@ -73,7 +73,7 @@ public void testGetTaskActionWithMultiProjectEnabled() {
         var transportService = mock(TransportService.class);
         var clusterService = mock(ClusterService.class);
         var nodeId = "node1";
-        NodeClient client = new NodeClient(Settings.EMPTY, threadPool) {
+        NodeClient client = new NodeClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow()) {
             @Override
             @SuppressWarnings("unchecked")
             public <Request extends ActionRequest, Response extends ActionResponse> void doExecute(

server/src/test/java/org/elasticsearch/action/admin/cluster/remote/RemoteClusterNodesActionTests.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.cluster.version.CompatibilityVersions;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
@@ -109,7 +110,7 @@ public void testDoExecuteForRemoteServerNodes() {
         final RemoteClusterNodesAction.TransportAction action = new RemoteClusterNodesAction.TransportAction(
             mock(TransportService.class),
             new ActionFilters(Set.of()),
-            new AbstractClient(Settings.EMPTY, threadPool) {
+            new AbstractClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow()) {
                 @SuppressWarnings("unchecked")
                 @Override
                 protected <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
@@ -187,7 +188,7 @@ public void testDoExecuteForRemoteNodes() {
         final RemoteClusterNodesAction.TransportAction action = new RemoteClusterNodesAction.TransportAction(
             mock(TransportService.class),
             new ActionFilters(Set.of()),
-            new AbstractClient(Settings.EMPTY, threadPool) {
+            new AbstractClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow()) {
                 @SuppressWarnings("unchecked")
                 @Override
                 protected <Request extends ActionRequest, Response extends ActionResponse> void doExecute(

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusActionTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.SnapshotsInProgress;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -65,7 +66,7 @@ public void initializeComponents() throws Exception {
             clusterService.getClusterSettings(),
             Set.of()
         );
-        final var nodeClient = new NodeClient(clusterService.getSettings(), threadPool);
+        final var nodeClient = new NodeClient(clusterService.getSettings(), threadPool, TestProjectResolvers.alwaysThrow());
         repositoriesService = new RepositoriesService(
             clusterService.getSettings(),
             clusterService,

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java

@@ -160,7 +160,7 @@ class TestTransportBulkAction extends TransportBulkAction {
                 transportService,
                 TransportBulkActionIngestTests.this.clusterService,
                 ingestService,
-                new NodeClient(Settings.EMPTY, TransportBulkActionIngestTests.this.threadPool),
+                new NodeClient(Settings.EMPTY, TransportBulkActionIngestTests.this.threadPool, TestProjectResolvers.alwaysThrow()),
                 new ActionFilters(Collections.emptySet()),
                 TestIndexNameExpressionResolver.newInstance(),
                 new IndexingPressure(SETTINGS),

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTests.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.cluster.routing.GlobalRoutingTableTestHelper;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -121,7 +122,7 @@ class TestTransportBulkAction extends TransportBulkAction {
                 transportService,
                 TransportBulkActionTests.this.clusterService,
                 null,
-                new NodeClient(Settings.EMPTY, TransportBulkActionTests.this.threadPool),
+                new NodeClient(Settings.EMPTY, TransportBulkActionTests.this.threadPool, TestProjectResolvers.alwaysThrow()),
                 new ActionFilters(Collections.emptySet()),
                 new Resolver(),
                 new IndexingPressure(Settings.EMPTY),

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTookTests.java

@@ -117,7 +117,7 @@ private TransportBulkAction createAction(boolean controlled, AtomicLong expected
         IndexNameExpressionResolver resolver = new Resolver();
         ActionFilters actionFilters = new ActionFilters(new HashSet<>());
 
-        NodeClient client = new NodeClient(Settings.EMPTY, threadPool) {
+        NodeClient client = new NodeClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow()) {
             @Override
             @SuppressWarnings("unchecked")
             public <Request extends ActionRequest, Response extends ActionResponse> void doExecute(

server/src/test/java/org/elasticsearch/action/get/TransportMultiGetActionTests.java

@@ -185,7 +185,7 @@ public static void afterClass() {
 
     public void testTransportMultiGetAction() {
         final Task task = createTask();
-        final NodeClient client = new NodeClient(Settings.EMPTY, threadPool);
+        final NodeClient client = new NodeClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow());
         final MultiGetRequestBuilder request = new MultiGetRequestBuilder(client);
         request.add(new MultiGetRequest.Item("index1", "1"));
         request.add(new MultiGetRequest.Item("index1", "2"));
@@ -219,7 +219,7 @@ protected void executeShardAction(
 
     public void testTransportMultiGetAction_withMissingRouting() {
         final Task task = createTask();
-        final NodeClient client = new NodeClient(Settings.EMPTY, threadPool);
+        final NodeClient client = new NodeClient(Settings.EMPTY, threadPool, TestProjectResolvers.alwaysThrow());
         final MultiGetRequestBuilder request = new MultiGetRequestBuilder(client);
         request.add(new MultiGetRequest.Item("index2", "1").routing("1"));
         request.add(new MultiGetRequest.Item("index2", "2"));

server/src/test/java/org/elasticsearch/action/search/MultiSearchActionTookTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
@@ -137,7 +138,7 @@ private TransportMultiSearchAction createTransportMultiSearchAction(boolean cont
         final Executor commonExecutor = randomExecutor(threadPool);
         final Set<SearchRequest> requests = Collections.newSetFromMap(Collections.synchronizedMap(new IdentityHashMap<>()));
 
-        NodeClient client = new NodeClient(settings, threadPool) {
+        NodeClient client = new NodeClient(settings, threadPool, TestProjectResolvers.alwaysThrow()) {
             @Override
             public void search(final SearchRequest request, final ActionListener<SearchResponse> listener) {
                 requests.add(request);