EQL: Add Head/Tail pipe support (#58536)

Introduce pipe support, in particular head and tail
(which can also be chained).

Author: costin

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

@@ -553,7 +553,12 @@ the events in ascending, lexicographic order.
         },
         "sort": [
           1607252647000
-        ]
+        ],
+        "fields": {
+          "@timestamp": [
+            "1607252647000"
+          ]
+        }
       },
       {
         "_index": "my_index",
@@ -583,7 +588,12 @@ the events in ascending, lexicographic order.
         },
         "sort": [
           1607339228000
-        ]
+        ],
+        "fields": {
+          "@timestamp": [
+            "1607339228000"
+          ]
+        }
       }
     ]
   }

docs/reference/eql/search.asciidoc

@@ -101,7 +101,12 @@ https://en.wikipedia.org/wiki/Unix_time[Unix epoch], in ascending order.
         },
         "sort": [
           1607252645000
-        ]
+        ],
+        "fields": {
+          "@timestamp": [
+            "1607252645000"
+          ]
+        }
       },
       {
         "_index": "sec_logs",
@@ -124,7 +129,12 @@ https://en.wikipedia.org/wiki/Unix_time[Unix epoch], in ascending order.
         },
         "sort": [
           1607339167000
-        ]
+        ],
+        "fields": {
+          "@timestamp": [
+            "1607339167000"
+          ]
+        }
       }
     ]
   }
@@ -169,6 +179,7 @@ GET /sec_logs/_eql/search
   """
 }
 ----
+// TEST[s/search/search\?filter_path\=\-\*\.sequences\.events\.\*fields/]
 
 The API returns the following response. Matching events in
 the `hits.sequences.events` property are sorted by
@@ -216,11 +227,6 @@ the https://en.wikipedia.org/wiki/Unix_time[Unix epoch], in ascending order.
                 "path": "C:\\Windows\\System32\\cmd.exe"
               }
             },
-            "fields": {
-              "@timestamp": [
-                "1607339228000"
-              ]
-            },
             "sort": [
               1607339228000
             ]
@@ -244,11 +250,6 @@ the https://en.wikipedia.org/wiki/Unix_time[Unix epoch], in ascending order.
                 "path": "C:\\Windows\\System32\\regsvr32.exe"
               }
             },
-            "fields": {
-              "@timestamp": [
-                "1607339229000"
-              ]
-            },
             "sort": [
               1607339229000
             ]
@@ -293,6 +294,7 @@ GET /sec_logs/_eql/search
   """
 }
 ----
+// TEST[s/search/search\?filter_path\=\-\*\.sequences\.events\.\*fields/]
 
 The API returns the following response. The `hits.sequences.join_keys` property
 contains the shared `agent.id` value for each matching event.
@@ -341,11 +343,6 @@ contains the shared `agent.id` value for each matching event.
                 "path": "C:\\Windows\\System32\\cmd.exe"
               }
             },
-            "fields": {
-              "@timestamp": [
-                "1607339228000"
-              ]
-            },
             "sort": [
               1607339228000
             ]
@@ -369,11 +366,6 @@ contains the shared `agent.id` value for each matching event.
                 "path": "C:\\Windows\\System32\\regsvr32.exe"
               }
             },
-            "fields": {
-              "@timestamp": [
-                "1607339229000"
-              ]
-            },
             "sort": [
               1607339229000
             ]

x-pack/plugin/eql/qa/common/src/main/resources/test_queries_supported.toml

@@ -1,7 +1,6 @@
 # This file is populated with additional EQL queries that were not present in the original EQL python implementation
 # test_queries.toml file in order to keep the original unchanges and easier to sync with the EQL reference implementation tests.
 
-
 [[queries]]
 expected_event_ids = [95]
 query = '''

x-pack/plugin/eql/qa/common/src/main/resources/test_queries_unsupported.toml

@@ -12,6 +12,9 @@
 # query = 'process where serial_event_id = 1'
 # expected_event_ids  = [1]
 
+[[queries]]
+expected_event_ids = []
+query = 'process where missing_field != null'
 
 # fails because of string check - msbuild does not match MSBuild
 [[queries]]
@@ -20,19 +23,10 @@ sequence by unique_pid [process where opcode=1 and process_name == 'msbuild.exe'
 expected_event_ids  = [75273, 75304]
 description = "test that process sequences are working correctly"
 
-
-[[queries]]
-query = 'process where true | head 6'
-expected_event_ids  = [1, 2, 3, 4, 5, 6]
-
 [[queries]]
 expected_event_ids = []
 query = 'process where missing_field != null'
 
-[[queries]]
-expected_event_ids  = [1, 2, 3, 4, 5]
-query = 'process where bad_field == null | head 5'
-
 [[queries]]
 tags = ["comparisons", "pipes"]
 query = '''
@@ -65,25 +59,6 @@ process where true
 '''
 expected_event_ids  = [9, 10]
 
-[[queries]]
-note = "check that comparisons against null values return false"
-expected_event_ids = []
-query = '''
-process where not (exit_code > -1)
-  and serial_event_id in (58, 64, 69, 74, 80, 85, 90, 93, 94)
-| head 10
-'''
-
-[[queries]]
-note = "check that comparisons against null values return false"
-expected_event_ids  = [1, 2, 3, 4, 5, 6, 7]
-query = 'process where not (exit_code > -1) | head 7'
-
-[[queries]]
-note = "check that comparisons against null values return false"
-expected_event_ids  = [1, 2, 3, 4, 5, 6, 7]
-query = 'process where not (-1 < exit_code) | head 7'
-
 [[queries]]
 query = 'process where (serial_event_id<9 and serial_event_id >= 7) or (opcode == pid)'
 expected_event_ids  = [7, 8]
@@ -182,12 +157,6 @@ process where opcode=1 and process_name == "smss.exe"
 '''
 expected_event_ids  = [78]
 
-[[queries]]
-query = '''
-file where true
-| tail 3'''
-expected_event_ids  = [92, 95, 96]
-
 [[queries]]
 query = '''
 process where opcode in (1,3) and process_name in (parent_process_name, "SYSTEM")
@@ -256,17 +225,6 @@ sequence
 expected_event_ids  = [1, 2, 2, 3]
 
 
-[[queries]]
-query = '''
-sequence
-  [file where event_subtype_full == "file_create_event"] by file_path
-  [process where opcode == 1] by process_path
-  [process where opcode == 2] by process_path
-  [file where event_subtype_full == "file_delete_event"] by file_path
-| head 4
-| tail 2'''
-expected_event_ids  = [67, 68, 69, 70, 72, 73, 74, 75]
-
 [[queries]]
 query = '''
 sequence with maxspan=1d
@@ -322,14 +280,6 @@ sequence with maxspan=0.5s
 | tail 2'''
 expected_event_ids = []
 
-[[queries]]
-query = '''
-sequence
-  [file where opcode=0] by unique_pid
-  [file where opcode=0] by unique_pid
-| head 1'''
-expected_event_ids  = [55, 61]
-
 [[queries]]
 query = '''
 sequence
@@ -466,6 +416,59 @@ query = '''
 registry where length(bad_field) > 0
 '''
 
+[[queries]]
+expected_event_ids  = [1, 2, 3, 4, 5]
+query = 'process where bad_field == null | head 5'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where exit_code >= 0'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where 0 <= exit_code'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where exit_code <= 0'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where exit_code < 1'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where exit_code > -1'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [58, 64, 69, 74, 80, 85, 90, 93, 94, 75303]
+query = 'process where -1 < exit_code'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids = []
+query = '''
+process where not (exit_code > -1)
+  and serial_event_id in (58, 64, 69, 74, 80, 85, 90, 93, 94)
+| head 10
+'''
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [1, 2, 3, 4, 5, 6, 7]
+query = 'process where not (exit_code > -1) | head 7'
+
+[[queries]]
+note = "check that comparisons against null values return false"
+expected_event_ids  = [1, 2, 3, 4, 5, 6, 7]
+query = 'process where not (-1 < exit_code) | head 7'
+
 [[queries]]
 expected_event_ids  = [1, 55, 57, 63, 75304]
 query = '''
@@ -539,16 +542,6 @@ expected_event_ids  = [55, 95]
 query = 'process where event of [process where process_name = "python.exe" ]'
 expected_event_ids  = [48, 50, 51, 54, 93]
 
-[[queries]]
-query = '''
-sequence by user_name
-  [file where opcode=0] by file_path
-  [process where opcode=1] by process_path
-  [process where opcode=2] by process_path
-  [file where opcode=2] by file_path
-| tail 1'''
-expected_event_ids  = [88, 89, 90, 91]
-
 [[queries]]
 query = '''
 sequence by user_name
@@ -567,16 +560,6 @@ until [process where opcode=5] by ppid,process_path
 | head 2'''
 expected_event_ids  = [55, 59, 61, 65]
 
-[[queries]]
-query = '''
-sequence by pid
-  [file where opcode=0] by file_path
-  [process where opcode=1] by process_path
-  [process where opcode=2] by process_path
-  [file where opcode=2] by file_path
-| tail 1'''
-expected_event_ids = []
-
 [[queries]]
 query = '''
 join by user_name

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

@@ -9,25 +9,35 @@
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.eql.EqlIllegalArgumentException;
+import org.elasticsearch.xpack.eql.execution.search.QueryRequest;
 import org.elasticsearch.xpack.ql.execution.search.extractor.HitExtractor;
 
 import java.util.List;
 
-public class Criterion {
+public class Criterion implements QueryRequest {
 
     private final SearchSourceBuilder searchSource;
     private final List<HitExtractor> keyExtractors;
     private final HitExtractor timestampExtractor;
     private final HitExtractor tiebreakerExtractor;
 
+    // search after markers
+    private Object[] startMarker;
+    private Object[] stopMarker;
+
+    //TODO: should accept QueryRequest instead of another SearchSourceBuilder
     public Criterion(SearchSourceBuilder searchSource, List<HitExtractor> searchAfterExractors, HitExtractor timestampExtractor,
                      HitExtractor tiebreakerExtractor) {
         this.searchSource = searchSource;
         this.keyExtractors = searchAfterExractors;
         this.timestampExtractor = timestampExtractor;
         this.tiebreakerExtractor = tiebreakerExtractor;
+
+        this.startMarker = null;
+        this.stopMarker = null;
     }
 
+    @Override
     public SearchSourceBuilder searchSource() {
         return searchSource;
     }
@@ -64,8 +74,34 @@ public Comparable<Object> tiebreaker(SearchHit hit) {
         throw new EqlIllegalArgumentException("Expected tiebreaker to be Comparable but got {}", tb);
     }
 
-    public void fromMarkers(Object[] markers) {
-        // TODO: this is likely to be rewritten afterwards
-        searchSource.searchAfter(markers);
+    public Object[] startMarker() {
+        return startMarker;
+    }
+
+    public Object[] stopMarker() {
+        return stopMarker;
+    }
+
+    private Object[] marker(SearchHit hit) {
+        long timestamp = timestamp(hit);
+        Object tiebreaker = null;
+        if (tiebreakerExtractor() != null) {
+            tiebreaker = tiebreaker(hit);
+        }
+
+        return tiebreaker != null ? new Object[] { timestamp, tiebreaker } : new Object[] { timestamp };
+    }
+
+    public void startMarker(SearchHit hit) {
+        startMarker = marker(hit);
+    }
+
+    public void stopMarker(SearchHit hit) {
+        stopMarker = marker(hit);
+    }
+
+    public Criterion useMarker(Object[] marker) {
+        searchSource.searchAfter(marker);
+        return this;
     }
 }