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,26 @@ 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.
+====
+
 `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

@@ -190,6 +190,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion LOGSDB_TELEMETRY_STATS = def(8_785_00_0);
     public static final TransportVersion KQL_QUERY_ADDED = def(8_786_00_0);
     public static final TransportVersion ROLE_MONITOR_STATS = def(8_787_00_0);
+    public static final TransportVersion EQL_ALLOW_PARTIAL_SEARCH_RESULTS = def(8_788_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -163,13 +163,19 @@ protected ObjectPath runQuery(String index, String query) throws Exception {
         if (maxSamplesPerKey > 0) {
             builder.field("max_samples_per_key", maxSamplesPerKey);
         }
+        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 (randomBoolean()) {
+            request.addParameter("allow_partial_search_results", String.valueOf(randomBoolean()));
+        }
         int timeout = Math.toIntExact(timeout().millis());
         RequestConfig config = RequestConfig.copy(RequestConfig.DEFAULT)
             .setConnectionRequestTimeout(timeout)

x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/10_basic.yml

@@ -478,3 +478,34 @@ setup:
           query: 'sequence with maxspan=10d [network where user == "ADMIN"] ![network where used == "SYSTEM"]'
   - match: { error.root_cause.0.type: "verification_exception" }
   - match: { error.root_cause.0.reason: "Found 1 problem\nline 1:75: Unknown column [used], did you mean [user]?" }
+
+
+---
+"Execute query with allow_partial_search_results":
+  - do:
+      eql.search:
+        index: eql_test
+        body:
+          query: 'process where user == "SYSTEM"'
+          fields: [{"field":"@timestamp","format":"epoch_millis"},"id","valid","day_of_week"]
+          allow_partial_search_results: true
+
+  - match: {timed_out: false}
+  - match: {hits.total.value: 3}
+  - match: {hits.total.relation: "eq"}
+  - match: {hits.events.0._source.user: "SYSTEM"}
+  - match: {hits.events.0._id: "1"}
+  - match: {hits.events.0.fields.@timestamp: ["1580733296000"]}
+  - match: {hits.events.0.fields.id: [123]}
+  - match: {hits.events.0.fields.valid: [false]}
+  - match: {hits.events.0.fields.day_of_week: ["Monday"]}
+  - match: {hits.events.1._id: "2"}
+  - match: {hits.events.1.fields.@timestamp: ["1580819696000"]}
+  - match: {hits.events.1.fields.id: [123]}
+  - match: {hits.events.1.fields.valid: [true]}
+  - match: {hits.events.1.fields.day_of_week: ["Tuesday"]}
+  - match: {hits.events.2._id: "3"}
+  - match: {hits.events.2.fields.@timestamp: ["1580906096000"]}
+  - match: {hits.events.2.fields.id: [123]}
+  - match: {hits.events.2.fields.valid: [true]}
+  - match: {hits.events.2.fields.day_of_week: ["Wednesday"]}

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

@@ -63,6 +63,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     private List<FieldAndFormat> fetchFields;
     private Map<String, Object> runtimeMappings = emptyMap();
     private int maxSamplesPerKey = RequestDefaults.MAX_SAMPLES_PER_KEY;
+    private Boolean allowPartialSearchResults;
 
     // Async settings
     private TimeValue waitForCompletionTimeout = null;
@@ -83,6 +84,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final String KEY_FETCH_FIELDS = "fields";
     static final String KEY_RUNTIME_MAPPINGS = "runtime_mappings";
     static final String KEY_MAX_SAMPLES_PER_KEY = "max_samples_per_key";
+    static final String KEY_ALLOW_PARTIAL_SEARCH_RESULTS = "allow_partial_search_results";
 
     static final ParseField FILTER = new ParseField(KEY_FILTER);
     static final ParseField TIMESTAMP_FIELD = new ParseField(KEY_TIMESTAMP_FIELD);
@@ -97,6 +99,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final ParseField RESULT_POSITION = new ParseField(KEY_RESULT_POSITION);
     static final ParseField FETCH_FIELDS_FIELD = SearchSourceBuilder.FETCH_FIELDS_FIELD;
     static final ParseField MAX_SAMPLES_PER_KEY = new ParseField(KEY_MAX_SAMPLES_PER_KEY);
+    static final ParseField ALLOW_PARTIAL_SEARCH_RESULTS = new ParseField(KEY_ALLOW_PARTIAL_SEARCH_RESULTS);
 
     private static final ObjectParser<EqlSearchRequest, Void> PARSER = objectParser(EqlSearchRequest::new);
 
@@ -135,6 +138,11 @@ public EqlSearchRequest(StreamInput in) throws IOException {
         if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_7_0)) {
             maxSamplesPerKey = in.readInt();
         }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.EQL_ALLOW_PARTIAL_SEARCH_RESULTS)) {
+            allowPartialSearchResults = in.readOptionalBoolean();
+        } else {
+            allowPartialSearchResults = false;
+        }
     }
 
     @Override
@@ -245,6 +253,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(KEY_RUNTIME_MAPPINGS, runtimeMappings);
         }
         builder.field(KEY_MAX_SAMPLES_PER_KEY, maxSamplesPerKey);
+        builder.field(KEY_ALLOW_PARTIAL_SEARCH_RESULTS, allowPartialSearchResults);
 
         return builder;
     }
@@ -279,6 +288,7 @@ protected static <R extends EqlSearchRequest> ObjectParser<R, Void> objectParser
         parser.declareField(EqlSearchRequest::fetchFields, EqlSearchRequest::parseFetchFields, FETCH_FIELDS_FIELD, ValueType.VALUE_ARRAY);
         parser.declareObject(EqlSearchRequest::runtimeMappings, (p, c) -> p.map(), SearchSourceBuilder.RUNTIME_MAPPINGS_FIELD);
         parser.declareInt(EqlSearchRequest::maxSamplesPerKey, MAX_SAMPLES_PER_KEY);
+        parser.declareBoolean(EqlSearchRequest::allowPartialSearchResults, ALLOW_PARTIAL_SEARCH_RESULTS);
         return parser;
     }
 
@@ -427,6 +437,15 @@ public EqlSearchRequest maxSamplesPerKey(int maxSamplesPerKey) {
         return this;
     }
 
+    public Boolean allowPartialSearchResults() {
+        return allowPartialSearchResults;
+    }
+
+    public EqlSearchRequest allowPartialSearchResults(Boolean val) {
+        this.allowPartialSearchResults = val;
+        return this;
+    }
+
     private static List<FieldAndFormat> parseFetchFields(XContentParser parser) throws IOException {
         List<FieldAndFormat> result = new ArrayList<>();
         Token token = parser.currentToken();
@@ -470,6 +489,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_7_0)) {
             out.writeInt(maxSamplesPerKey);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.EQL_ALLOW_PARTIAL_SEARCH_RESULTS)) {
+            out.writeOptionalBoolean(allowPartialSearchResults);
+        }
     }
 
     @Override
@@ -496,7 +518,8 @@ public boolean equals(Object o) {
             && Objects.equals(resultPosition, that.resultPosition)
             && Objects.equals(fetchFields, that.fetchFields)
             && Objects.equals(runtimeMappings, that.runtimeMappings)
-            && Objects.equals(maxSamplesPerKey, that.maxSamplesPerKey);
+            && Objects.equals(maxSamplesPerKey, that.maxSamplesPerKey)
+            && Objects.equals(allowPartialSearchResults, that.allowPartialSearchResults);
     }
 
     @Override
@@ -517,7 +540,8 @@ public int hashCode() {
             resultPosition,
             fetchFields,
             runtimeMappings,
-            maxSamplesPerKey
+            maxSamplesPerKey,
+            allowPartialSearchResults
         );
     }
 

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

@@ -7,8 +7,11 @@
 package org.elasticsearch.xpack.eql.action;
 
 import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ShardOperationFailedException;
+import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -17,6 +20,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParserUtils;
 import org.elasticsearch.core.Nullable;
@@ -42,6 +46,7 @@
 import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.action.search.ShardSearchFailure.readShardSearchFailure;
 import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
 import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
 import static org.elasticsearch.xpack.eql.util.SearchHitUtils.qualifiedIndex;
@@ -54,6 +59,7 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
     private final String asyncExecutionId;
     private final boolean isRunning;
     private final boolean isPartial;
+    private final ShardSearchFailure[] shardFailures;
 
     private static final class Fields {
         static final String TOOK = "took";
@@ -62,6 +68,7 @@ private static final class Fields {
         static final String ID = "id";
         static final String IS_RUNNING = "is_running";
         static final String IS_PARTIAL = "is_partial";
+        static final String SHARD_FAILURES = "shard_failures";
     }
 
     private static final ParseField TOOK = new ParseField(Fields.TOOK);
@@ -70,8 +77,10 @@ private static final class Fields {
     private static final ParseField ID = new ParseField(Fields.ID);
     private static final ParseField IS_RUNNING = new ParseField(Fields.IS_RUNNING);
     private static final ParseField IS_PARTIAL = new ParseField(Fields.IS_PARTIAL);
+    private static final ParseField SHARD_FAILURES = new ParseField(Fields.SHARD_FAILURES);
 
     private static final InstantiatingObjectParser<EqlSearchResponse, Void> PARSER;
+
     static {
         InstantiatingObjectParser.Builder<EqlSearchResponse, Void> parser = InstantiatingObjectParser.builder(
             "eql/search_response",
@@ -84,11 +93,12 @@ private static final class Fields {
         parser.declareString(optionalConstructorArg(), ID);
         parser.declareBoolean(constructorArg(), IS_RUNNING);
         parser.declareBoolean(constructorArg(), IS_PARTIAL);
+        parser.declareObjectArray(optionalConstructorArg(), (p, c) -> ShardSearchFailure.EMPTY_ARRAY, SHARD_FAILURES); // TODO fix this
         PARSER = parser.build();
     }
 
-    public EqlSearchResponse(Hits hits, long tookInMillis, boolean isTimeout) {
-        this(hits, tookInMillis, isTimeout, null, false, false);
+    public EqlSearchResponse(Hits hits, long tookInMillis, boolean isTimeout, ShardSearchFailure[] shardFailures) {
+        this(hits, tookInMillis, isTimeout, null, false, false, shardFailures);
     }
 
     public EqlSearchResponse(
@@ -97,7 +107,8 @@ public EqlSearchResponse(
         boolean isTimeout,
         String asyncExecutionId,
         boolean isRunning,
-        boolean isPartial
+        boolean isPartial,
+        ShardSearchFailure[] shardFailures
     ) {
         super();
         this.hits = hits == null ? Hits.EMPTY : hits;
@@ -106,6 +117,7 @@ public EqlSearchResponse(
         this.asyncExecutionId = asyncExecutionId;
         this.isRunning = isRunning;
         this.isPartial = isPartial;
+        this.shardFailures = shardFailures;
     }
 
     public EqlSearchResponse(StreamInput in) throws IOException {
@@ -116,6 +128,19 @@ public EqlSearchResponse(StreamInput in) throws IOException {
         asyncExecutionId = in.readOptionalString();
         isPartial = in.readBoolean();
         isRunning = in.readBoolean();
+        if (in.getTransportVersion().onOrAfter(TransportVersions.EQL_ALLOW_PARTIAL_SEARCH_RESULTS)) {
+            int size = in.readVInt();
+            if (size == 0) {
+                shardFailures = ShardSearchFailure.EMPTY_ARRAY;
+            } else {
+                shardFailures = new ShardSearchFailure[size];
+                for (int i = 0; i < shardFailures.length; i++) {
+                    shardFailures[i] = readShardSearchFailure(in);
+                }
+            }
+        } else {
+            shardFailures = ShardSearchFailure.EMPTY_ARRAY;
+        }
     }
 
     public static EqlSearchResponse fromXContent(XContentParser parser) {
@@ -130,6 +155,12 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalString(asyncExecutionId);
         out.writeBoolean(isPartial);
         out.writeBoolean(isRunning);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.EQL_ALLOW_PARTIAL_SEARCH_RESULTS)) {
+            out.writeVInt(shardFailures.length);
+            for (ShardSearchFailure shardSearchFailure : shardFailures) {
+                shardSearchFailure.writeTo(out);
+            }
+        }
     }
 
     @Override
@@ -147,6 +178,13 @@ private XContentBuilder innerToXContent(XContentBuilder builder, Params params)
         builder.field(IS_RUNNING.getPreferredName(), isRunning);
         builder.field(TOOK.getPreferredName(), tookInMillis);
         builder.field(TIMED_OUT.getPreferredName(), isTimeout);
+        if (CollectionUtils.isEmpty(shardFailures) == false) {
+            builder.startArray(SHARD_FAILURES.getPreferredName());
+            for (ShardOperationFailedException shardFailure : ExceptionsHelper.groupBy(shardFailures)) {
+                shardFailure.toXContent(builder, params);
+            }
+            builder.endArray();
+        }
         hits.toXContent(builder, params);
         return builder;
     }
@@ -178,6 +216,10 @@ public boolean isPartial() {
         return isPartial;
     }
 
+    public ShardSearchFailure[] shardFailures() {
+        return shardFailures;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -190,12 +232,13 @@ public boolean equals(Object o) {
         return Objects.equals(hits, that.hits)
             && Objects.equals(tookInMillis, that.tookInMillis)
             && Objects.equals(isTimeout, that.isTimeout)
-            && Objects.equals(asyncExecutionId, that.asyncExecutionId);
+            && Objects.equals(asyncExecutionId, that.asyncExecutionId)
+            && Objects.equals(shardFailures, that.shardFailures);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(hits, tookInMillis, isTimeout, asyncExecutionId);
+        return Objects.hash(hits, tookInMillis, isTimeout, asyncExecutionId, shardFailures);
     }
 
     @Override

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.eql.action;
 
+import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xpack.core.async.AsyncExecutionId;
@@ -39,7 +40,8 @@ public EqlSearchResponse getCurrentResult() {
             false,
             getExecutionId().getEncoded(),
             true,
-            true
+            true,
+            ShardSearchFailure.EMPTY_ARRAY
         );
     }
 }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/assembler/ExecutionManager.java

@@ -167,7 +167,8 @@ public Executable assemble(
             criteria.subList(0, completionStage),
             criteria.get(completionStage),
             matcher,
-            listOfKeys
+            listOfKeys,
+            cfg.allowPartialSearchResults()
         );
 
         return w;
@@ -235,7 +236,8 @@ public Executable assemble(List<List<Attribute>> listOfKeys, List<PhysicalPlan>
             cfg.fetchSize(),
             limit,
             session.circuitBreaker(),
-            cfg.maxSamplesPerKey()
+            cfg.maxSamplesPerKey(),
+            cfg.allowPartialSearchResults()
         );
     }