Add support for project_routing for _search and _async_search

docs/changelog/137566.yaml

@@ -0,0 +1,5 @@
+pr: 137566
+summary: Add support for `project_routing` for `_search` and `_async_search`
+area: CCS
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/search/SearchRequest.java

@@ -116,6 +116,7 @@ public class SearchRequest extends LegacyActionRequest implements IndicesRequest
      * enabling synthetic source natively in the index.
      */
     private boolean forceSyntheticSource = false;
+    private String projectRouting;
 
     public SearchRequest() {
         this.localClusterAlias = null;
@@ -168,6 +169,19 @@ public boolean allowsCrossProject() {
         return true;
     }
 
+    @Override
+    public String getProjectRouting() {
+        return projectRouting;
+    }
+
+    public void setProjectRouting(@Nullable String projectRouting) {
+        if (this.projectRouting != null) {
+            throw new IllegalArgumentException("project_routing is already set to [" + this.projectRouting + "]");
+        }
+
+        this.projectRouting = projectRouting;
+    }
+
     /**
      * Creates a new sub-search request starting from the original search request that is provided.
      * For internal use only, allows to fork a search request into multiple search requests that will be executed independently.
@@ -228,6 +242,7 @@ private SearchRequest(
         this.waitForCheckpoints = searchRequest.waitForCheckpoints;
         this.waitForCheckpointsTimeout = searchRequest.waitForCheckpointsTimeout;
         this.forceSyntheticSource = searchRequest.forceSyntheticSource;
+        this.projectRouting = searchRequest.projectRouting;
     }
 
     /**

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -620,7 +620,8 @@ public void onFailure(Exception e) {
                         }),
                         forceConnectTimeoutSecs,
                         resolvesCrossProject,
-                        rewritten.getResolvedIndexExpressions()
+                        rewritten.getResolvedIndexExpressions(),
+                        rewritten.getProjectRouting()
                     );
                 }
             }
@@ -1065,7 +1066,8 @@ static void collectSearchShards(
         ActionListener<Map<String, SearchShardsResponse>> listener,
         TimeValue forceConnectTimeoutSecs,
         boolean resolvesCrossProject,
-        ResolvedIndexExpressions originResolvedIdxExpressions
+        ResolvedIndexExpressions originResolvedIdxExpressions,
+        String projectRouting
     ) {
         RemoteClusterService remoteClusterService = transportService.getRemoteClusterService();
         final CountDown responsesCountDown = new CountDown(remoteIndicesByCluster.size());
@@ -1104,7 +1106,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
                         // We do not use the relaxed index options here when validating indices' existence.
                         ElasticsearchException validationEx = CrossProjectIndexResolutionValidator.validate(
                             originalIdxOpts,
-                            null,
+                            projectRouting,
                             originResolvedIdxExpressions,
                             resolvedIndexExpressions
                         );

server/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
 import org.elasticsearch.search.fetch.StoredFieldsContext;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.search.internal.SearchContext;
@@ -68,15 +69,13 @@ public class RestSearchAction extends BaseRestHandler {
     private final SearchUsageHolder searchUsageHolder;
     private final Predicate<NodeFeature> clusterSupportsFeature;
     private final Settings settings;
-
-    public RestSearchAction(SearchUsageHolder searchUsageHolder, Predicate<NodeFeature> clusterSupportsFeature) {
-        this(searchUsageHolder, clusterSupportsFeature, null);
-    }
+    private final CrossProjectModeDecider crossProjectModeDecider;
 
     public RestSearchAction(SearchUsageHolder searchUsageHolder, Predicate<NodeFeature> clusterSupportsFeature, Settings settings) {
         this.searchUsageHolder = searchUsageHolder;
         this.clusterSupportsFeature = clusterSupportsFeature;
         this.settings = settings;
+        this.crossProjectModeDecider = new CrossProjectModeDecider(settings);
     }
 
     @Override
@@ -109,10 +108,9 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         // this might be set by old clients
         request.param("min_compatible_shard_node");
 
-        final boolean crossProjectEnabled = settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false);
+        final boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
         if (crossProjectEnabled) {
-            // accept but drop project_routing param until fully supported
-            request.param("project_routing");
+            searchRequest.setProjectRouting(request.param("project_routing"));
         }
 
         /*
@@ -202,11 +200,18 @@ public static void parseSearchRequest(
             searchRequest.source(new SearchSourceBuilder());
         }
         searchRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
+        /*
+         * We pass this object to the request body parser so that we can extract info such as project_routing.
+         * We only do it if in a Cross Project Environment, though, because outside it, such details are not
+         * expected and valid.
+         */
+        SearchRequest searchRequestForParsing = crossProjectEnabled ? searchRequest : null;
         if (requestContentParser != null) {
             if (searchUsageHolder == null) {
-                searchRequest.source().parseXContent(requestContentParser, true, clusterSupportsFeature);
+                searchRequest.source().parseXContent(searchRequestForParsing, requestContentParser, true, clusterSupportsFeature);
             } else {
-                searchRequest.source().parseXContent(requestContentParser, true, searchUsageHolder, clusterSupportsFeature);
+                searchRequest.source()
+                    .parseXContent(searchRequestForParsing, requestContentParser, true, searchUsageHolder, clusterSupportsFeature);
             }
         }
 

server/src/main/java/org/elasticsearch/search/builder/SearchSourceBuilder.java

@@ -129,6 +129,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
     public static final ParseField POINT_IN_TIME = new ParseField("pit");
     public static final ParseField RUNTIME_MAPPINGS_FIELD = new ParseField("runtime_mappings");
     public static final ParseField RETRIEVER = new ParseField("retriever");
+    public static final ParseField PROJECT_ROUTING = new ParseField("project_routing");
 
     private static final boolean RANK_SUPPORTED = Booleans.parseBoolean(System.getProperty("es.search.rank_supported"), true);
 
@@ -1329,18 +1330,62 @@ private SearchSourceBuilder shallowCopy(
     /**
      * Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
      *
+     * @param searchRequest The SearchRequest object that's representing the request we're parsing which shall receive
+     *                      the parsed info. Currently, this is non-null only if we expect project_routing to appear in
+     *                      the request body, and we allow it to appear because we're in a Cross Project Search
+     *                      environment and require this info.
      * @param parser The xContent parser.
      * @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
      * @param searchUsageHolder holder for the search usage statistics
      * @param clusterSupportsFeature used to check if certain features are available on this cluster
      */
     public SearchSourceBuilder parseXContent(
+        SearchRequest searchRequest,
         XContentParser parser,
         boolean checkTrailingTokens,
         SearchUsageHolder searchUsageHolder,
         Predicate<NodeFeature> clusterSupportsFeature
     ) throws IOException {
-        return parseXContent(parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
+        return parseXContent(searchRequest, parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
+    }
+
+    /**
+     * Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
+     *
+     * @param parser The xContent parser.
+     * @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
+     * @param searchUsageHolder holder for the search usage statistics
+     * @param clusterSupportsFeature used to check if certain features are available on this cluster
+     */
+    public SearchSourceBuilder parseXContent(
+        XContentParser parser,
+        boolean checkTrailingTokens,
+        SearchUsageHolder searchUsageHolder,
+        Predicate<NodeFeature> clusterSupportsFeature
+    ) throws IOException {
+        return parseXContent(null, parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
+    }
+
+    /**
+     * Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
+     * This variant does not record search features usage. Most times the variant that accepts a {@link SearchUsageHolder} and records
+     * usage stats into it is the one to use.
+     *
+     * @param searchRequest The SearchRequest object that's representing the request we're parsing which shall receive
+     *                      the parsed info. Currently, this is non-null only if we expect project_routing to appear in
+     *                      the request body, and we allow it to appear because we're in a Cross Project Search
+     *                      environment and require this info.
+     * @param parser The xContent parser.
+     * @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
+     * @param clusterSupportsFeature used to check if certain features are available on this cluster
+     */
+    public SearchSourceBuilder parseXContent(
+        SearchRequest searchRequest,
+        XContentParser parser,
+        boolean checkTrailingTokens,
+        Predicate<NodeFeature> clusterSupportsFeature
+    ) throws IOException {
+        return parseXContent(searchRequest, parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
     }
 
     /**
@@ -1357,10 +1402,11 @@ public SearchSourceBuilder parseXContent(
         boolean checkTrailingTokens,
         Predicate<NodeFeature> clusterSupportsFeature
     ) throws IOException {
-        return parseXContent(parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
+        return parseXContent(null, parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
     }
 
     private SearchSourceBuilder parseXContent(
+        SearchRequest searchRequest,
         XContentParser parser,
         boolean checkTrailingTokens,
         Consumer<SearchUsage> searchUsageConsumer,
@@ -1382,7 +1428,13 @@ private SearchSourceBuilder parseXContent(
             if (token == XContentParser.Token.FIELD_NAME) {
                 currentFieldName = parser.currentName();
             } else if (token.isValue()) {
-                if (FROM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
+                if (PROJECT_ROUTING.match(currentFieldName, parser.getDeprecationHandler()) && searchRequest != null) {
+                    /*
+                     * If project_routing was specified as a query parameter too, setProjectRouting() will throw
+                     * an error to prevent setting twice or overwriting previously set value.
+                     */
+                    searchRequest.setProjectRouting(parser.text());
+                } else if (FROM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     from(parser.intValue());
                 } else if (SIZE_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     size(parser.intValue());

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

@@ -1104,6 +1104,7 @@ public void testCollectSearchShards() throws Exception {
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
                     null,
                     false,
+                    null,
                     null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1136,6 +1137,7 @@ public void testCollectSearchShards() throws Exception {
                     new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
                     null,
                     false,
+                    null,
                     null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1191,6 +1193,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
                     null,
                     false,
+                    null,
                     null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1224,6 +1227,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
                     null,
                     false,
+                    null,
                     null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1273,6 +1277,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
                     null,
                     false,
+                    null,
                     null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);

server/src/test/java/org/elasticsearch/rest/action/search/RestSearchActionTests.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.suggest.SuggestBuilder;
@@ -33,7 +34,7 @@ public final class RestSearchActionTests extends RestActionTestCase {
 
     @Before
     public void setUpAction() {
-        action = new RestSearchAction(new UsageService().getSearchUsageHolder(), nf -> false);
+        action = new RestSearchAction(new UsageService().getSearchUsageHolder(), nf -> false, Settings.EMPTY);
         controller().registerHandler(action);
         verifyingClient.setExecuteVerifier((actionType, request) -> mock(SearchResponse.class));
         verifyingClient.setExecuteLocallyVerifier((actionType, request) -> mock(SearchResponse.class));

x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/ProjectRoutingDisallowedInNonCpsEnvIT.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.search;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.util.Collection;
+
+public class ProjectRoutingDisallowedInNonCpsEnvIT extends ESIntegTestCase {
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false;
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return CollectionUtils.appendToCopyNoNullElements(super.nodePlugins(), AsyncSearch.class);
+    }
+
+    public void testDisallowProjectRouting() throws IOException {
+        Request createAsyncRequest = new Request("POST", "/*,*:*/" + randomFrom("_async_search", "_search"));
+        createAsyncRequest.setJsonEntity("""
+            {
+              "project_routing": "_alias:_origin"
+            }
+            """);
+
+        ResponseException err = expectThrows(ResponseException.class, () -> getRestClient().performRequest(createAsyncRequest));
+        assertThat(err.toString(), Matchers.containsString("Unknown key for a VALUE_STRING in [project_routing]"));
+    }
+}

x-pack/plugin/async-search/src/main/java/org/elasticsearch/xpack/search/RestSubmitAsyncSearchAction.java

@@ -70,8 +70,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
 
         boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
         if (crossProjectEnabled) {
-            // accept but drop project_routing param until fully supported
-            request.param("project_routing");
+            submit.getSearchRequest().setProjectRouting(request.param("project_routing"));
         }
 
         IntConsumer setSize = size -> submit.getSearchRequest().source().size(size);