Skip to content

Commit 3953600

Browse files
phananh1010phananh1010
committed
Mirror upstream elastic#137566 as single snapshot commit for AI review
BASE=e916243b86560e938fa1d2c5f2607c28d7a49dd3 HEAD=0c7ee5d1fa023820437f540cf94528fe48d2488b Branch=main
1 parent e916243 commit 3953600

File tree

8 files changed

+91
-18
lines changed

8 files changed

+91
-18
lines changed

‎docs/changelog/137566.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 137566
2+
summary: Add support for `project_routing` for `_search` and `_async_search`
3+
area: CCS
4+
type: enhancement
5+
issues: []

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,7 @@ public class SearchRequest extends LegacyActionRequest implements IndicesRequest
116116
* enabling synthetic source natively in the index.
117117
*/
118118
private boolean forceSyntheticSource = false;
119+
private String projectRouting;
119120

120121
public SearchRequest() {
121122
this.localClusterAlias = null;
@@ -168,6 +169,19 @@ public boolean allowsCrossProject() {
168169
return true;
169170
}
170171

172+
@Override
173+
public String getProjectRouting() {
174+
return projectRouting;
175+
}
176+
177+
public void setProjectRouting(@Nullable String projectRouting) {
178+
if (this.projectRouting != null) {
179+
throw new IllegalArgumentException("project_routing is already set to [" + this.projectRouting + "]");
180+
}
181+
182+
this.projectRouting = projectRouting;
183+
}
184+
171185
/**
172186
* Creates a new sub-search request starting from the original search request that is provided.
173187
* 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(
228242
this.waitForCheckpoints = searchRequest.waitForCheckpoints;
229243
this.waitForCheckpointsTimeout = searchRequest.waitForCheckpointsTimeout;
230244
this.forceSyntheticSource = searchRequest.forceSyntheticSource;
245+
this.projectRouting = searchRequest.projectRouting;
231246
}
232247

233248
/**

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -620,7 +620,8 @@ public void onFailure(Exception e) {
620620
}),
621621
forceConnectTimeoutSecs,
622622
resolvesCrossProject,
623-
rewritten.getResolvedIndexExpressions()
623+
rewritten.getResolvedIndexExpressions(),
624+
rewritten.getProjectRouting()
624625
);
625626
}
626627
}
@@ -1065,7 +1066,8 @@ static void collectSearchShards(
10651066
ActionListener<Map<String, SearchShardsResponse>> listener,
10661067
TimeValue forceConnectTimeoutSecs,
10671068
boolean resolvesCrossProject,
1068-
ResolvedIndexExpressions originResolvedIdxExpressions
1069+
ResolvedIndexExpressions originResolvedIdxExpressions,
1070+
String projectRouting
10691071
) {
10701072
RemoteClusterService remoteClusterService = transportService.getRemoteClusterService();
10711073
final CountDown responsesCountDown = new CountDown(remoteIndicesByCluster.size());
@@ -1104,7 +1106,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
11041106
// We do not use the relaxed index options here when validating indices' existence.
11051107
ElasticsearchException validationEx = CrossProjectIndexResolutionValidator.validate(
11061108
originalIdxOpts,
1107-
null,
1109+
projectRouting,
11081110
originResolvedIdxExpressions,
11091111
resolvedIndexExpressions
11101112
);

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

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
3131
import org.elasticsearch.search.SearchService;
3232
import org.elasticsearch.search.builder.SearchSourceBuilder;
33+
import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
3334
import org.elasticsearch.search.fetch.StoredFieldsContext;
3435
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
3536
import org.elasticsearch.search.internal.SearchContext;
@@ -68,15 +69,13 @@ public class RestSearchAction extends BaseRestHandler {
6869
private final SearchUsageHolder searchUsageHolder;
6970
private final Predicate<NodeFeature> clusterSupportsFeature;
7071
private final Settings settings;
71-
72-
public RestSearchAction(SearchUsageHolder searchUsageHolder, Predicate<NodeFeature> clusterSupportsFeature) {
73-
this(searchUsageHolder, clusterSupportsFeature, null);
74-
}
72+
private final CrossProjectModeDecider crossProjectModeDecider;
7573

7674
public RestSearchAction(SearchUsageHolder searchUsageHolder, Predicate<NodeFeature> clusterSupportsFeature, Settings settings) {
7775
this.searchUsageHolder = searchUsageHolder;
7876
this.clusterSupportsFeature = clusterSupportsFeature;
7977
this.settings = settings;
78+
this.crossProjectModeDecider = new CrossProjectModeDecider(settings);
8079
}
8180

8281
@Override
@@ -109,10 +108,9 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
109108
// this might be set by old clients
110109
request.param("min_compatible_shard_node");
111110

112-
final boolean crossProjectEnabled = settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false);
111+
final boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
113112
if (crossProjectEnabled) {
114-
// accept but drop project_routing param until fully supported
115-
request.param("project_routing");
113+
searchRequest.setProjectRouting(request.param("project_routing"));
116114
}
117115

118116
/*
@@ -204,9 +202,9 @@ public static void parseSearchRequest(
204202
searchRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
205203
if (requestContentParser != null) {
206204
if (searchUsageHolder == null) {
207-
searchRequest.source().parseXContent(requestContentParser, true, clusterSupportsFeature);
205+
searchRequest.source().parseXContent(searchRequest, requestContentParser, true, clusterSupportsFeature);
208206
} else {
209-
searchRequest.source().parseXContent(requestContentParser, true, searchUsageHolder, clusterSupportsFeature);
207+
searchRequest.source().parseXContent(searchRequest, requestContentParser, true, searchUsageHolder, clusterSupportsFeature);
210208
}
211209
}
212210

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

Lines changed: 51 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
129129
public static final ParseField POINT_IN_TIME = new ParseField("pit");
130130
public static final ParseField RUNTIME_MAPPINGS_FIELD = new ParseField("runtime_mappings");
131131
public static final ParseField RETRIEVER = new ParseField("retriever");
132+
public static final ParseField PROJECT_ROUTING = new ParseField("project_routing");
132133

133134
private static final boolean RANK_SUPPORTED = Booleans.parseBoolean(System.getProperty("es.search.rank_supported"), true);
134135

@@ -1329,18 +1330,58 @@ private SearchSourceBuilder shallowCopy(
13291330
/**
13301331
* Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
13311332
*
1333+
* @param searchRequest The SearchRequest object that's representing the request we're parsing which shall receive
1334+
* the parsed info.
13321335
* @param parser The xContent parser.
13331336
* @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
13341337
* @param searchUsageHolder holder for the search usage statistics
13351338
* @param clusterSupportsFeature used to check if certain features are available on this cluster
13361339
*/
13371340
public SearchSourceBuilder parseXContent(
1341+
SearchRequest searchRequest,
13381342
XContentParser parser,
13391343
boolean checkTrailingTokens,
13401344
SearchUsageHolder searchUsageHolder,
13411345
Predicate<NodeFeature> clusterSupportsFeature
13421346
) throws IOException {
1343-
return parseXContent(parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
1347+
return parseXContent(searchRequest, parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
1348+
}
1349+
1350+
/**
1351+
* Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
1352+
*
1353+
* @param parser The xContent parser.
1354+
* @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
1355+
* @param searchUsageHolder holder for the search usage statistics
1356+
* @param clusterSupportsFeature used to check if certain features are available on this cluster
1357+
*/
1358+
public SearchSourceBuilder parseXContent(
1359+
XContentParser parser,
1360+
boolean checkTrailingTokens,
1361+
SearchUsageHolder searchUsageHolder,
1362+
Predicate<NodeFeature> clusterSupportsFeature
1363+
) throws IOException {
1364+
return parseXContent(null, parser, checkTrailingTokens, searchUsageHolder::updateUsage, clusterSupportsFeature);
1365+
}
1366+
1367+
/**
1368+
* Parse some xContent into this SearchSourceBuilder, overwriting any values specified in the xContent.
1369+
* This variant does not record search features usage. Most times the variant that accepts a {@link SearchUsageHolder} and records
1370+
* usage stats into it is the one to use.
1371+
*
1372+
* @param searchRequest The SearchRequest object that's representing the request we're parsing which shall receive
1373+
* the parsed info.
1374+
* @param parser The xContent parser.
1375+
* @param checkTrailingTokens If true throws a parsing exception when extra tokens are found after the main object.
1376+
* @param clusterSupportsFeature used to check if certain features are available on this cluster
1377+
*/
1378+
public SearchSourceBuilder parseXContent(
1379+
SearchRequest searchRequest,
1380+
XContentParser parser,
1381+
boolean checkTrailingTokens,
1382+
Predicate<NodeFeature> clusterSupportsFeature
1383+
) throws IOException {
1384+
return parseXContent(searchRequest, parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
13441385
}
13451386

13461387
/**
@@ -1357,10 +1398,11 @@ public SearchSourceBuilder parseXContent(
13571398
boolean checkTrailingTokens,
13581399
Predicate<NodeFeature> clusterSupportsFeature
13591400
) throws IOException {
1360-
return parseXContent(parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
1401+
return parseXContent(null, parser, checkTrailingTokens, s -> {}, clusterSupportsFeature);
13611402
}
13621403

13631404
private SearchSourceBuilder parseXContent(
1405+
SearchRequest searchRequest,
13641406
XContentParser parser,
13651407
boolean checkTrailingTokens,
13661408
Consumer<SearchUsage> searchUsageConsumer,
@@ -1382,7 +1424,13 @@ private SearchSourceBuilder parseXContent(
13821424
if (token == XContentParser.Token.FIELD_NAME) {
13831425
currentFieldName = parser.currentName();
13841426
} else if (token.isValue()) {
1385-
if (FROM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
1427+
if (PROJECT_ROUTING.match(currentFieldName, parser.getDeprecationHandler()) && searchRequest != null) {
1428+
/*
1429+
* If project_routing was specified as a query parameter too, setProjectRouting() will throw
1430+
* an error to prevent setting twice or overwriting previously set value.
1431+
*/
1432+
searchRequest.setProjectRouting(parser.text());
1433+
} else if (FROM_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
13861434
from(parser.intValue());
13871435
} else if (SIZE_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
13881436
size(parser.intValue());

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1104,6 +1104,7 @@ public void testCollectSearchShards() throws Exception {
11041104
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
11051105
null,
11061106
false,
1107+
null,
11071108
null
11081109
);
11091110
awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1136,6 +1137,7 @@ public void testCollectSearchShards() throws Exception {
11361137
new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
11371138
null,
11381139
false,
1140+
null,
11391141
null
11401142
);
11411143
awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1191,6 +1193,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
11911193
new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
11921194
null,
11931195
false,
1196+
null,
11941197
null
11951198
);
11961199
awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1224,6 +1227,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
12241227
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
12251228
null,
12261229
false,
1230+
null,
12271231
null
12281232
);
12291233
awaitLatch(latch, 5, TimeUnit.SECONDS);
@@ -1273,6 +1277,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
12731277
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
12741278
null,
12751279
false,
1280+
null,
12761281
null
12771282
);
12781283
awaitLatch(latch, 5, TimeUnit.SECONDS);

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import org.elasticsearch.action.search.SearchRequest;
1212
import org.elasticsearch.action.search.SearchResponse;
1313
import org.elasticsearch.action.search.SearchType;
14+
import org.elasticsearch.common.settings.Settings;
1415
import org.elasticsearch.rest.RestRequest;
1516
import org.elasticsearch.search.builder.SearchSourceBuilder;
1617
import org.elasticsearch.search.suggest.SuggestBuilder;
@@ -33,7 +34,7 @@ public final class RestSearchActionTests extends RestActionTestCase {
3334

3435
@Before
3536
public void setUpAction() {
36-
action = new RestSearchAction(new UsageService().getSearchUsageHolder(), nf -> false);
37+
action = new RestSearchAction(new UsageService().getSearchUsageHolder(), nf -> false, Settings.EMPTY);
3738
controller().registerHandler(action);
3839
verifyingClient.setExecuteVerifier((actionType, request) -> mock(SearchResponse.class));
3940
verifyingClient.setExecuteLocallyVerifier((actionType, request) -> mock(SearchResponse.class));

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,8 +70,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
7070

7171
boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
7272
if (crossProjectEnabled) {
73-
// accept but drop project_routing param until fully supported
74-
request.param("project_routing");
73+
submit.getSearchRequest().setProjectRouting(request.param("project_routing"));
7574
}
7675

7776
IntConsumer setSize = size -> submit.getSearchRequest().source().size(size);

0 commit comments

Comments
 (0)