NextGen Pipeline (CODENAME: Big Gulp)

[andrewvc]

(moved from #4153) (moved to #4254 )

ATTN: This is no longer a POC but production ready code!

This is PR that aims to accomplish the following objectives.

1. Provide an in-memory implementation (no persistence), of the pipeline I described in Next Generation pipeline architecture for persistence and performance #3693
2. Be completely backward compatible with current logstash plugins, nothing should break
3. Allow for complete pipeline snapshotting (even with the Elasticsearch plugin) to address concerns raised by @jsvd in add shutdown controller to force exit on stalled shutdown #4051. This would allow forced shutdowns without losing data. This would require the elasticsearch output to implement receive_multi and no longer use its own internal queue
4. Expose a new 'batch' interface for filters and output plugins, allowing them to process events in microbatches for efficiency. This could have very significant potential for speed improvements for filters like the useragent filter, which would only have to lookup the number of unique keys per batch vs. each key individually. This would also obviate the need for a separate buffer in the elasticsearch output.

I would say that, along with these changes we should contemplate adding a new plugin config option which would by default set the number of workers for the plugin to be == the number of worker threads. That will give the best user experience for this.

Notes on implementation

This is 90% the system described in #3693 , the one exception to this is that output plugins are assumed to be not threadsafe. This was necessary to make it backward compatible with 2.x series plugins. When the # of output workers is equal to the number of worker threads specified with -w the behavior should be very similar to what the original behavior was. I think the default worker fix described above might actually be a long term solution here.

Persistence Examples
Since this is just a POC, the pipeline shutdown persistence is demonstrated in the writing of two files on shutdown:

Update: This functionality remains in the dump_inflight function, but is currently never called. We may enable this in a future release

1. /tmp/ls_current_batches: The contents of in-memory buffers BEFORE the shutdown sequence is initiated. A 'fast' shutdown (as described in add shutdown controller to force exit on stalled shutdown #4051 could simple write this then exit safely).
2. /tmp/ls_current_batches_post_close: This is just there for debugging in the POC, this is written after a clean shutdown. It will always be empty (unless there's a bug in the pipeline!).

** Notes For Reviewers **

1. This PR involves a small number of changes to the AST. To enable batch oriented processing for each batch we:
   1. Execute the filters one by one accumulating a new batch of filtered events.
   2. Execute output_func once per each function which no longer invokes the output func but rather returns an array of outputs that the event should be sent to.
   3. We use the info from the step above to build separate batches per output as seen in output_batch (in the pipeline)
2. This PR adds 2 new CLI options batch_size and batch_poll_wait to control the new features of the pipeline.
3. PLEASE verify my performance results for yourself. I'm currently using this config and running it with time bin/logstash -f that_config.conf. Then comparing it with Logstash 2.0
4. I set the default batch size to be 125 to match the Elasticsearch output defaults assuming four workers (4*125 = 500). This seemed like a safe default.

Performance

I used the following config to test this thing out:
bin/logstash -w 4 -e "input { generator {} } filter { mutate { add_field => {'foo' => 'bar'} } } output { http { http_method => 'post' url => 'http://localhost:4567/hi' workers => 2 } }"

Note that this requires a little web service to hit. I used the following

require 'sinatra'

post '/hi' do
  puts "Got #{request.body.read}"
  [200, "Thanks!"]
end

Performance

Also worth noting, this new pipeline seems, in at least one micro-benchmark, to be as much as 63% faster!

~/p/logstash (pipeline_poc) $ bin/logstash -w 4 -e "input { generator {} } filter {  mutate { add_field => {'foo' => 'bar'} } } output { stdout {} }" | pv -l > /dev/null
^C61M 0:01:00 [67.1k/s] [                                                                                                                                                    <=>                         ]
~/p/logstash (pipeline_poc) $ cd ~/Downloads/logstash-2.0.0
~/D/logstash-2.0.0 $ bin/logstash -w 4 -e "input { generator {} } filter {  mutate { add_field => {'foo' => 'bar'} } } output { stdout {} }" | pv -l > /dev/null
^C24M 0:01:00 [  41k/s] [

Doing further benchmarking I was able to get the throughput up to 74k/s using a batch size of 40 (the default is 20). Upping the batch size to 200 made it go down to 71k/s, and upping it to 2000 brought it down a lot, to 50k/s. So, batch size has a big impact! Probably due to not being able to use memory/the cache as efficiently. We'll have to do further tests with different messages sizes as well.

[jsvd]

this is a no-op unless I'm missing something

[andrewvc]

Yep! Good find.

[jsvd]

@andrewvc I love this refactor, makes the pipeline a lot simpler 👍
we need to test this heavily now (pretty sure now a lot of rake test:core breaks with these changes) and check the actual impact on performance and check if doesn't break on edge cases.

Regarding the dump of batches, I'm not sure how useful it is until there's a complementary implementation of output buffer dumps. and once that's implemented then we need to figure out a way to make the dumps readable with proper configuration fingerprint checking, etc.

I'd suggest leaving dumping out of this PR, the same way I'm going to remove it from #4051

[andrewvc]

@jsvd thanks!

To be clear, the buffer does cover outputs so long as the output doesn't implement its own internal buffer. I think I'd like to keep them in as an option for the upcoming refactor of the ES outputs

[suyograo]

@andrewvc +1 on moving the dump option to another PR as @jsvd mentioned. This refactor really does not have anything to do with dumping the queue contents, its about simplifying the pipeline to have one queue which benefits the upcoming persistence queue work. The other benefit is being able to enable fast shutdown

[andrewvc]

@suyograo sure thing, I'll just leave in place the locking guarantees and internal tracking of the buffers. It'll be useful for future metrics work, and I'll be able to add those stats to the reporter module. It'll be a good to get a metric like 4/5 workers active, 65 items inflight.

[andrewvc]

This should probably invoke receive_multi to avoid the unnecessary mutex synchronize in handle_multi

[ph]

wew puts?

[ph]

Could be a one liner?

[andrewvc]

Good call :)

[ph]

@andrewvc I am not done on the review yet, I'll continue later.

[guyboertje]

Consider a setting @is_multi_worker once and test on that...
either:

if (@is_multi_worker = @worker > 1)
  define_singleton...
  ...
else
  @worker_plugins = [self]
end

or

@is_multi_worker = @worker > 1
if @is_multi_worker
  ...
else
  ...
end

[andrewvc]

Moved to #4254