EQL: Start adding support for CPS

luigidellaquila · luigidellaquila · commit 59a6c364aa99 · 2025-10-29T11:04:11.000+01:00
diff --git a/server/src/main/resources/transport/definitions/referable/eql_project_routing.csv b/server/src/main/resources/transport/definitions/referable/eql_project_routing.csv
@@ -0,0 +1 @@
+9204000
diff --git a/server/src/main/resources/transport/upper_bounds/9.3.csv b/server/src/main/resources/transport/upper_bounds/9.3.csv
@@ -1 +1 @@
-esql_resolve_fields_response_removed_min_tv,9203000
+eql_project_routing,9204000
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchRequest.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchRequest.java
@@ -6,10 +6,12 @@
  */
 package org.elasticsearch.xpack.eql.action;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 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;
@@ -44,6 +46,8 @@
 
 public class EqlSearchRequest extends LegacyActionRequest implements IndicesRequest.Replaceable, ToXContent {
 
+    private static final TransportVersion EQL_PROJECT_ROUTING = TransportVersion.fromName("eql_project_routing");
+
     public static final long MIN_KEEP_ALIVE = TimeValue.timeValueMinutes(1).millis();
     public static final TimeValue DEFAULT_KEEP_ALIVE = TimeValue.timeValueDays(5);
     public static final IndicesOptions DEFAULT_INDICES_OPTIONS = IndicesOptions.fromOptions(true, true, true, false);
@@ -65,6 +69,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;
@@ -140,6 +146,19 @@ public EqlSearchRequest(StreamInput in) throws IOException {
             allowPartialSearchResults = false;
             allowPartialSequenceResults = false;
         }
+        if (in.getTransportVersion().supports(EQL_PROJECT_ROUTING)) {
+            projectRouting = in.readOptionalString();
+            resolvedIndexExpressions = in.readOptionalWriteable(ResolvedIndexExpressions::new);
+        }
+    }
+
+    public String projectRouting() {
+        return projectRouting;
+    }
+
+    public EqlSearchRequest projectRouting(String projectRouting) {
+        this.projectRouting = projectRouting;
+        return this;
     }
 
     @Override
@@ -297,6 +316,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;
     }
@@ -495,6 +529,10 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalBoolean(allowPartialSearchResults);
             out.writeOptionalBoolean(allowPartialSequenceResults);
         }
+        if (out.getTransportVersion().supports(EQL_PROJECT_ROUTING)) {
+            out.writeOptionalString(projectRouting);
+            out.writeOptionalWriteable(resolvedIndexExpressions);
+        }
     }
 
     @Override
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/RestEqlSearchAction.java b/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())
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/TransportEqlSearchAction.java b/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,6 +239,7 @@ public static void operation(
                 request.allowPartialSequenceResults() == null
                     ? defaultAllowPartialSequenceResults(clusterService)
                     : request.allowPartialSequenceResults(),
+                request.projectRouting(),
                 clientId,
                 new TaskId(nodeId, task.getId()),
                 task
@@ -288,7 +289,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;
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlConfiguration.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlConfiguration.java
@@ -32,6 +32,7 @@ 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;
 
     @Nullable
     private final QueryBuilder filter;
@@ -54,6 +55,7 @@ public EqlConfiguration(
         int maxSamplesPerKey,
         boolean allowPartialSearchResults,
         boolean allowPartialSequenceResults,
+        String projectRouting,
         String clientId,
         TaskId taskId,
         EqlSearchTask task
@@ -73,6 +75,11 @@ public EqlConfiguration(
         this.maxSamplesPerKey = maxSamplesPerKey;
         this.allowPartialSearchResults = allowPartialSearchResults;
         this.allowPartialSequenceResults = allowPartialSequenceResults;
+        this.projectRouting = projectRouting;
+    }
+
+    public String projectRouting() {
+        return projectRouting;
     }
 
     public String[] indices() {
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlSession.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlSession.java
@@ -122,6 +122,7 @@ private <T> void preAnalyze(LogicalPlan parsed, ActionListener<LogicalPlan> list
             return;
         }
         Set<String> fieldNames = fieldNames(parsed);
+        // TODO pass configuration.projectRouting();
         indexResolver.resolveAsMergedMapping(
             indexWildcard,
             fieldNames,
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/EqlTestUtils.java b/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()
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/search/PITAwareQueryClientTests.java b/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(
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/sequence/CircuitBreakerTests.java b/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/sequence/CircuitBreakerTests.java
@@ -352,6 +352,7 @@ private QueryClient buildQueryClient(ESMockClient esClient, CircuitBreaker eqlCi
             1,
             randomBoolean(),
             randomBoolean(),
+            null,
             "",
             new TaskId("test", 123),
             new EqlSearchTask(
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/sequence/PITFailureTests.java b/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/execution/sequence/PITFailureTests.java
@@ -85,6 +85,7 @@ public void testHandlingPitFailure() {
                 1,
                 randomBoolean(),
                 randomBoolean(),
+                null,
                 "",
                 new TaskId("test", 123),
                 new EqlSearchTask(

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-esql_resolve_fields_response_removed_min_tv,9203000`
	`1`	`+eql_project_routing,9204000`
Original file line number	Diff line number	Diff line change
`@@ -122,6 +122,7 @@ private <T> void preAnalyze(LogicalPlan parsed, ActionListener<LogicalPlan> list`
`122`	`122`	`return;`
`123`	`123`	`}`
`124`	`124`	`Set<String> fieldNames = fieldNames(parsed);`
	`125`	`+ // TODO pass configuration.projectRouting();`
`125`	`126`	`indexResolver.resolveAsMergedMapping(`
`126`	`127`	`indexWildcard,`
`127`	`128`	`fieldNames,`