Skip to content

Commit bee48f8

Browse files
piergmpiergm
authored
allows PIT to be cross project (#137966)
* wip * cps indices options * iter * iter * iter * iter * Update docs/changelog/137966.yaml * iter * iter * iter * iter * iter * project routing can be present in the body of the request * changed logger * iter * iter
1 parent ec4daed commit bee48f8

File tree

8 files changed

+249
-11
lines changed

8 files changed

+249
-11
lines changed

‎docs/changelog/137966.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 137966
2+
summary: Allows PIT to be cross project
3+
area: Search
4+
type: enhancement
5+
issues: []

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

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.elasticsearch.action.ActionRequestValidationException;
1414
import org.elasticsearch.action.IndicesRequest;
1515
import org.elasticsearch.action.LegacyActionRequest;
16+
import org.elasticsearch.action.ResolvedIndexExpressions;
1617
import org.elasticsearch.action.support.IndicesOptions;
1718
import org.elasticsearch.common.io.stream.StreamInput;
1819
import org.elasticsearch.common.io.stream.StreamOutput;
@@ -40,11 +41,16 @@ public final class OpenPointInTimeRequest extends LegacyActionRequest implements
4041
@Nullable
4142
private String preference;
4243

44+
private ResolvedIndexExpressions resolvedIndexExpressions;
45+
@Nullable
46+
private String projectRouting;
47+
4348
private QueryBuilder indexFilter;
4449

4550
private boolean allowPartialSearchResults = false;
4651

4752
public static final IndicesOptions DEFAULT_INDICES_OPTIONS = SearchRequest.DEFAULT_INDICES_OPTIONS;
53+
public static final IndicesOptions DEFAULT_CPS_INDICES_OPTIONS = SearchRequest.DEFAULT_CPS_INDICES_OPTIONS;
4854

4955
public OpenPointInTimeRequest(String... indices) {
5056
this.indices = Objects.requireNonNull(indices, "[index] is not specified");
@@ -186,6 +192,33 @@ public boolean allowsRemoteIndices() {
186192
return true;
187193
}
188194

195+
@Override
196+
public void setResolvedIndexExpressions(ResolvedIndexExpressions expressions) {
197+
this.resolvedIndexExpressions = expressions;
198+
}
199+
200+
@Override
201+
public ResolvedIndexExpressions getResolvedIndexExpressions() {
202+
return resolvedIndexExpressions;
203+
}
204+
205+
@Override
206+
public boolean allowsCrossProject() {
207+
return true;
208+
}
209+
210+
@Override
211+
public String getProjectRouting() {
212+
return projectRouting;
213+
}
214+
215+
public void projectRouting(@Nullable String projectRouting) {
216+
if (this.projectRouting != null) {
217+
throw new IllegalArgumentException("project_routing is already set to [" + this.projectRouting + "]");
218+
}
219+
this.projectRouting = projectRouting;
220+
}
221+
189222
@Override
190223
public boolean includeDataStreams() {
191224
return true;

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

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.elasticsearch.rest.Scope;
2020
import org.elasticsearch.rest.ServerlessScope;
2121
import org.elasticsearch.rest.action.RestToXContentListener;
22+
import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
2223
import org.elasticsearch.xcontent.ObjectParser;
2324
import org.elasticsearch.xcontent.ParseField;
2425

@@ -31,10 +32,10 @@
3132
@ServerlessScope(Scope.PUBLIC)
3233
public class RestOpenPointInTimeAction extends BaseRestHandler {
3334

34-
private final Settings settings;
35+
private final CrossProjectModeDecider crossProjectModeDecider;
3536

3637
public RestOpenPointInTimeAction(Settings settings) {
37-
this.settings = settings;
38+
this.crossProjectModeDecider = new CrossProjectModeDecider(settings);
3839
}
3940

4041
@Override
@@ -49,14 +50,16 @@ public List<Route> routes() {
4950

5051
@Override
5152
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
52-
if (settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false)) {
53-
// accept but drop project_routing param until fully supported
54-
request.param("project_routing");
55-
}
5653

5754
final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
5855
final OpenPointInTimeRequest openRequest = new OpenPointInTimeRequest(indices);
59-
openRequest.indicesOptions(IndicesOptions.fromRequest(request, OpenPointInTimeRequest.DEFAULT_INDICES_OPTIONS));
56+
final boolean crossProjectEnabled = crossProjectModeDecider.crossProjectEnabled();
57+
if (crossProjectEnabled) {
58+
openRequest.projectRouting(request.param("project_routing", null));
59+
openRequest.indicesOptions(IndicesOptions.fromRequest(request, OpenPointInTimeRequest.DEFAULT_CPS_INDICES_OPTIONS));
60+
} else {
61+
openRequest.indicesOptions(IndicesOptions.fromRequest(request, OpenPointInTimeRequest.DEFAULT_INDICES_OPTIONS));
62+
}
6063
openRequest.routing(request.param("routing"));
6164
openRequest.preference(request.param("preference"));
6265
openRequest.keepAlive(TimeValue.parseTimeValue(request.param("keep_alive"), null, "keep_alive"));
@@ -80,8 +83,10 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
8083

8184
private static final ObjectParser<OpenPointInTimeRequest, Void> PARSER = new ObjectParser<>("open_point_in_time_request");
8285
private static final ParseField INDEX_FILTER_FIELD = new ParseField("index_filter");
86+
private static final ParseField PROJECT_ROUTING = new ParseField("project_routing");
8387

8488
static {
8589
PARSER.declareObject(OpenPointInTimeRequest::indexFilter, (p, c) -> parseTopLevelQuery(p), INDEX_FILTER_FIELD);
90+
PARSER.declareString(OpenPointInTimeRequest::projectRouting, PROJECT_ROUTING);
8691
}
8792
}

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@ public class SearchRequest extends LegacyActionRequest implements IndicesRequest
102102
private boolean ccsMinimizeRoundtrips;
103103

104104
public static final IndicesOptions DEFAULT_INDICES_OPTIONS = IndicesOptions.strictExpandOpenAndForbidClosedIgnoreThrottled();
105+
public static final IndicesOptions DEFAULT_CPS_INDICES_OPTIONS = IndicesOptions.cpsStrictExpandOpenAndForbidClosedIgnoreThrottled();
105106

106107
private IndicesOptions indicesOptions = DEFAULT_INDICES_OPTIONS;
107108

@@ -380,9 +381,13 @@ public ActionRequestValidationException validate() {
380381
validationException
381382
);
382383
}
383-
if (indicesOptions().equals(DEFAULT_INDICES_OPTIONS) == false) {
384+
if (indicesOptions().equals(DEFAULT_INDICES_OPTIONS) == false
385+
&& indicesOptions().equals(DEFAULT_CPS_INDICES_OPTIONS) == false) {
384386
validationException = addValidationError("[indicesOptions] cannot be used with point in time", validationException);
385387
}
388+
if (getProjectRouting() != null) {
389+
validationException = addValidationError("[projectRouting] cannot be used with point in time", validationException);
390+
}
386391
if (routing() != null) {
387392
validationException = addValidationError("[routing] cannot be used with point in time", validationException);
388393
}

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

Lines changed: 162 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,15 @@
1919
import org.elasticsearch.action.ActionType;
2020
import org.elasticsearch.action.IndicesRequest;
2121
import org.elasticsearch.action.OriginalIndices;
22+
import org.elasticsearch.action.ResolvedIndexExpression;
23+
import org.elasticsearch.action.ResolvedIndexExpressions;
24+
import org.elasticsearch.action.admin.indices.resolve.ResolveIndexAction;
2225
import org.elasticsearch.action.support.ActionFilters;
2326
import org.elasticsearch.action.support.ChannelActionListener;
27+
import org.elasticsearch.action.support.GroupedActionListener;
2428
import org.elasticsearch.action.support.HandledTransportAction;
2529
import org.elasticsearch.action.support.IndicesOptions;
30+
import org.elasticsearch.action.support.SubscribableListener;
2631
import org.elasticsearch.cluster.ClusterState;
2732
import org.elasticsearch.cluster.service.ClusterService;
2833
import org.elasticsearch.common.bytes.BytesReference;
@@ -38,25 +43,36 @@
3843
import org.elasticsearch.search.SearchPhaseResult;
3944
import org.elasticsearch.search.SearchService;
4045
import org.elasticsearch.search.builder.SearchSourceBuilder;
46+
import org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator;
47+
import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
4148
import org.elasticsearch.search.internal.AliasFilter;
4249
import org.elasticsearch.search.internal.ShardSearchContextId;
4350
import org.elasticsearch.tasks.Task;
4451
import org.elasticsearch.threadpool.ThreadPool;
4552
import org.elasticsearch.transport.AbstractTransportRequest;
53+
import org.elasticsearch.transport.RemoteClusterAware;
54+
import org.elasticsearch.transport.RemoteClusterService;
4655
import org.elasticsearch.transport.Transport;
4756
import org.elasticsearch.transport.TransportActionProxy;
4857
import org.elasticsearch.transport.TransportChannel;
4958
import org.elasticsearch.transport.TransportRequestHandler;
59+
import org.elasticsearch.transport.TransportRequestOptions;
5060
import org.elasticsearch.transport.TransportResponseHandler;
5161
import org.elasticsearch.transport.TransportService;
5262

5363
import java.io.IOException;
64+
import java.util.Collection;
65+
import java.util.HashSet;
5466
import java.util.List;
5567
import java.util.Map;
68+
import java.util.Set;
5669
import java.util.concurrent.Executor;
5770
import java.util.function.BiFunction;
71+
import java.util.stream.Collectors;
5872

5973
import static org.elasticsearch.core.Strings.format;
74+
import static org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator.indicesOptionsForCrossProjectFanout;
75+
import static org.elasticsearch.transport.RemoteClusterAware.buildRemoteIndexName;
6076

6177
public class TransportOpenPointInTimeAction extends HandledTransportAction<OpenPointInTimeRequest, OpenPointInTimeResponse> {
6278

@@ -72,6 +88,8 @@ public class TransportOpenPointInTimeAction extends HandledTransportAction<OpenP
7288
private final SearchService searchService;
7389
private final ClusterService clusterService;
7490
private final SearchResponseMetrics searchResponseMetrics;
91+
private final CrossProjectModeDecider crossProjectModeDecider;
92+
private final TimeValue forceConnectTimeoutSecs;
7593

7694
@Inject
7795
public TransportOpenPointInTimeAction(
@@ -92,6 +110,9 @@ public TransportOpenPointInTimeAction(
92110
this.namedWriteableRegistry = namedWriteableRegistry;
93111
this.clusterService = clusterService;
94112
this.searchResponseMetrics = searchResponseMetrics;
113+
this.crossProjectModeDecider = new CrossProjectModeDecider(clusterService.getSettings());
114+
this.forceConnectTimeoutSecs = clusterService.getSettings()
115+
.getAsTime("search.ccs.force_connect_timeout", TimeValue.timeValueSeconds(3L));
95116
transportService.registerRequestHandler(
96117
OPEN_SHARD_READER_CONTEXT_NAME,
97118
EsExecutors.DIRECT_EXECUTOR_SERVICE,
@@ -123,6 +144,146 @@ protected void doExecute(Task task, OpenPointInTimeRequest request, ActionListen
123144
);
124145
return;
125146
}
147+
148+
final boolean resolveCrossProject = crossProjectModeDecider.resolvesCrossProject(request);
149+
if (resolveCrossProject) {
150+
executeOpenPitCrossProject((SearchTask) task, request, listener);
151+
} else {
152+
executeOpenPit((SearchTask) task, request, listener);
153+
}
154+
}
155+
156+
private void executeOpenPitCrossProject(
157+
SearchTask task,
158+
OpenPointInTimeRequest request,
159+
ActionListener<OpenPointInTimeResponse> listener
160+
) {
161+
String[] indices = request.indices();
162+
IndicesOptions originalIndicesOptions = request.indicesOptions();
163+
// in CPS before executing the open pit request we need to get index resolution and possibly throw based on merged project view
164+
// rules. This should happen only if either ignore_unavailable or allow_no_indices is set to false (strict).
165+
// If instead both are true we can continue with the "normal" pit execution.
166+
if (originalIndicesOptions.ignoreUnavailable() && originalIndicesOptions.allowNoIndices()) {
167+
// lenient indicesOptions thus execute standard pit
168+
executeOpenPit(task, request, listener);
169+
return;
170+
}
171+
172+
// ResolvedIndexExpression for the origin cluster (only) as determined by the Security Action Filter
173+
final ResolvedIndexExpressions localResolvedIndexExpressions = request.getResolvedIndexExpressions();
174+
175+
RemoteClusterService remoteClusterService = searchTransportService.getRemoteClusterService();
176+
final Map<String, OriginalIndices> indicesPerCluster = remoteClusterService.groupIndices(
177+
indicesOptionsForCrossProjectFanout(originalIndicesOptions),
178+
indices
179+
);
180+
// local indices resolution was already taken care of by the Security Action Filter
181+
indicesPerCluster.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
182+
183+
if (indicesPerCluster.isEmpty()) {
184+
// for CPS requests that are targeting origin only, could be because of project_routing or other reasons, execute standard pit.
185+
final Exception ex = CrossProjectIndexResolutionValidator.validate(
186+
originalIndicesOptions,
187+
request.getProjectRouting(),
188+
localResolvedIndexExpressions,
189+
Map.of()
190+
);
191+
if (ex != null) {
192+
listener.onFailure(ex);
193+
return;
194+
}
195+
executeOpenPit(task, request, listener);
196+
return;
197+
}
198+
199+
// CPS
200+
final int linkedProjectsToQuery = indicesPerCluster.size();
201+
ActionListener<Collection<Map.Entry<String, ResolveIndexAction.Response>>> responsesListener = listener.delegateFailureAndWrap(
202+
(l, responses) -> {
203+
Map<String, ResolvedIndexExpressions> resolvedRemoteExpressions = responses.stream()
204+
.filter(e -> e.getValue().getResolvedIndexExpressions() != null)
205+
.collect(
206+
Collectors.toMap(
207+
Map.Entry::getKey,
208+
e -> e.getValue().getResolvedIndexExpressions()
209+
210+
)
211+
);
212+
final Exception ex = CrossProjectIndexResolutionValidator.validate(
213+
originalIndicesOptions,
214+
request.getProjectRouting(),
215+
localResolvedIndexExpressions,
216+
resolvedRemoteExpressions
217+
);
218+
if (ex != null) {
219+
listener.onFailure(ex);
220+
return;
221+
}
222+
Set<String> collectedIndices = new HashSet<>(indices.length);
223+
224+
for (Map.Entry<String, ResolvedIndexExpressions> resolvedRemoteExpressionEntry : resolvedRemoteExpressions.entrySet()) {
225+
String remoteAlias = resolvedRemoteExpressionEntry.getKey();
226+
for (ResolvedIndexExpression expression : resolvedRemoteExpressionEntry.getValue().expressions()) {
227+
ResolvedIndexExpression.LocalExpressions oneRemoteExpression = expression.localExpressions();
228+
if (false == oneRemoteExpression.indices().isEmpty()
229+
&& oneRemoteExpression
230+
.localIndexResolutionResult() == ResolvedIndexExpression.LocalIndexResolutionResult.SUCCESS) {
231+
collectedIndices.addAll(
232+
oneRemoteExpression.indices()
233+
.stream()
234+
.map(i -> buildRemoteIndexName(remoteAlias, i))
235+
.collect(Collectors.toSet())
236+
);
237+
}
238+
}
239+
}
240+
if (localResolvedIndexExpressions != null) { // this should never be null in CPS
241+
collectedIndices.addAll(localResolvedIndexExpressions.getLocalIndicesList());
242+
}
243+
request.indices(collectedIndices.toArray(String[]::new));
244+
executeOpenPit(task, request, listener);
245+
}
246+
);
247+
ActionListener<Map.Entry<String, ResolveIndexAction.Response>> groupedListener = new GroupedActionListener<>(
248+
linkedProjectsToQuery,
249+
responsesListener
250+
);
251+
252+
// make CPS calls
253+
for (Map.Entry<String, OriginalIndices> remoteClusterIndices : indicesPerCluster.entrySet()) {
254+
String clusterAlias = remoteClusterIndices.getKey();
255+
OriginalIndices originalIndices = remoteClusterIndices.getValue();
256+
IndicesOptions relaxedFanoutIdxOptions = originalIndices.indicesOptions(); // from indicesOptionsForCrossProjectFanout
257+
ResolveIndexAction.Request remoteRequest = new ResolveIndexAction.Request(originalIndices.indices(), relaxedFanoutIdxOptions);
258+
259+
SubscribableListener<Transport.Connection> connectionListener = new SubscribableListener<>();
260+
connectionListener.addTimeout(forceConnectTimeoutSecs, transportService.getThreadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
261+
262+
connectionListener.addListener(groupedListener.delegateResponse((l, failure) -> {
263+
logger.info("failed to resolve indices on remote cluster [" + clusterAlias + "]", failure);
264+
l.onFailure(failure);
265+
})
266+
.delegateFailure(
267+
(ignored, connection) -> transportService.sendRequest(
268+
connection,
269+
ResolveIndexAction.REMOTE_TYPE.name(),
270+
remoteRequest,
271+
TransportRequestOptions.EMPTY,
272+
new ActionListenerResponseHandler<>(groupedListener.delegateResponse((l, failure) -> {
273+
logger.info("Error occurred on remote cluster [" + clusterAlias + "]", failure);
274+
l.onFailure(failure);
275+
}).map(resolveIndexResponse -> Map.entry(clusterAlias, resolveIndexResponse)),
276+
ResolveIndexAction.Response::new,
277+
EsExecutors.DIRECT_EXECUTOR_SERVICE
278+
)
279+
)
280+
));
281+
282+
remoteClusterService.maybeEnsureConnectedAndGetConnection(clusterAlias, true, connectionListener);
283+
}
284+
}
285+
286+
private void executeOpenPit(SearchTask task, OpenPointInTimeRequest request, ActionListener<OpenPointInTimeResponse> listener) {
126287
final SearchRequest searchRequest = new SearchRequest().indices(request.indices())
127288
.indicesOptions(request.indicesOptions())
128289
.preference(request.preference())
@@ -132,7 +293,7 @@ protected void doExecute(Task task, OpenPointInTimeRequest request, ActionListen
132293
searchRequest.setMaxConcurrentShardRequests(request.maxConcurrentShardRequests());
133294
searchRequest.setCcsMinimizeRoundtrips(false);
134295

135-
transportSearchAction.executeOpenPit((SearchTask) task, searchRequest, listener.map(r -> {
296+
transportSearchAction.executeOpenPit(task, searchRequest, listener.map(r -> {
136297
assert r.pointInTimeId() != null : r;
137298
return new OpenPointInTimeResponse(
138299
r.pointInTimeId(),

0 commit comments

Comments
 (0)