EQL: add support for partial shard results

docs/changelog/116388.yaml

@@ -0,0 +1,5 @@
+pr: 116388
+summary: Add support for partial shard results
+area: EQL
+type: enhancement
+issues: []

docs/reference/eql/eql-search-api.asciidoc

@@ -88,6 +88,53 @@ request that targets only `bar*` still returns an error.
 +
 Defaults to `true`.
 
+`allow_partial_search_results`::
+(Optional, Boolean)
+
+If `false`, the request returns an error if one or more shards involved in the query are unavailable.
++
+If `true`, the query is executed only on the available shards, ignoring shard request timeouts and
+<<shard-failures,shard failures>>.
++
+Defaults to `false`.
++
+To override the default for this field, set the
+`xpack.eql.default_allow_partial_results` cluster setting to `true`.
+
+
+[IMPORTANT]
+====
+You can also specify this value using the `allow_partial_search_results` request body parameter.
+If both parameters are specified, only the query parameter is used.
+====
+
+
+`allow_partial_sequence_results`::
+(Optional, Boolean)
+
+
+Used together with `allow_partial_search_results=true`, controls the behavior of sequence queries specifically
+(if `allow_partial_search_results=false`, this setting has no effect).
+If `true` and if some shards are unavailable, the sequences are calculated on available shards only.
++
+If `false` and if some shards are unavailable, the query only returns information about the shard failures,
+but no further results.
++
+Defaults to `false`.
++
+Consider that sequences calculated with `allow_partial_search_results=true` can return incorrect results
+(eg. if a <<eql-missing-events, missing event>> clause matches records in unavailable shards)
++
+To override the default for this field, set the
+`xpack.eql.default_allow_partial_sequence_results` cluster setting to `true`.
+
+
+[IMPORTANT]
+====
+You can also specify this value using the `allow_partial_sequence_results` request body parameter.
+If both parameters are specified, only the query parameter is used.
+====
+
 `ccs_minimize_roundtrips`::
 (Optional, Boolean) If `true`, network round-trips between the local and the
 remote cluster are minimized when running cross-cluster search (CCS) requests.

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -212,6 +212,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion LOGSDB_TELEMETRY_CUSTOM_CUTOFF_DATE = def(8_801_00_0);
     public static final TransportVersion SOURCE_MODE_TELEMETRY = def(8_802_00_0);
     public static final TransportVersion NEW_REFRESH_CLUSTER_BLOCK = def(8_803_00_0);
+    public static final TransportVersion EQL_ALLOW_PARTIAL_SEARCH_RESULTS = def(8_804_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -2453,7 +2453,7 @@ public static void afterClass() throws Exception {
     /**
      *  After the cluster is stopped, there are a few netty threads that can linger, so we make sure we don't leak any tasks on them.
      */
-    static void awaitGlobalNettyThreadsFinish() throws Exception {
+    public static void awaitGlobalNettyThreadsFinish() throws Exception {
         // Don't use GlobalEventExecutor#awaitInactivity. It will waste up to 1s for every call and we expect no tasks queued for it
         // except for the odd scheduled shutdown task.
         assertBusy(() -> assertEquals(0, GlobalEventExecutor.INSTANCE.pendingTasks()));

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/BaseEqlSpecTestCase.java

@@ -33,6 +33,9 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.is;
+
 public abstract class BaseEqlSpecTestCase extends RemoteClusterAwareEqlRestTestCase {
 
     protected static final String PARAM_FORMATTING = "%2$s";
@@ -52,6 +55,9 @@ public abstract class BaseEqlSpecTestCase extends RemoteClusterAwareEqlRestTestC
      */
     private final int size;
     private final int maxSamplesPerKey;
+    private final Boolean allowPartialSearchResults;
+    private final Boolean allowPartialSequenceResults;
+    private final Boolean expectShardFailures;
 
     @Before
     public void setup() throws Exception {
@@ -104,7 +110,16 @@ protected static List<Object[]> asArray(List<EqlSpec> specs) {
             }
 
             results.add(
-                new Object[] { spec.query(), name, spec.expectedEventIds(), spec.joinKeys(), spec.size(), spec.maxSamplesPerKey() }
+                new Object[] {
+                    spec.query(),
+                    name,
+                    spec.expectedEventIds(),
+                    spec.joinKeys(),
+                    spec.size(),
+                    spec.maxSamplesPerKey(),
+                    spec.allowPartialSearchResults(),
+                    spec.allowPartialSequenceResults(),
+                    spec.expectShardFailures() }
             );
         }
 
@@ -118,7 +133,10 @@ protected static List<Object[]> asArray(List<EqlSpec> specs) {
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
         this.index = index;
 
@@ -128,6 +146,9 @@ protected static List<Object[]> asArray(List<EqlSpec> specs) {
         this.joinKeys = joinKeys;
         this.size = size == null ? -1 : size;
         this.maxSamplesPerKey = maxSamplesPerKey == null ? -1 : maxSamplesPerKey;
+        this.allowPartialSearchResults = allowPartialSearchResults;
+        this.allowPartialSequenceResults = allowPartialSequenceResults;
+        this.expectShardFailures = expectShardFailures;
     }
 
     public void test() throws Exception {
@@ -137,6 +158,7 @@ public void test() throws Exception {
     private void assertResponse(ObjectPath response) throws Exception {
         List<Map<String, Object>> events = response.evaluate("hits.events");
         List<Map<String, Object>> sequences = response.evaluate("hits.sequences");
+        Object shardFailures = response.evaluate("shard_failures");
 
         if (events != null) {
             assertEvents(events);
@@ -145,6 +167,7 @@ private void assertResponse(ObjectPath response) throws Exception {
         } else {
             fail("No events or sequences found");
         }
+        assertShardFailures(shardFailures);
     }
 
     protected ObjectPath runQuery(String index, String query) throws Exception {
@@ -163,13 +186,34 @@ protected ObjectPath runQuery(String index, String query) throws Exception {
         if (maxSamplesPerKey > 0) {
             builder.field("max_samples_per_key", maxSamplesPerKey);
         }
+        boolean allowPartialResultsInBody = randomBoolean();
+        if (allowPartialSearchResults != null) {
+            if (allowPartialResultsInBody) {
+                builder.field("allow_partial_search_results", String.valueOf(allowPartialSearchResults));
+                if (allowPartialSequenceResults != null) {
+                    builder.field("allow_partial_sequence_results", String.valueOf(allowPartialSequenceResults));
+                }
+            }
+        } else if (randomBoolean()) {
+            builder.field("allow_partial_search_results", randomBoolean());
+        }
         builder.endObject();
 
         Request request = new Request("POST", "/" + index + "/_eql/search");
         Boolean ccsMinimizeRoundtrips = ccsMinimizeRoundtrips();
         if (ccsMinimizeRoundtrips != null) {
             request.addParameter("ccs_minimize_roundtrips", ccsMinimizeRoundtrips.toString());
         }
+        if (allowPartialSearchResults != null) {
+            if (allowPartialResultsInBody == false) {
+                request.addParameter("allow_partial_search_results", String.valueOf(allowPartialSearchResults));
+                if (allowPartialSequenceResults != null) {
+                    request.addParameter("allow_partial_sequence_results", String.valueOf(allowPartialSequenceResults));
+                }
+            }
+        } else if (randomBoolean()) {
+            request.addParameter("allow_partial_search_results", String.valueOf(randomBoolean()));
+        }
         int timeout = Math.toIntExact(timeout().millis());
         RequestConfig config = RequestConfig.copy(RequestConfig.DEFAULT)
             .setConnectionRequestTimeout(timeout)
@@ -182,6 +226,18 @@ protected ObjectPath runQuery(String index, String query) throws Exception {
         return ObjectPath.createFromResponse(client().performRequest(request));
     }
 
+    private void assertShardFailures(Object shardFailures) {
+        if (expectShardFailures != null) {
+            if (expectShardFailures) {
+                assertNotNull(shardFailures);
+                List<?> list = (List<?>) shardFailures;
+                assertThat(list.size(), is(greaterThan(0)));
+            } else {
+                assertNull(shardFailures);
+            }
+        }
+    }
+
     private void assertEvents(List<Map<String, Object>> events) {
         assertNotNull(events);
         logger.debug("Events {}", new Object() {

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/DataLoader.java

@@ -52,6 +52,7 @@
  */
 public class DataLoader {
     public static final String TEST_INDEX = "endgame-140";
+    public static final String TEST_SHARD_FAILURES_INDEX = "endgame-shard-failures";
     public static final String TEST_EXTRA_INDEX = "extra";
     public static final String TEST_NANOS_INDEX = "endgame-140-nanos";
     public static final String TEST_SAMPLE = "sample1,sample2,sample3";
@@ -103,6 +104,11 @@ public static void loadDatasetIntoEs(RestClient client, CheckedBiFunction<XConte
         //
         load(client, TEST_MISSING_EVENTS_INDEX, null, null, p);
         load(client, TEST_SAMPLE_MULTI, null, null, p);
+        //
+        // index with a runtime field ("broken", type long) that causes shard failures.
+        // the rest of the mapping is the same as TEST_INDEX
+        //
+        load(client, TEST_SHARD_FAILURES_INDEX, null, DataLoader::timestampToUnixMillis, p);
     }
 
     private static void load(

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/EqlDateNanosSpecTestCase.java

@@ -27,9 +27,23 @@ public EqlDateNanosSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        this(TEST_NANOS_INDEX, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        this(
+            TEST_NANOS_INDEX,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     // constructor for multi-cluster tests
@@ -40,9 +54,23 @@ public EqlDateNanosSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        super(index, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        super(
+            index,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     @Override

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/EqlExtraSpecTestCase.java

@@ -27,9 +27,23 @@ public EqlExtraSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        this(TEST_EXTRA_INDEX, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        this(
+            TEST_EXTRA_INDEX,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     // constructor for multi-cluster tests
@@ -40,9 +54,23 @@ public EqlExtraSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        super(index, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        super(
+            index,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     @Override

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/EqlMissingEventsSpecTestCase.java

@@ -27,9 +27,23 @@ public EqlMissingEventsSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        this(TEST_MISSING_EVENTS_INDEX, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        this(
+            TEST_MISSING_EVENTS_INDEX,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     // constructor for multi-cluster tests
@@ -40,9 +54,23 @@ public EqlMissingEventsSpecTestCase(
         List<long[]> eventIds,
         String[] joinKeys,
         Integer size,
-        Integer maxSamplesPerKey
+        Integer maxSamplesPerKey,
+        Boolean allowPartialSearchResults,
+        Boolean allowPartialSequenceResults,
+        Boolean expectShardFailures
     ) {
-        super(index, query, name, eventIds, joinKeys, size, maxSamplesPerKey);
+        super(
+            index,
+            query,
+            name,
+            eventIds,
+            joinKeys,
+            size,
+            maxSamplesPerKey,
+            allowPartialSearchResults,
+            allowPartialSequenceResults,
+            expectShardFailures
+        );
     }
 
     @Override