EQL: Introduce repeatable queries (#75082)

Allow individual queries within a sequence to be ran multiple times
through using the [runs=number] construct as a suffix without 
having to redeclare the query.

sequence
  queryA [runs=2]
  queryB
  queryC [runs=3]
  queryD

is the same as:

sequence
  queryA
  queryA
  queryB
  queryC
  queryC
  queryC
  queryD

but more concise.

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

@@ -228,19 +228,35 @@ sequence by unique_pid
   [any where true]
   [any where serial_event_id < 72]
 '''
-expected_event_ids  = [
-                       54, 55, 59,
-                       55, 59, 61,
-                       59, 61, 65,
-                       16, 60, 66,
-                       61, 65, 67,
-                       65, 67, 70,
-                       60, 66, 71]
+expected_event_ids = [
+                      54, 55, 59,
+                      55, 59, 61,
+                      59, 61, 65,
+                      16, 60, 66,
+                      61, 65, 67,
+                      65, 67, 70,
+                      60, 66, 71]
+
+[[queries]]
+name = "sequenceWithMoreThan10Results-Runs"
+query = '''
+sequence by unique_pid
+  [any where true] [runs=2]
+  [any where serial_event_id < 72]
+'''
+expected_event_ids = [
+                      54, 55, 59,
+                      55, 59, 61,
+                      59, 61, 65,
+                      16, 60, 66,
+                      61, 65, 67,
+                      65, 67, 70,
+                      60, 66, 71]
 
 [[queries]]
 name = "seqSingleArg"
 query = 'process where string(serial_event_id) : ("1")'
-expected_event_ids  = [1]
+expected_event_ids = [1]
 
 [[queries]]
 name = "seqSingleArgPattern"

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

@@ -554,7 +554,16 @@ sequence
   [process where true]
   [process where true]
 '''
-expected_event_ids  = [1, 2, 3]
+expected_event_ids = [1, 2, 3]
+
+[[queries]]
+name = "sequenceOneManyMany-Runs"
+query = '''
+sequence
+  [process where serial_event_id == 1]
+  [process where true] [runs=2]
+'''
+expected_event_ids = [1, 2, 3]
 
 [[queries]]
 name = "sequenceConditionManyMany"
@@ -567,6 +576,18 @@ sequence
 expected_event_ids = [1, 2, 3,
                       2, 3, 4,
                       3, 4, 5]
+
+[[queries]]
+name = "sequenceConditionManyMany-Runs"
+query = '''
+sequence
+  [process where serial_event_id <= 3]
+  [process where true] [runs=2]
+'''
+expected_event_ids = [1, 2, 3,
+                      2, 3, 4,
+                      3, 4, 5]
+
 [[queries]]
 name = "sequenceManyConditionMany"
 query = '''
@@ -577,6 +598,7 @@ sequence
 '''
 expected_event_ids = [1, 2, 3,
                       2, 3, 4]
+
 [[queries]]
 name = "sequenceManyManyCondition"
 query = '''
@@ -585,7 +607,16 @@ sequence
   [process where true]
   [process where serial_event_id <= 3]
 '''
-expected_event_ids  = [1, 2, 3]
+expected_event_ids = [1, 2, 3]
+
+[[queries]]
+name = "sequenceManyManyCondition-Runs"
+query = '''
+sequence
+  [process where true] [runs=2]
+  [process where serial_event_id <= 3]
+'''
+expected_event_ids = [1, 2, 3]
 
 [[queries]]
 name = "sequenceThreeManyCondition1"
@@ -596,10 +627,22 @@ sequence
   [process where true]
   [process where true]
 '''
-expected_event_ids  = [1, 2, 3, 4,
-                       2, 3, 4, 5,
-                       3, 4, 5, 6,
-                       4, 5, 6, 7]
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5,
+                      3, 4, 5, 6,
+                      4, 5, 6, 7]
+
+[[queries]]
+name = "sequenceThreeManyCondition1-Runs"
+query = '''
+sequence
+  [process where serial_event_id <= 4]
+  [process where true] [runs=3]
+'''
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5,
+                      3, 4, 5, 6,
+                      4, 5, 6, 7]
 
 [[queries]]
 name = "sequenceThreeManyCondition2"
@@ -610,9 +653,21 @@ sequence
   [process where true]
   [process where true]
 '''
-expected_event_ids  = [1, 2, 3, 4,
-                       2, 3, 4, 5,
-                       3, 4, 5, 6]
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5,
+                      3, 4, 5, 6]
+
+[[queries]]
+name = "sequenceThreeManyCondition2-Runs"
+query = '''
+sequence
+  [process where true]
+  [process where serial_event_id <= 4]
+  [process where true] [runs=2]
+'''
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5,
+                      3, 4, 5, 6]
 
 [[queries]]
 name = "sequenceThreeManyCondition3"
@@ -623,8 +678,19 @@ sequence
   [process where serial_event_id <= 4]
   [process where true]
 '''
-expected_event_ids  = [1, 2, 3, 4,
-                       2, 3, 4, 5]
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5]
+
+[[queries]]
+name = "sequenceThreeManyCondition3-Runs"
+query = '''
+sequence
+  [process where true] [runs=2]
+  [process where serial_event_id <= 4]
+  [process where true]
+'''
+expected_event_ids = [1, 2, 3, 4,
+                      2, 3, 4, 5]
 
 [[queries]]
 name = "sequenceThreeManyCondition4"
@@ -635,7 +701,16 @@ sequence
   [process where true]
   [process where serial_event_id <= 4]
 '''
-expected_event_ids  = [1, 2, 3, 4]
+expected_event_ids = [1, 2, 3, 4]
+
+[[queries]]
+name = "sequenceThreeManyCondition4-Runs"
+query = '''
+sequence
+  [process where true] [runs=3]
+  [process where serial_event_id <= 4]
+'''
+expected_event_ids = [1, 2, 3, 4]
 
 [[queries]]
 name = "twoSequencesWithKeys"
@@ -644,10 +719,10 @@ sequence
   [process where true]        by unique_pid
   [process where opcode == 1] by unique_ppid
 '''
-expected_event_ids  = [48, 53,
-                       53, 54,
-                       54, 56,
-                       97, 98]
+expected_event_ids = [48, 53,
+                      53, 54,
+                      54, 56,
+                      97, 98]
 
 [[queries]]
 name = "twoSequencesWithTwoKeys"
@@ -672,7 +747,19 @@ sequence
 until
   [file where opcode == 2]    by unique_pid
 '''
-expected_event_ids  = []
+expected_event_ids = []
+
+[[queries]]
+name = "fourSequencesByPidWithUntil1-Runs"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid
+  [file where opcode == 0]    by unique_pid  [runs=3]
+until
+  [file where opcode == 2]    by unique_pid
+'''
+expected_event_ids = []
+
 
 [[queries]]
 name = "fourSequencesByPidWithUntil2"
@@ -685,7 +772,19 @@ sequence
 until
   [file where opcode == 200]  by unique_pid
 '''
-expected_event_ids  = [54, 55, 61, 67]
+expected_event_ids = [54, 55, 61, 67]
+
+[[queries]]
+name = "fourSequencesByPidWithUntil2-Runs"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid
+  [file where opcode == 0]    by unique_pid  [runs=3]
+until
+  [file where opcode == 200]  by unique_pid
+'''
+expected_event_ids = [54, 55, 61, 67]
+
 
 #[[queries]]
 #name = "fourSequencesByPidWithUntil3"
@@ -707,7 +806,16 @@ sequence
   [file where opcode == 0]    by unique_pid
   [file where opcode == 0]    by unique_pid
 '''
-expected_event_ids  = [54, 55, 61, 67]
+expected_event_ids = [54, 55, 61, 67]
+
+[[queries]]
+name = "fourSequencesByPid-Runs"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid
+  [file where opcode == 0]    by unique_pid  [runs=3]
+'''
+expected_event_ids = [54, 55, 61, 67]
 
 
 [[queries]]
@@ -719,8 +827,16 @@ sequence
   [file where opcode == 0]    by unique_pid,  process_path
   [file where opcode == 0]    by unique_pid,  process_path
 '''
-expected_event_ids  = [54, 55, 61, 67]
+expected_event_ids = [54, 55, 61, 67]
 
+[[queries]]
+name = "fourSequencesByPidAndProcessPath1-Runs"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid,  process_path
+  [file where opcode == 0]    by unique_pid,  process_path [runs=3]
+'''
+expected_event_ids = [54, 55, 61, 67]
 
 [[queries]]
 name = "fourSequencesByPidAndProcessPathWithUntil"
@@ -733,7 +849,30 @@ sequence
 until
   [file where opcode == 200]  by unique_pid,  process_path
 '''
-expected_event_ids  = [54, 55, 61, 67]
+expected_event_ids = [54, 55, 61, 67]
+
+[[queries]]
+name = "fourSequencesByPidAndProcessPathWithUntil-Runs"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid,  process_path
+  [file where opcode == 0]    by unique_pid,  process_path [runs=3]
+until
+  [file where opcode == 200]  by unique_pid,  process_path
+'''
+expected_event_ids = [54, 55, 61, 67]
+
+[[queries]]
+name = "fourSequencesByPidAndProcessPathWithUntil-RunsExtra"
+query = '''
+sequence
+  [process where opcode == 1] by unique_pid,  process_path
+  [file where opcode == 0]    by unique_pid,  process_path [runs=2]
+  [file where opcode == 0]    by unique_pid,  process_path [runs=1]
+until
+  [file where opcode == 200]  by unique_pid,  process_path
+'''
+expected_event_ids = [54, 55, 61, 67]
 
 [[queries]]
 name = "sequenceOneManyWithJoin"
@@ -742,7 +881,7 @@ sequence
   [process where serial_event_id==1] by unique_pid
   [process where true] by unique_ppid
 '''
-expected_event_ids  = [1, 2]
+expected_event_ids = [1, 2]
 
 
 [[queries]]
@@ -879,9 +1018,20 @@ sequence
   [process where serial_event_id < 5]
   [process where serial_event_id < 5]
 '''
-expected_event_ids  = [1, 2,
-                       2, 3,
-                       3, 4]
+expected_event_ids = [1, 2,
+                      2, 3,
+                      3, 4]
+
+[[queries]]
+name = "doubleSameSequence-Runs"
+query = '''
+sequence
+  [process where serial_event_id < 5] [runs=2]
+'''
+expected_event_ids = [1, 2,
+                      2, 3,
+                      3, 4]
+
 
 [[queries]]
 name = "sequencesOnDifferentEventTypesWithBy"
@@ -890,7 +1040,7 @@ sequence
   [file where opcode==0 and file_name:"svchost.exe"] by unique_pid
   [process where opcode == 1] by unique_ppid
 '''
-expected_event_ids  = [55, 56]
+expected_event_ids = [55, 56]
 
 [[queries]]
 name = "doubleSameSequenceWithBy"
@@ -900,7 +1050,17 @@ sequence
   [file where opcode==0] by unique_pid
 | head 1
 '''
-expected_event_ids  = [55, 61]
+expected_event_ids = [55, 61]
+
+[[queries]]
+name = "doubleSameSequenceWithBy-Runs"
+query = '''
+sequence
+  [file where opcode==0] by unique_pid  [runs=2]
+| head 1
+'''
+expected_event_ids = [55, 61]
+
 
 #[[queries]]
 #name = "doubleSameSequenceWithByAndFilter"
@@ -921,7 +1081,17 @@ sequence
 until [process where opcode==5000] by unique_ppid
 | head 1
 '''
-expected_event_ids  = [55, 61]
+expected_event_ids = [55, 61]
+
+[[queries]]
+name = "doubleSameSequenceWithByUntilAndHead1-Runs"
+query = '''
+sequence
+  [file where opcode==0 and file_name:"*.exe"] by unique_pid [runs=2]
+until [process where opcode==5000] by unique_ppid
+| head 1
+'''
+expected_event_ids = [55, 61]
 
 [[queries]]
 name = "doubleSameSequenceWithByUntilAndHead2"
@@ -934,6 +1104,16 @@ until [process where opcode==1] by unique_ppid
 '''
 expected_event_ids = []
 
+[[queries]]
+name = "doubleSameSequenceWithByUntilAndHead2-Runs"
+query = '''
+sequence
+  [file where opcode==0 and file_name:"*.exe"] by unique_pid [runs=2]
+until [process where opcode==1] by unique_ppid
+| head 1
+'''
+expected_event_ids = []
+
 #[[queries]]
 #name = "doubleJoinWithByUntilAndHead"
 #query = '''

x-pack/plugin/eql/qa/correctness/src/javaRestTest/resources/queries.toml

@@ -383,6 +383,25 @@ sequence by source_address, hostname with maxspan=5s
 time = 2.8286166191101074
 type = "sequence"
 
+[[queries]]
+queryNo = 231
+count = 0
+expected_event_ids = []
+filter_counts = [6, 6, 6, 6]
+filters = [
+  'security where hostname != "newyork" and event_id == 4625',
+  'security where hostname != "newyork" and event_id == 4625',
+  'security where hostname != "newyork" and event_id == 4625',
+  'security where hostname != "newyork" and event_id == 4625'
+]
+query = '''
+sequence by source_address, hostname with maxspan=5s
+  [security where hostname != "newyork" and event_id == 4625] [runs=4]
+'''
+time = 2.8286166191101074
+type = "sequence"
+
+
 [[queries]]
 queryNo = 24
 count = 1
@@ -402,6 +421,23 @@ sequence by source_address, hostname with maxspan=10s
 time = 2.765869617462158
 type = "sequence"
 
+[[queries]]
+queryNo = 241
+count = 1
+expected_event_ids = [2860083, 2860090, 2860098]
+filter_counts = [6, 6, 6]
+filters = [
+  'security where hostname != "newyork" and event_id == 4625',
+  'security where hostname != "newyork" and event_id == 4625',
+  'security where hostname != "newyork" and event_id == 4625'
+]
+query = '''
+sequence by source_address, hostname with maxspan=10s
+  [security where hostname != "newyork" and event_id == 4625] [runs=3]
+'''
+time = 2.765869617462158
+type = "sequence"
+
 [[queries]]
 queryNo = 25
 count = 0

x-pack/plugin/eql/src/main/antlr/EqlBase.g4

@@ -32,7 +32,7 @@ sequenceParams
 
 sequence
     : SEQUENCE (by=joinKeys sequenceParams? | sequenceParams disallowed=joinKeys?)?
-      sequenceTerm sequenceTerm+
+      sequenceTerm+
       (UNTIL until=sequenceTerm)?
     ;
 
@@ -56,7 +56,7 @@ joinTerm
    ;
 
 sequenceTerm
-   : subquery (by=joinKeys)?
+   : subquery (by=joinKeys)? (LB key=IDENTIFIER ASGN value=number RB)?
    ;
 
 subquery

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/parser/LogicalPlanBuilder.java

@@ -6,9 +6,10 @@
  */
 package org.elasticsearch.xpack.eql.parser;
 
+import org.antlr.v4.runtime.Token;
 import org.antlr.v4.runtime.tree.ParseTree;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xpack.eql.parser.EqlBaseParser.BooleanExpressionContext;
 import org.elasticsearch.xpack.eql.parser.EqlBaseParser.EventFilterContext;
 import org.elasticsearch.xpack.eql.parser.EqlBaseParser.IntegerLiteralContext;
@@ -62,7 +63,7 @@ import static org.elasticsearch.xpack.ql.tree.Source.synthetic;
 
 public abstract class LogicalPlanBuilder extends ExpressionBuilder {
 
-    static final String FILTER_PIPE = "filter", HEAD_PIPE = "head", TAIL_PIPE = "tail";
+    static final String FILTER_PIPE = "filter", HEAD_PIPE = "head", TAIL_PIPE = "tail", RUNS = "runs";
 
     static final Set<String> SUPPORTED_PIPES = Sets.newHashSet("count", FILTER_PIPE, HEAD_PIPE, "sort", TAIL_PIPE, "unique",
             "unique_count");
@@ -253,7 +254,45 @@ public abstract class LogicalPlanBuilder extends ExpressionBuilder {
                             found);
                 }
             }
-            queries.add(sequenceTerm);
+            // check runs
+            Token key = sequenceTermCtx.key;
+            if (key != null) {
+                String k = key.getText();
+                if (RUNS.equals(k) == false) {
+                    throw new ParsingException(source(key), "Unrecognized option [{}], expecting [{}]", k, RUNS);
+                }
+            }
+
+            int runs = 1;
+            NumberContext numberCtx = sequenceTermCtx.number();
+            if (numberCtx instanceof IntegerLiteralContext) {
+                Number number = (Number) visitIntegerLiteral((IntegerLiteralContext) numberCtx).fold();
+                long value = number.longValue();
+                if (value < 1) {
+                    throw new ParsingException(source(numberCtx), "A positive runs value is required; found [{}]", value);
+                }
+                if (value > 100) {
+                    throw new ParsingException(source(numberCtx), "A query cannot be repeated more than 100 times; found [{}]", value);
+                }
+                runs = (int) value;
+            }
+
+            int numberOfQueries = queries.size() + runs;
+            if (numberOfQueries > 256) {
+                throw new ParsingException(
+                    source(sequenceTermCtx),
+                    "Sequence cannot contain more than 256 queries; found [{}]",
+                    numberOfQueries
+                );
+            }
+
+            for (int i = 0; i < runs; i++) {
+                queries.add(sequenceTerm);
+            }
+        }
+
+        if (queries.size() < 2) {
+            throw new ParsingException(source, "A sequence requires a minimum of 2 queries, found [{}]", queries.size());
         }
 
         // until is already parsed through sequenceTerm() above
@@ -266,7 +305,7 @@ public abstract class LogicalPlanBuilder extends ExpressionBuilder {
         return new Sequence(source, queries, until, maxSpan, fieldTimestamp(), fieldTiebreaker(), resultPosition());
     }
 
-    public KeyedFilter visitSequenceTerm(SequenceTermContext ctx, List<Attribute> joinKeys) {
+    private KeyedFilter visitSequenceTerm(SequenceTermContext ctx, List<Attribute> joinKeys) {
         return keyedFilter(joinKeys, ctx, ctx.by, ctx.subquery());
     }
 

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/parser/LogicalPlanTests.java

@@ -179,6 +179,16 @@ public class LogicalPlanTests extends ESTestCase {
         assertEquals(new TimeValue(2, TimeUnit.SECONDS), maxSpan);
     }
 
+    public void testRepeatedQuery() throws Exception {
+        LogicalPlan plan = parser.createStatement("sequence " + " [any where true] [runs=2]" + " [any where true]");
+        plan = defaultPipes(plan);
+        assertEquals(Sequence.class, plan.getClass());
+        Sequence seq = (Sequence) plan;
+
+        List<? extends LogicalPlan> queries = seq.queries();
+        assertEquals(3, queries.size());
+    }
+
     private LogicalPlan wrapFilter(Expression exp) {
         LogicalPlan filter = new Filter(Source.EMPTY, relation(), exp);
         Order order = new Order(Source.EMPTY, timestamp(), OrderDirection.ASC, NullsPosition.FIRST);

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/planner/QueryTranslatorFailTests.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.eql.planner;
 
+import org.elasticsearch.xpack.eql.EqlClientException;
 import org.elasticsearch.xpack.eql.analysis.VerificationException;
 import org.elasticsearch.xpack.ql.ParsingException;
 import org.elasticsearch.xpack.ql.QlIllegalArgumentException;
@@ -208,9 +209,22 @@ public class QueryTranslatorFailTests extends AbstractQueryTranslatorTestCase {
 
     public void testLikeWithNumericField() {
         VerificationException e = expectThrows(VerificationException.class,
-                () -> plan("process where pid like \"*.exe\""));
+            () -> plan("process where pid like \"*.exe\"")
+        );
         String msg = e.getMessage();
         assertEquals("Found 1 problem\n" +
-                "line 1:15: argument of [pid like \"*.exe\"] must be [string], found value [pid] type [long]", msg);
+            "line 1:15: argument of [pid like \"*.exe\"] must be [string], found value [pid] type [long]", msg);
     }
+
+    public void testSequenceWithTooLittleQueries() throws Exception {
+        String s = errorParsing("sequence [any where true]");
+        assertEquals("1:2: A sequence requires a minimum of 2 queries, found [1]", s);
+    }
+
+    public void testSequenceWithIncorrectOption() throws Exception {
+        EqlClientException e = expectThrows(EqlClientException.class, () -> plan("sequence [any where true] [repeat=123]"));
+        String msg = e.getMessage();
+        assertEquals("line 1:29: Unrecognized option [repeat], expecting [runs]", msg);
+    }
+
 }