Graph Pipeline Design

I found some time while in Lake tahoe to experiment (https://github.com/elastic/logstash/pull/4727) with what I'm terming a 'graph' pipeline execution model. The critical ideas behind this design are as follows:
1. **Improve performance:** I've seen gains ~27% in the apache example
2. **Reduce complexity:** Cross-compiling to Ruby is a source of complexity as anyone who has read [config_ast](https://github.com/elastic/logstash/blob/master/logstash-core/lib/logstash/config/config_ast.rb) can attest.
3. **Move us closer to a pure Java core/java plugins:** Implementing this in pure java is easy, we don't rely on the JRuby complier at all.
4. **Enable more filter operations by allowing filters to operate on batches instead of single events:** The wins here are significant for filters that perform IO. Even for ones that do batch functionality can be exploited. The GeoIP and UserAgent filters would be able to group their cache lookups for instance. For critical sections of code the granularity of mutexes is easier to manage as well and should require fewer context switches.
5. **Have an easy to work with IR for Logstash configs:** The benefits for testing and debugging are huge.
6. **Pave the way for generated configs via UI**: By using a graph format it's much easier to visually format and export Logstash configs. Initially this would be an internal IR generated by @colinsurprenant 's ANTLR grammar (https://github.com/colinsurprenant/logstash-antlr-config).
7. **Ease debugging and tracing**: Visualizing a graph is simple in a UI, it's also easy to show hotspots in one. It's also easy to enable tracing plugins that modify the graph such that extra stats are recorded.
### Implementation

The current model compiles the Logstash config to ruby code, then repeatedly executes it. An abbreviated version of the execution is what is shown below

``` ruby
queue = LogStash::Queue.new
start_input_threads(queue)
worker_threads.times do
Thread.new do
loop do
batch = take_batch(queue)
# Events are filtered one at a time
filtered = batch.map {|e| filter_func(e)}
outputs_to_events = map_outputs_to_events(filtered)
outputs_to_events.each {|output, events| output.multi_receive(events)}
end
end
```

Note that the entire filter chain for a single event is compiled into a single function, `filter_func`. This is also true of `output_func`, though in that case it returns which outputs the event should be sent to rather than directly executing it.

The graph model is more simplistic. We model the entire pipeline as a graph (from inputs, to queues, to filters, to outputs. The IR serialized to yaml might look like the following:

``` yaml

---
graph:
log-reader:
component: input-stdin
to: [main-queue]
main-queue:
component: queue-synchronous
to: [apache-grok]
apache-grok:
component: filter-grok
options:
match:
message: '%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}'
to: [code-splitter]
code-splitter:
component: predicate-ifelse-ruby
to:
- ["value = event['[response]'] && value ? value > 399 : false", [error-tagger]]
- ["__ELSE__", [apache-geoip]]
error-tagger:
component: filter-mutate
options:
add_tag:
fb: weird_error
to: [apache-geoip]
apache-geoip:
component: filter-geoip
options:
source: geoip
target: geoip
to: [apache-ua]
apache-ua:
component: filter-useragent
options:
source: agent
target: useragent
to: [apache-date]
apache-date:
component: filter-date
options:
match: [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
locale: en
to: [main-out]
main-out:
component: output-file
options:
path: /tmp/graph-out.json
codec: json_lines
```

With the graph expressed as such we can execute according to any strategy of our choice.

``` java
// Pseudo-java-code
Graph graph = GraphConfig.loadFile("myconfig")

// For now we just support one queue, but easy to change!
Vertex queue = graph.getById("main-queue")

// Parse out inputs for execution in a separate thread
List inputs = graph.inputs()
startInputs(inputs, queue)

// Get the subgraph of nodes underneath the main queue
Graph pipelineWorkerGraph = graph.getById("main-queue").subGraph()
// Sort the vertices topologically to optimize execution order (and check for cycles in the DAG)
// This will mean we deliver the maximum batch size to each Vertex
List executionOrderedVertices = pipelineWorkerGraph.topologicalSort()

for (i =0; i < workerCount; i++) {
new Thread(new Runnable() {
@Override
public void run() throws Exception {
while (true) {
Map> edgeEvents = new HashMap>;
// There are some interesting opportunities beyond the toposort to optimize
// traversal here for empty boolean paths
for (Vertex v in executionOrderedVertices) {
// Process the vertex, returning a list of results mapped to edges
// pass in the current batche's edgeEvents map so it can pull from
// previously calculated edge values.
Map> vEdgeEvents = v.execute(edgeEvents)
// Merge the result map into the list of all event results
vEdgeEvents.stream().putAll(vEdgeEvents)
}
}
}
});
}
```
### Why is it Faster?

I'm not sure. I know that implementing a similar graph pattern in ruby yielded similar results. My money is on some inefficiency in the generated ruby OR the greater CPU cache locality afforded by filtering batches. Either way I believe regardless of performance the other benefits stack.