Collapse multiline logs based on a start line.

[jeschkies]

Summary:
This is a very simple approach based on #1380 to provide multiline
or block log entries in promtail.

A multiline stage is added to pipelines. This stages matches a start
line. Once a start line is matched all following lines are appended
to an entry and "dropped". Once a new start line is matched the former
block of multilines is send.

This approach has two downside because log entires are not sent until a
new start line is matched.

1. Lines can linger for a long time. The multiline stage should flush out
   lines if now new start line is matched in a certain time frame.
   However, the current pipeline interface cannot actively push entries.
   So a time based flushing would require a bigger refactoring.

2. If the observed system crashes the last log lines are not sent. Thus
   important information might be lost.

[jeschkies]

I'm sure I did not cover all the edge cases.

[jeschkies]

This is probably the confusing part. A new line is pushed but we get the older multiline blog. Ideally a pipeline stage would be able to push out lines without a call. This could be done with channels.

[jeschkies]

Once we are processing inputs from channels and push changes out to channels the multiline stage could add a simple timeout to flush a block:

struct  Entry type {
    labels model.LabelSet,
    extracted map[string]interface{},
    t *time.Time,
    entry *string
}

// Process implements Stage
func (m *multilineStage) Process(entries <- chan Entry) <- chan Entry {
  out := make(chan Entry)
  currentMultiline := make(bytes.Buffer)
 
  loop { // This should be in a goroutine
    select {
    case next := <-entries:
        if isNewStart(entry) {
            // Flush
            out <- NewEntry(currentMultiline.String())
            currentMultiline.reset()
        }
        currentMultiline.WriteString(entry.entry)
    case <-time.After(timeout * time.Second):
            // Flush
            out <- currentMultiline.String()
            currentMultiline.reset()
    }
  }
  return out
}

Process is not a fitting name any more since it is called once for all incoming entries. Anyhow, I think it illustrates how the flush time out becomes much simpler.

[codecov-io]

Codecov Report

Merging #2971 (7be2aec) into master (8ea5fd1) will decrease coverage by 0.13%.
The diff coverage is 41.66%.

@@            Coverage Diff             @@
##           master    #2971      +/-   ##
==========================================
- Coverage   61.84%   61.71%   -0.14%     
==========================================
  Files         182      183       +1     
  Lines       14870    14906      +36     
==========================================
+ Hits         9197     9199       +2     
- Misses       4824     4857      +33     
- Partials      849      850       +1     

Impacted Files	Coverage Δ
pkg/logentry/stages/stage.go	41.79% <0.00%> (-2.66%)	⬇️
pkg/logentry/stages/multiline.go	46.87% <46.87%> (ø)
pkg/promtail/positions/positions.go	46.80% <0.00%> (-11.71%)	⬇️
pkg/logql/evaluator.go	91.24% <0.00%> (-0.41%)	⬇️

[jeschkies]

Superseded by cyriltovena#7