EQL: enable CPS (#137833)

Event queries only for now, since PIT is not supported yet

Author: luigidellaquila

docs/changelog/137833.yaml

@@ -0,0 +1,5 @@
+pr: 137833
+summary: Enable CPS
+area: EQL
+type: enhancement
+issues: []

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchRequest.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.LegacyActionRequest;
+import org.elasticsearch.action.ResolvedIndexExpressions;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -65,6 +66,8 @@ public class EqlSearchRequest extends LegacyActionRequest implements IndicesRequ
     private int maxSamplesPerKey = RequestDefaults.MAX_SAMPLES_PER_KEY;
     private Boolean allowPartialSearchResults;
     private Boolean allowPartialSequenceResults;
+    private String projectRouting;
+    private ResolvedIndexExpressions resolvedIndexExpressions;
 
     // Async settings
     private TimeValue waitForCompletionTimeout = null;
@@ -142,6 +145,16 @@ public EqlSearchRequest(StreamInput in) throws IOException {
         }
     }
 
+    @Override
+    public String getProjectRouting() {
+        return projectRouting;
+    }
+
+    public EqlSearchRequest projectRouting(String projectRouting) {
+        this.projectRouting = projectRouting;
+        return this;
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = null;
@@ -297,6 +310,21 @@ public EqlSearchRequest indices(String... indices) {
         return this;
     }
 
+    @Override
+    public boolean allowsCrossProject() {
+        return true;
+    }
+
+    @Override
+    public void setResolvedIndexExpressions(ResolvedIndexExpressions expressions) {
+        this.resolvedIndexExpressions = expressions;
+    }
+
+    @Override
+    public ResolvedIndexExpressions getResolvedIndexExpressions() {
+        return resolvedIndexExpressions;
+    }
+
     public QueryBuilder filter() {
         return this.filter;
     }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/BasicQueryClient.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.index.query.IdsQueryBuilder;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.tasks.TaskCancelledException;
@@ -61,7 +62,6 @@ public void query(QueryRequest request, ActionListener<SearchResponse> listener)
         SearchSourceBuilder searchSource = request.searchSource();
         // set query timeout
         searchSource.timeout(cfg.requestTimeout());
-
         SearchRequest search = prepareRequest(searchSource, false, allowPartialSearchResults, indices);
         search(search, allowPartialSearchResults, searchLogListener(listener, log, allowPartialSearchResults));
     }
@@ -76,6 +76,9 @@ protected void search(SearchRequest search, boolean allowPartialSearchResults, A
             log.trace("About to execute query {} on {}", StringUtils.toString(search.source()), indices);
         }
 
+        if (cfg.crossProjectEnabled()) {
+            search.indicesOptions(CrossProjectIndexResolutionValidator.indicesOptionsForCrossProjectFanout(search.indicesOptions()));
+        }
         client.search(search, listener);
     }
 
@@ -92,7 +95,9 @@ protected void search(MultiSearchRequest search, boolean allowPartialSearchResul
             }
             log.trace("About to execute multi-queries {} on {}", sj, indices);
         }
-
+        if (cfg.crossProjectEnabled()) {
+            search.indicesOptions(CrossProjectIndexResolutionValidator.indicesOptionsForCrossProjectFanout(search.indicesOptions()));
+        }
         client.multiSearch(search, multiSearchLogListener(listener, allowPartialSearchResults, log));
     }
 

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/PITAwareQueryClient.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
 import org.elasticsearch.action.search.TransportOpenPointInTimeAction;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.util.CollectionUtils;
@@ -25,6 +26,7 @@
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.search.builder.PointInTimeBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator;
 import org.elasticsearch.xpack.eql.session.EqlSession;
 import org.elasticsearch.xpack.ql.index.IndexResolver;
 
@@ -136,10 +138,15 @@ private <Response> ActionListener<Response> pitListener(
     }
 
     private <Response> void openPIT(ActionListener<Response> listener, Runnable runnable, boolean allowPartialSearchResults) {
-        OpenPointInTimeRequest request = new OpenPointInTimeRequest(indices).indicesOptions(IndexResolver.FIELD_CAPS_INDICES_OPTIONS)
+        IndicesOptions indicesOptions = IndexResolver.FIELD_CAPS_INDICES_OPTIONS;
+        if (cfg.crossProjectEnabled()) {
+            indicesOptions = CrossProjectIndexResolutionValidator.indicesOptionsForCrossProjectFanout(indicesOptions);
+        }
+        OpenPointInTimeRequest request = new OpenPointInTimeRequest(indices).indicesOptions(indicesOptions)
             .keepAlive(keepAlive)
             .allowPartialSearchResults(allowPartialSearchResults);
         request.indexFilter(filter);
+
         client.execute(TransportOpenPointInTimeAction.TYPE, request, listener.delegateFailureAndWrap((l, r) -> {
             pitId = r.getPointInTimeId();
             runnable.run();

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/RestEqlSearchAction.java

@@ -53,18 +53,21 @@ public List<Route> routes() {
 
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        if (settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false)) {
-            // accept but drop project_routing param until fully supported
-            request.param("project_routing");
-        }
-
+        final boolean crossProjectEnabled = settings != null && settings.getAsBoolean("serverless.cross_project.enabled", false);
         EqlSearchRequest eqlRequest;
         String indices;
         try (XContentParser parser = request.contentOrSourceParamParser()) {
             eqlRequest = EqlSearchRequest.fromXContent(parser);
             indices = request.param("index");
             eqlRequest.indices(Strings.splitStringByCommaToArray(indices));
-            eqlRequest.indicesOptions(IndicesOptions.fromRequest(request, eqlRequest.indicesOptions()));
+            IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, eqlRequest.indicesOptions());
+            if (crossProjectEnabled) {
+                indicesOptions = IndicesOptions.builder(indicesOptions)
+                    .crossProjectModeOptions(new IndicesOptions.CrossProjectModeOptions(true))
+                    .build();
+                eqlRequest.projectRouting(request.param("project_routing"));
+            }
+            eqlRequest.indicesOptions(indicesOptions);
             if (request.hasParam("wait_for_completion_timeout")) {
                 eqlRequest.waitForCompletionTimeout(
                     request.paramAsTime("wait_for_completion_timeout", eqlRequest.waitForCompletionTimeout())

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/TransportEqlSearchAction.java

@@ -193,7 +193,7 @@ public static void operation(
             request.indicesOptions()
         );
         Set<String> clusterAliases = remoteClusterRegistry.clusterAliases(request.indices(), false);
-        if (canMinimizeRountrips(request, clusterAliases)) {
+        if (canMinimizeRountrips(request, clusterAliases, transportService.getRemoteClusterService().crossProjectEnabled())) {
             String clusterAlias = clusterAliases.iterator().next();
             String[] remoteIndices = new String[request.indices().length];
             for (int i = 0; i < request.indices().length; i++) {
@@ -239,9 +239,12 @@ public static void operation(
                 request.allowPartialSequenceResults() == null
                     ? defaultAllowPartialSequenceResults(clusterService)
                     : request.allowPartialSequenceResults(),
+                request.getProjectRouting(),
                 clientId,
                 new TaskId(nodeId, task.getId()),
-                task
+                task,
+                transportService.getRemoteClusterService().crossProjectEnabled(),
+                request.getResolvedIndexExpressions()
             );
             planExecutor.eql(
                 cfg,
@@ -288,7 +291,10 @@ private static boolean requestIsAsync(EqlSearchRequest request) {
     }
 
     // can the request be proxied to the remote cluster?
-    private static boolean canMinimizeRountrips(EqlSearchRequest request, Set<String> clusterAliases) {
+    private static boolean canMinimizeRountrips(EqlSearchRequest request, Set<String> clusterAliases, boolean crossProjectEnabled) {
+        if (crossProjectEnabled) {
+            return false;
+        }
         // Has minimizing the round trips been (explicitly) disabled?
         if (request.ccsMinimizeRoundtrips() == false) {
             return false;

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlConfiguration.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.eql.session;
 
+import org.elasticsearch.action.ResolvedIndexExpressions;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.Nullable;
@@ -32,6 +33,9 @@ public class EqlConfiguration extends org.elasticsearch.xpack.ql.session.Configu
     private final int maxSamplesPerKey;
     private final boolean allowPartialSearchResults;
     private final boolean allowPartialSequenceResults;
+    private final String projectRouting;
+    private final boolean crossProjectEnabled;
+    private final ResolvedIndexExpressions resolvedIndexExpressions;
 
     @Nullable
     private final QueryBuilder filter;
@@ -54,9 +58,54 @@ public EqlConfiguration(
         int maxSamplesPerKey,
         boolean allowPartialSearchResults,
         boolean allowPartialSequenceResults,
+        String projectRouting,
         String clientId,
         TaskId taskId,
         EqlSearchTask task
+    ) {
+        this(
+            indices,
+            zi,
+            username,
+            clusterName,
+            filter,
+            runtimeMappings,
+            fetchFields,
+            requestTimeout,
+            indicesOptions,
+            fetchSize,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            projectRouting,
+            clientId,
+            taskId,
+            task,
+            false,
+            null
+        );
+    }
+
+    public EqlConfiguration(
+        String[] indices,
+        ZoneId zi,
+        String username,
+        String clusterName,
+        QueryBuilder filter,
+        Map<String, Object> runtimeMappings,
+        List<FieldAndFormat> fetchFields,
+        TimeValue requestTimeout,
+        IndicesOptions indicesOptions,
+        int fetchSize,
+        int maxSamplesPerKey,
+        boolean allowPartialSearchResults,
+        boolean allowPartialSequenceResults,
+        String projectRouting,
+        String clientId,
+        TaskId taskId,
+        EqlSearchTask task,
+        boolean crossProjectEnabled,
+        ResolvedIndexExpressions resolvedIndexExpressions
     ) {
         super(zi, username, clusterName);
 
@@ -73,6 +122,21 @@ public EqlConfiguration(
         this.maxSamplesPerKey = maxSamplesPerKey;
         this.allowPartialSearchResults = allowPartialSearchResults;
         this.allowPartialSequenceResults = allowPartialSequenceResults;
+        this.projectRouting = projectRouting;
+        this.crossProjectEnabled = crossProjectEnabled;
+        this.resolvedIndexExpressions = resolvedIndexExpressions;
+    }
+
+    public boolean crossProjectEnabled() {
+        return crossProjectEnabled;
+    }
+
+    public ResolvedIndexExpressions resolvedIndexExpressions() {
+        return resolvedIndexExpressions;
+    }
+
+    public String projectRouting() {
+        return projectRouting;
     }
 
     public String[] indices() {
@@ -130,4 +194,5 @@ public boolean isCancelled() {
     public TaskId getTaskId() {
         return taskId;
     }
+
 }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlSession.java

@@ -127,6 +127,9 @@ private <T> void preAnalyze(LogicalPlan parsed, ActionListener<LogicalPlan> list
             fieldNames,
             configuration.indicesOptions(),
             configuration.runtimeMappings(),
+            configuration.crossProjectEnabled(),
+            configuration.projectRouting(),
+            configuration.resolvedIndexExpressions(),
             map(listener, r -> preAnalyzer.preAnalyze(parsed, r))
         );
     }

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/EqlTestUtils.java

@@ -53,6 +53,7 @@ private EqlTestUtils() {}
         1,
         false,
         true,
+        null,
         "",
         new TaskId("test", 123),
         null
@@ -73,6 +74,7 @@ public static EqlConfiguration randomConfiguration() {
             randomIntBetween(1, 1000),
             randomBoolean(),
             randomBoolean(),
+            null,
             randomAlphaOfLength(16),
             new TaskId(randomAlphaOfLength(10), randomNonNegativeLong()),
             randomTask()

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/search/PITAwareQueryClientTests.java

@@ -104,6 +104,7 @@ public void testQueryFilterUsedInPitAndSearches() {
                 1,
                 randomBoolean(),
                 randomBoolean(),
+                null,
                 "",
                 new TaskId("test", 123),
                 new EqlSearchTask(