Generalization of reader pipeline in Beats

[kvch]

Goal

Filebeat has various readers that are strictly tied to the log input. However, the readers (usually multiline and syslog) are requested to be used from multiple inputs. The goal is to generalize these readers so more inputs can utilize them.

Current state

The reader pipeline consists of the following readers:

• readfile.EncoderReader: reads an input with the configured encoding
• readfile.LineReader: reads a line from a file
• readfile.LimitReader: truncates message if it is too long
• readfile.StripNewLine: removes the configured newline characters from the end of the message
• readfile.TimeoutReader: flushes the message if the configured time has elapsed
• multiline.Reader: creates a single message from multiple ones based on patterns and timeout
• readjson.DockerJSON: parses JSON events from Docker
• readjson.JSONReader: parses arbitrary JSON events

All of the readers above implement the following reader.Reader interface:

type Reader interface {
    Next() (Message, error)
}

All readers except for readfile.EncoderReader read a reader.Message from an underlying reader.Reader.

These readers are only exposed and used in log input and in DockerJSON in container input. However, these readers could be reused in more inputs (e.g. journal, tcp, etc.). It is possible that someone needs to read multiline messages from a systemd journal.

But right now adding this feature to more inputs is not straightforward because the reader pipeline is too tightly coupled with the log input. The readers are part of libbeat, but they can only be used from Filebeat. In order to provide a similar experience for more Beats, the pipeline has to be generalized.

Proposal

The "source" parts of inputs have to be decoupled. TCP and UDP inputs were created like this. The concrete data sources are part of the package inputsource. More sources can be extracted from inputs to provide flexibility:

• log: reads from a log file
• journal: reads a journal

The inputsources should implement the interface oi.Reader or reader.Reader in order to plug into the pipeline.

Progress

So far filestream input is the only one that leverages the same reader.Reader functionality. During adopting the readers, a few things have been sorted out. In the interim, a new parser interface was introduced so the existing reader.Reader structures can function as previously in the log input. The new interface already has an improved JSON handling. The interface is not exposed yet.

// parser transforms or translates the Content attribute of a Message.
// They are able to aggregate two or more Messages into a single one.
type parser interface {
    io.Closer
    Next() (reader.Message, error)
}

Next steps

• Decide if the new inputs should use reader.Reader or beat.Event
• Create a transformer that converts a reader.Message to a beat.Event
• Come up with interface between a reader.Reader and inputs to support e.g. journals

[urso]

We should also check if there are other parsers in Filebeat inputs that we might want to generalize. E.g. syslog or xml coming to mind

[kvch]

This has been done. The new processor is called parsers. There is an open issue for adding this to all inputs in Filebeat: #26130