Pipeline changes for resiliency, phase1

[suyograo]

This meta issue describes further changes which are required in the pipeline, to implement event resiliency in Logstash, phase 1. See issues #2605, #2606 and #2607 for more details.

Background

In release 2.2, we combined filters and outputs in one execution unit and removed the two, fixed-sized queues. In 2.3, we'll be introducing Event implementation in Java, which will reduce serialization/deserialization costs when persisting Events to disk.

---

Define pluggable queue interface

Create a well defined interface for handling queuing of Events inside Logstash. This would allow the queuing implementation to be pluggable. Users can be provided an option of using a fast, in-memory queue or more resilient, disk-based queue. This interface will also provide us a way to assess and contrast performance impacts.

Disk based queuing

The default implementation of the queue interface will be disk backed. For this, port @colinsurprenant mmap based persistent queues work to Java. This will form the disk based queuing infrastructure inside of Logstash.

In-memory queuing

Wrap the existing Synchronous Queue to this interface so we can validate behavior, and provide alternative in-memory option.

Codecs

An important design consideration is whether codecs - which deal with event representation both at input and output stages - should be executed before or after persisting Events to disk. After much discussion with the Logstash team, codecs will now be executed after the persistence layer. The advantages of codecs-after-persistence is two fold:

• Ensures events (in byte representation) are written immediately to disk once they are received in the Logstash input. For some inputs like TCP, which handle raw events off the wire, this makes serialization to disk faster.
• Codecs can be parallelized when they are part of the pipeline. So essentially, they become part of filter and output worker unit. This can bring in a good performance improvements for codecs like json, nmap etc

Event Mill - a new stage inside Inputs

The main argument for keeping codecs in the input stage was the usecase of multiline events. Multiline events can appear out of order, and a full event needs to be constructed at the input plugin. This involves identifying beginning of the multiline event, buffering partial events until the end pattern is reached, and finally creating a single Event object. Today this process is implemented inside the Multiline Codec. There are also other usecases like json_lines codec.

The idea is to create a new stage called Event Mill (yay logs!) which creates a fully formed event object from event data that spans multiple lines. This means we can extract the implementation from ML codec to a common library which can be reused in inputs like File, TCP, S3 etc. The output of Event Mill stage will be a raw Event, not something you can use to munge data. The idea is to be able to persist this with some metadata.

For users, this change will be transparent - i.e. the same config options to identify event delimiters (beginning and end patterns) will continue to work. This hugely simplifies persistence infrastructure since events are fully formed at the time of writing to disk.

Alternatively, multiline can be handles after the persistence layer, but this means event parts will be out of order in the persisted queue. One way to solve this would be to group data based on stream identities (host, fileID etc). Since this adds unnecessary complexity, its better for multiline to be handled upstream. Also in many cases like Filebeat, multiline is handled close to the source and input data will already be a fully formed Event.

With these changes, pipeline structure becomes:

| Input -> Tokenizer | -> | Disk-based Queue | -> | Event Batcher | -> | Deserializer (Decoder) -> Filters -> Outputs |

Output Batching and Acking

All Outputs should be synchronous. Outputs using Stud::Buffer should now use buffering from the pipeline level (implement #multi_receive).

There is also scope for introducing batching in other stages like codecs and filters to improve performance. This will allow us to acknowledge that data has been processed to the queueing stage. For disk based persistence, bookkeeping can be performed after an ack has been received when all the outputs are finished processing the events.

---

Tasks

• Create pluggable queue interface Create pluggable queue interface #4768
• Wrap existing Synchronous Queue implementation with the queue interface Wrap existing Synchronous Queue with the queue interface #4804
• Port JRuby mmap data structure to Java Implement mmap-based queues #4805
• Extract multiline event handling into shared library Extract multiline event handling into shared library #4769
• Introduce Event Mill (thanks @guyboertje ) stage in inputs Event Mills or how to turn a text stream into an Event stream #4858
• Make codecs optional in inputs and remove codec.decode call in inputs Make codecs optional in inputs and remove codec.decode call in inputs Make codecs optional in inputs and remove codec.decode call in inputs #4807
• Make output synchronous so we can ack (Removing Stud::Buffer and replace with #multi_receive Review Outputs for use of Stud::Buffer and replace with #multi_receive #4808

---

• Think about partitioning strategy to ack out-of-order (when using multiple workers). Do we do logical or physical partition? # of partitions = # of workers?
• Move codec's deserialize component into the pipeline, after queuing stage
• Long term, think about merging decoding stage to filter (since they'll be doing the same thing)
• Provide metrics from the persisted queue - consumer lag, fsync stats etc

[suyograo]

Closed in favor of #5638