Skip to content

Commit

Permalink
EQL: Introduce until functionality (elastic#59292)
Browse files Browse the repository at this point in the history
Sequences now support until conditional, which prevents a match from
occurring if the until matches a document while doing look-ups.
Thus a sequence must complete before the until condition matches - if
any document within the sequence occurs at, or after, the until hit, the
sequence is discarded.

(cherry picked from commit 1ba1b9f)
  • Loading branch information
costin committed Jul 9, 2020
1 parent d07b11b commit d9c1e53
Show file tree
Hide file tree
Showing 14 changed files with 506 additions and 235 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.junit.BeforeClass;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
Expand Down Expand Up @@ -159,7 +160,10 @@ private EqlClient eqlClient() {

protected void assertSearchHits(List<SearchHit> events) {
assertNotNull(events);
assertArrayEquals("unexpected result for spec: [" + spec.toString() + "]", spec.expectedEventIds(), extractIds(events));
long[] expected = spec.expectedEventIds();
long[] actual = extractIds(events);
assertArrayEquals("unexpected result for spec: [" + spec.toString() + "]" + Arrays.toString(expected) + " vs " + Arrays.toString(
actual), expected, actual);
}

private static long[] extractIds(List<SearchHit> events) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -345,36 +345,9 @@ process where true
| sort serial_event_id
'''


[[queries]]
name = "fourSequencesByPidWithUntil1"
query = '''
sequence
[process where opcode == 1] by unique_pid
[file where opcode == 0] by unique_pid
[file where opcode == 0] by unique_pid
[file where opcode == 0] by unique_pid
until
[file where opcode == 2] by unique_pid
'''
expected_event_ids = []

[[queries]]
name = "fourSequencesByPidWithUntil2"
query = '''
sequence
[process where opcode == 1] by unique_pid
[file where opcode == 0] by unique_pid
[file where opcode == 0] by unique_pid
[file where opcode == 0] by unique_pid
until
[file where opcode == 200] by unique_pid
'''
expected_event_ids = [54, 55, 61, 67]

[[queries]]
name = "doubleSameSequenceWithByAndFilter"
query = '''
sequence
[file where opcode=0] by unique_pid
[file where opcode=0] by unique_pid
Expand All @@ -385,17 +358,6 @@ expected_event_ids = [87, 92]
[[queries]]
name = "doubleSameSequenceWithByUntilAndHead2"
query = '''
sequence
[file where opcode=0 and file_name="*.exe"] by unique_pid
[file where opcode=0 and file_name="*.exe"] by unique_pid
until [process where opcode=1] by unique_ppid
| head 1
'''
expected_event_ids = []

[[queries]]
name = "doubleJoinWithByUntilAndHead"
query = '''
join
[file where opcode=0 and file_name="*.exe"] by unique_pid
[file where opcode=2 and file_name="*.exe"] by unique_pid
Expand Down Expand Up @@ -698,29 +660,6 @@ expected_event_ids = [48, 50, 51, 54, 93]
[[queries]]
name = "twoSequencesWithTwoKeysAndUntil"
query = '''
sequence by user_name
[file where opcode=0] by pid,file_path
[file where opcode=2] by pid,file_path
until
[process where opcode == 2] by ppid,process_path
'''
expected_event_ids = []

[[queries]]
name = "twoSequencesWithUntil"
query = '''
sequence by user_name
[file where opcode=0] by pid,file_path
[file where opcode=2] by pid,file_path
until
[process where opcode == 5] by ppid,process_path
| head 2
'''
expected_event_ids = [55, 59, 61, 65]

[[queries]]
name = "twoSequencesWithHead"
query = '''
join by user_name
[file where true] by pid,file_path
[process where true] by ppid,process_path
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,10 @@ public BoxedQueryRequest from(Ordinal begin) {
return this;
}

public Ordinal after() {
return after;
}

public Ordinal from() {
return from;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,14 @@ public class KeyAndOrdinal {
this.ordinal = ordinal;
}

public SequenceKey key() {
return key;
}

public Ordinal ordinal() {
return ordinal;
}

@Override
public int hashCode() {
return Objects.hash(key, ordinal);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,12 @@
*/
package org.elasticsearch.xpack.eql.execution.assembler;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.xpack.eql.execution.search.Limit;
import org.elasticsearch.xpack.eql.execution.search.Ordinal;
import org.elasticsearch.xpack.eql.execution.sequence.Sequence;
import org.elasticsearch.xpack.eql.execution.sequence.SequenceStateMachine;
import org.elasticsearch.xpack.eql.session.Payload;
Expand All @@ -21,6 +22,8 @@
*/
class Matcher {

private final Logger log = LogManager.getLogger(Matcher.class);

// NB: just like in a list, this represents the total number of stages yet counting starts at 0
private final SequenceStateMachine stateMachine;
private final int numberOfStages;
Expand Down Expand Up @@ -48,27 +51,33 @@ boolean match(int stage, Iterable<Tuple<KeyAndOrdinal, SearchHit>> hits) {
// early skip in case of reaching the limit
// check the last stage to avoid calling the state machine in other stages
if (stateMachine.reachedLimit()) {
log.trace("Limit reached {}", stateMachine.stats());
return false;
}
}
}
log.trace("{}", stateMachine.stats());
return true;
}

boolean until(Iterable<Ordinal> markers) {
// no-op so far

return false;
void until(Iterable<KeyAndOrdinal> markers) {
stateMachine.until(markers);
}

public boolean hasCandidates(int stage) {
boolean hasCandidates(int stage) {
return stateMachine.hasCandidates(stage);
}

void dropUntil() {
stateMachine.dropUntil();
}

Payload payload(long startTime) {
List<Sequence> completed = stateMachine.completeSequences();
TimeValue tookTime = new TimeValue(System.currentTimeMillis() - startTime);
return new SequencePayload(completed, false, tookTime);
Payload p = new SequencePayload(completed, false, tookTime);
stateMachine.clear();
return p;
}

@Override
Expand Down
Loading

0 comments on commit d9c1e53

Please sign in to comment.