EQL: fix async missing events and re-enable the feature (elastic#97718)

luigidellaquila · web-flow · commit 9ab8f71b01bb · 2023-07-19T16:01:39.000+02:00
diff --git a/docs/changelog/97718.yaml b/docs/changelog/97718.yaml
@@ -0,0 +1,6 @@
+pr: 97718
+summary: Fix async missing events
+area: EQL
+type: bug
+issues:
+ - 97644
diff --git a/server/src/main/java/org/elasticsearch/TransportVersion.java b/server/src/main/java/org/elasticsearch/TransportVersion.java
@@ -159,9 +159,10 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
     // Introduced for stateless plugin
     public static final TransportVersion V_8_500_036 = registerTransportVersion(8_500_036, "3343c64f-d7ac-4f02-9262-3e1acfc56f89");
     public static final TransportVersion V_8_500_037 = registerTransportVersion(8_500_037, "d76a4f22-8878-43e0-acfa-15e452195fa7");
+    public static final TransportVersion V_8_500_038 = registerTransportVersion(8_500_038, "9ef93580-feae-409f-9989-b49e411ca7a9");
 
     private static class CurrentHolder {
-        private static final TransportVersion CURRENT = findCurrent(V_8_500_037);
+        private static final TransportVersion CURRENT = findCurrent(V_8_500_038);
 
         // finds the pluggable current version, or uses the given fallback
         private static TransportVersion findCurrent(TransportVersion fallback) {
diff --git a/x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/BaseEqlSpecTestCase.java b/x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/BaseEqlSpecTestCase.java
@@ -244,10 +244,10 @@ private long[] extractIds(List<Map<String, Object>> events) {
         final long[] ids = new long[len];
         for (int i = 0; i < len; i++) {
             Map<String, Object> event = events.get(i);
-            Map<String, Object> source = (Map<String, Object>) event.get("_source");
-            if (source == null) {
+            if (Boolean.TRUE.equals(event.get("missing"))) {
                 ids[i] = -1;
             } else {
+                Map<String, Object> source = (Map<String, Object>) event.get("_source");
                 Object field = source.get(idField());
                 ids[i] = ((Number) field).longValue();
             }
diff --git a/x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/DataLoader.java b/x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/DataLoader.java
@@ -107,7 +107,7 @@ public static void loadDatasetIntoEs(
         //
         // missing_events index
         //
-        // load(client, TEST_MISSING_EVENTS_INDEX, null, null, p);
+        load(client, TEST_MISSING_EVENTS_INDEX, null, null, p);
         load(client, TEST_SAMPLE_MULTI, null, null, p);
     }
 
diff --git a/x-pack/plugin/eql/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/eql/EqlMissingEventsIT.java b/x-pack/plugin/eql/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/eql/EqlMissingEventsIT.java
@@ -7,12 +7,10 @@
 
 package org.elasticsearch.xpack.eql;
 
-import org.apache.lucene.tests.util.LuceneTestCase;
 import org.elasticsearch.test.eql.EqlMissingEventsSpecTestCase;
 
 import java.util.List;
 
-@LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/97644")
 public class EqlMissingEventsIT extends EqlMissingEventsSpecTestCase {
 
     public EqlMissingEventsIT(String query, String name, List<long[]> eventIds, String[] joinKeys, Integer size, Integer maxSamplesPerKey) {
diff --git a/x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/10_basic.yml b/x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/10_basic.yml
@@ -436,9 +436,7 @@ setup:
 
 ---
 "Sequence with missing events.":
-  - skip:
-      version: "all"
-      reason: "AwaitsFix https://github.com/elastic/elasticsearch/issues/97644"
+
   - do:
       eql.search:
         index: eql_test
diff --git a/x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/30_async_missing_events.yml b/x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/30_async_missing_events.yml
@@ -0,0 +1,61 @@
+---
+setup:
+  - do:
+      indices.create:
+          index:  eql_test
+          body:
+            mappings:
+              properties:
+                "@timestamp":
+                  type: date
+                event.category:
+                  type: keyword
+                user:
+                  type: keyword
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: eql_test
+              _id:    "1"
+          - event:
+              - category: process
+            "@timestamp": 2023-07-11T11:09:05.529Z
+            user: foo
+          - index:
+              _index: eql_test
+              _id:    "2"
+          - event:
+              - category: process
+            "@timestamp": 2023-07-11T11:09:06.529Z
+            user: bar
+
+---
+
+"Execute async EQL with missing events":
+  - do:
+      eql.search:
+        index: eql_test
+        wait_for_completion_timeout: "0ms"
+        keep_on_completion: true
+        body:
+          query: 'sequence with maxspan=24h [ process where true ] ![ process where true ]'
+
+  - is_true: id
+  - set: {id: id}
+  - gte: {took: 0}
+
+  - do:
+      eql.get:
+        id: $id
+        wait_for_completion_timeout: "10s"
+
+  - match: {is_running: false}
+  - match: {is_partial: false}
+  - match: {timed_out: false}
+  - match: {hits.total.value: 1}
+  - match: {hits.total.relation: "eq"}
+  - match: {hits.sequences.0.events.0._source.user: "bar"}
+  - match: {hits.sequences.0.events.1.missing: true}
diff --git a/x-pack/plugin/eql/src/main/antlr/EqlBase.g4 b/x-pack/plugin/eql/src/main/antlr/EqlBase.g4
@@ -66,7 +66,7 @@ sequenceTerm
    ;
 
 subquery
-    : LB eventFilter RB
+    : (LB | MISSING_EVENT_OPEN) eventFilter RB
     ;
 
 eventQuery
@@ -212,6 +212,7 @@ LP: '(';
 RP: ')';
 PIPE: '|';
 OPTIONAL: '?';
+MISSING_EVENT_OPEN: '![';
 
 fragment STRING_ESCAPE
     : '\\' [btnfr"'\\]
diff --git a/x-pack/plugin/eql/src/main/antlr/EqlBase.tokens b/x-pack/plugin/eql/src/main/antlr/EqlBase.tokens
@@ -41,15 +41,16 @@ LP=40
 RP=41
 PIPE=42
 OPTIONAL=43
-STRING=44
-INTEGER_VALUE=45
-DECIMAL_VALUE=46
-IDENTIFIER=47
-QUOTED_IDENTIFIER=48
-TILDE_IDENTIFIER=49
-LINE_COMMENT=50
-BRACKETED_COMMENT=51
-WS=52
+MISSING_EVENT_OPEN=44
+STRING=45
+INTEGER_VALUE=46
+DECIMAL_VALUE=47
+IDENTIFIER=48
+QUOTED_IDENTIFIER=49
+TILDE_IDENTIFIER=50
+LINE_COMMENT=51
+BRACKETED_COMMENT=52
+WS=53
 'and'=1
 'any'=2
 'by'=3
@@ -93,3 +94,4 @@ WS=52
 ')'=41
 '|'=42
 '?'=43
+'!['=44
diff --git a/x-pack/plugin/eql/src/main/antlr/EqlBaseLexer.tokens b/x-pack/plugin/eql/src/main/antlr/EqlBaseLexer.tokens
@@ -41,15 +41,16 @@ LP=40
 RP=41
 PIPE=42
 OPTIONAL=43
-STRING=44
-INTEGER_VALUE=45
-DECIMAL_VALUE=46
-IDENTIFIER=47
-QUOTED_IDENTIFIER=48
-TILDE_IDENTIFIER=49
-LINE_COMMENT=50
-BRACKETED_COMMENT=51
-WS=52
+MISSING_EVENT_OPEN=44
+STRING=45
+INTEGER_VALUE=46
+DECIMAL_VALUE=47
+IDENTIFIER=48
+QUOTED_IDENTIFIER=49
+TILDE_IDENTIFIER=50
+LINE_COMMENT=51
+BRACKETED_COMMENT=52
+WS=53
 'and'=1
 'any'=2
 'by'=3
@@ -93,3 +94,4 @@ WS=52
 ')'=41
 '|'=42
 '?'=43
+'!['=44
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchResponse.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchResponse.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -33,6 +34,7 @@
 import org.elasticsearch.xpack.ql.async.QlStatusResponse;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -202,23 +204,33 @@ public String toString() {
     // Event
     public static class Event implements Writeable, ToXContentObject {
 
+        public static Event MISSING_EVENT = new Event("", "", new BytesArray("{}".getBytes(StandardCharsets.UTF_8)), null, true);
+
         private static final class Fields {
             static final String INDEX = GetResult._INDEX;
             static final String ID = GetResult._ID;
             static final String SOURCE = SourceFieldMapper.NAME;
             static final String FIELDS = "fields";
+            static final String MISSING = "missing";
         }
 
         private static final ParseField INDEX = new ParseField(Fields.INDEX);
         private static final ParseField ID = new ParseField(Fields.ID);
         private static final ParseField SOURCE = new ParseField(Fields.SOURCE);
         private static final ParseField FIELDS = new ParseField(Fields.FIELDS);
+        private static final ParseField MISSING = new ParseField(Fields.MISSING);
 
         @SuppressWarnings("unchecked")
         private static final ConstructingObjectParser<Event, Void> PARSER = new ConstructingObjectParser<>(
             "eql/search_response_event",
             true,
-            args -> new Event((String) args[0], (String) args[1], (BytesReference) args[2], (Map<String, DocumentField>) args[3])
+            args -> new Event(
+                (String) args[0],
+                (String) args[1],
+                (BytesReference) args[2],
+                (Map<String, DocumentField>) args[3],
+                (Boolean) args[4]
+            )
         );
 
         static {
@@ -238,21 +250,29 @@ private static final class Fields {
                 }
                 return fields;
             }, FIELDS);
+            PARSER.declareBoolean(optionalConstructorArg(), MISSING);
         }
 
         private String index;
         private final String id;
         private final BytesReference source;
         private final Map<String, DocumentField> fetchFields;
 
+        private final boolean missing;
+
         public Event(String index, String id, BytesReference source, Map<String, DocumentField> fetchFields) {
+            this(index, id, source, fetchFields, false);
+        }
+
+        public Event(String index, String id, BytesReference source, Map<String, DocumentField> fetchFields, Boolean missing) {
             this.index = index;
             this.id = id;
             this.source = source;
             this.fetchFields = fetchFields;
+            this.missing = missing != null && missing;
         }
 
-        public Event(StreamInput in) throws IOException {
+        private Event(StreamInput in) throws IOException {
             index = in.readString();
             id = in.readString();
             source = in.readBytesReference();
@@ -261,6 +281,16 @@ public Event(StreamInput in) throws IOException {
             } else {
                 fetchFields = null;
             }
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_038)) {
+                missing = in.readBoolean();
+            } else {
+                missing = false;
+            }
+        }
+
+        public static Event readFrom(StreamInput in) throws IOException {
+            Event result = new Event(in);
+            return result.missing() ? MISSING_EVENT : result;
         }
 
         @Override
@@ -274,6 +304,9 @@ public void writeTo(StreamOutput out) throws IOException {
                     out.writeMap(fetchFields, StreamOutput::writeString, (stream, documentField) -> documentField.writeTo(stream));
                 }
             }
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_038)) {
+                out.writeBoolean(missing);
+            }
         }
 
         @Override
@@ -295,6 +328,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 }
                 builder.endObject();
             }
+            if (missing) {
+                // preserve original event structure (before introduction of missing events): avoid "missing: false" for normal events
+                builder.field(Fields.MISSING, missing);
+            }
             builder.endObject();
             return builder;
         }
@@ -323,9 +360,13 @@ public Map<String, DocumentField> fetchFields() {
             return fetchFields;
         }
 
+        public boolean missing() {
+            return missing;
+        }
+
         @Override
         public int hashCode() {
-            return Objects.hash(index, id, source, fetchFields);
+            return Objects.hash(index, id, source, fetchFields, missing);
         }
 
         @Override
@@ -342,7 +383,8 @@ public boolean equals(Object obj) {
             return Objects.equals(index, other.index)
                 && Objects.equals(id, other.id)
                 && Objects.equals(source, other.source)
-                && Objects.equals(fetchFields, other.fetchFields);
+                && Objects.equals(fetchFields, other.fetchFields)
+                && Objects.equals(missing, other.missing);
         }
 
         @Override
@@ -395,7 +437,7 @@ public Sequence(List<Object> joinKeys, List<Event> events) {
         @SuppressWarnings("unchecked")
         public Sequence(StreamInput in) throws IOException {
             this.joinKeys = (List<Object>) in.readGenericValue();
-            this.events = in.readList(Event::new);
+            this.events = in.readList(Event::readFrom);
         }
 
         public static Sequence fromXContent(XContentParser parser) {
@@ -417,13 +459,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             if (events.isEmpty() == false) {
                 builder.startArray(Fields.EVENTS);
                 for (Event event : events) {
-                    if (event == null) {
-                        builder.startObject();
-                        builder.field("missing", true);
-                        builder.endObject();
-                    } else {
-                        event.toXContent(builder, params);
-                    }
+                    event.toXContent(builder, params);
                 }
                 builder.endArray();
             }
@@ -483,7 +519,7 @@ public Hits(StreamInput in) throws IOException {
             } else {
                 totalHits = null;
             }
-            events = in.readBoolean() ? in.readList(Event::new) : null;
+            events = in.readBoolean() ? in.readList(Event::readFrom) : null;
             sequences = in.readBoolean() ? in.readList(Sequence::new) : null;
         }
 
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/sequence/SequencePayload.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/sequence/SequencePayload.java
@@ -31,7 +31,7 @@ class SequencePayload extends AbstractPayload {
             List<Event> events = new ArrayList<>(hits.size());
             for (SearchHit hit : hits) {
                 if (hit == null) {
-                    events.add(null);
+                    events.add(Event.MISSING_EVENT);
                 } else {
                     events.add(new Event(qualifiedIndex(hit), hit.getId(), hit.getSourceRef(), hit.getDocumentFields()));
                 }
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/EqlBaseLexer.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/EqlBaseLexer.java
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/EqlBaseParser.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/EqlBaseParser.java
diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/LogicalPlanBuilder.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/LogicalPlanBuilder.java
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/action/EqlSearchResponseTests.java b/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/action/EqlSearchResponseTests.java
diff --git a/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/planner/QueryTranslatorFailTests.java b/x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/planner/QueryTranslatorFailTests.java

Original file line number	Diff line number	Diff line change
`@@ -107,7 +107,7 @@ public static void loadDatasetIntoEs(`
`107`	`107`	`//`
`108`	`108`	`// missing_events index`
`109`	`109`	`//`
`110`		`- // load(client, TEST_MISSING_EVENTS_INDEX, null, null, p);`
	`110`	`+ load(client, TEST_MISSING_EVENTS_INDEX, null, null, p);`
`111`	`111`	`load(client, TEST_SAMPLE_MULTI, null, null, p);`
`112`	`112`	`}`
`113`	`113`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ sequenceTerm`
`66`	`66`	`;`
`67`	`67`
`68`	`68`	`subquery`
`69`		`- : LB eventFilter RB`
	`69`	`+ : (LB \| MISSING_EVENT_OPEN) eventFilter RB`
`70`	`70`	`;`
`71`	`71`
`72`	`72`	`eventQuery`
`@@ -212,6 +212,7 @@ LP: '(';`
`212`	`212`	`RP: ')';`
`213`	`213`	`PIPE: '\|';`
`214`	`214`	`OPTIONAL: '?';`
	`215`	`+MISSING_EVENT_OPEN: '![';`
`215`	`216`
`216`	`217`	`fragment STRING_ESCAPE`
`217`	`218`	`: '\\' [btnfr"'\\]`