Add multiline.flush_pattern option

This allows for specifying a regex, which will flush the current multiline, thus ending the current multiline. Useful for using multiline to capture application events with 'start' and 'end' lines.

Example configuration
  multiline.pattern: 'start'
  multiline.negate: true
  multiline.match: after
  multiline.flush_pattern: 'end'

(#3964)

CHANGELOG.asciidoc

@@ -343,6 +343,7 @@ https://github.com/elastic/beats/compare/v5.2.2...v5.3.0[View commits]
 - The `symlinks` and `harverster_limit` settings are now GA, instead of experimental. {pull}3525[3525]
 - close_timeout is also applied when the output is blocking. {pull}3511[3511]
 - Improve handling of different path variants on Windows. {pull}3781[3781]
+- Add multiline.flush_pattern option, for specifying the 'end' of a multiline pattern {pull}4019[4019]
 
 
 *Metricbeat*

filebeat/docs/multiline.asciidoc

@@ -16,6 +16,8 @@ lines that match the specified regular expression are considered either continua
 
 * the `match` option, which specifies how Filebeat combines matching lines into an event. You can specify `before` or `after`.
 
+* the 'flush_pattern option, which specifies a regular expression, in which the current multiline will be flushed from memory, ending the multline-message.
+
 See the full documentation for <<multiline>> to learn more about these options. Also read <<yaml-tips>> and 
 <<regexp-support>> to avoid common mistakes.
 
@@ -137,8 +139,28 @@ multiline.match: after
 This configuration uses the `negate: true` and `match: after` settings to specify that any line that does not match the 
 specified pattern belongs to the previous line.
 
+==== Application events
+
+Sometimes your application logs contain events, that begin and end with custom markers, such as the following example:
+
+[source,shell]
+-------------------------------------------------------------------------------------
+[2015-08-24 11:49:14,389] Start new event
+[2015-08-24 11:49:14,395] Content of processeing something
+[2015-08-24 11:49:14,399] End event
+-------------------------------------------------------------------------------------
+
+To consolidate this as a single event in Filebeat, use the following multiline configuration:
 
+[source,yaml]
+-------------------------------------------------------------------------------------
+multiline.pattern: 'Start new event'
+multiline.negate: true
+multiline.match: after
+multiline.flush_pattern: 'End event'
+-------------------------------------------------------------------------------------
 
+The 'flush_pattern' option, specifies a regex at which the current multiline will be flushed. If you think of the 'pattern' option specifying the beginning of an event, the 'flush_pattern' option will specify the end or last line of the event.
 
 
 

filebeat/harvester/reader/multiline.go

@@ -20,16 +20,17 @@ import (
 // Errors will force the multiline reader to return the currently active
 // multiline event first and finally return the actual error on next call to Next.
 type Multiline struct {
-	reader    Reader
-	pred      matcher
-	maxBytes  int // bytes stored in content
-	maxLines  int
-	separator []byte
-	last      []byte
-	numLines  int
-	err       error // last seen error
-	state     func(*Multiline) (Message, error)
-	message   Message
+	reader       Reader
+	pred         matcher
+	flushMatcher *match.Matcher
+	maxBytes     int // bytes stored in content
+	maxLines     int
+	separator    []byte
+	last         []byte
+	numLines     int
+	err          error // last seen error
+	state        func(*Multiline) (Message, error)
+	message      Message
 }
 
 const (
@@ -71,6 +72,8 @@ func NewMultiline(
 		return nil, err
 	}
 
+	flushMatcher := config.FlushPattern
+
 	if config.Negate {
 		matcher = negatedMatcher(matcher)
 	}
@@ -93,13 +96,14 @@ func NewMultiline(
 	}
 
 	mlr := &Multiline{
-		reader:    reader,
-		pred:      matcher,
-		state:     (*Multiline).readFirst,
-		maxBytes:  maxBytes,
-		maxLines:  maxLines,
-		separator: []byte(separator),
-		message:   Message{},
+		reader:       reader,
+		pred:         matcher,
+		flushMatcher: flushMatcher,
+		state:        (*Multiline).readFirst,
+		maxBytes:     maxBytes,
+		maxLines:     maxLines,
+		separator:    []byte(separator),
+		message:      Message{},
 	}
 	return mlr, nil
 }
@@ -186,6 +190,20 @@ func (mlr *Multiline) readNext() (Message, error) {
 			return msg, nil
 		}
 
+		// handle case when endPattern is reached
+		if mlr.flushMatcher != nil {
+			endPatternReached := (mlr.flushMatcher.Match(message.Content))
+
+			if endPatternReached == true {
+				// return collected multiline event and
+				// empty buffer for new multiline event
+				mlr.addLine(message)
+				msg := mlr.finalize()
+				mlr.resetState()
+				return msg, nil
+			}
+		}
+
 		// if predicate does not match current multiline -> return multiline event
 		if mlr.message.Bytes > 0 && !mlr.pred(mlr.last, message.Content) {
 			msg := mlr.finalize()

filebeat/harvester/reader/multiline_config.go

@@ -8,11 +8,12 @@ import (
 )
 
 type MultilineConfig struct {
-	Negate   bool           `config:"negate"`
-	Match    string         `config:"match"       validate:"required"`
-	MaxLines *int           `config:"max_lines"`
-	Pattern  match.Matcher  `config:"pattern"`
-	Timeout  *time.Duration `config:"timeout"     validate:"positive"`
+	Negate       bool           `config:"negate"`
+	Match        string         `config:"match"       validate:"required"`
+	MaxLines     *int           `config:"max_lines"`
+	Pattern      match.Matcher  `config:"pattern"`
+	Timeout      *time.Duration `config:"timeout"     validate:"positive"`
+	FlushPattern *match.Matcher `config:"flush_pattern"`
 }
 
 func (c *MultilineConfig) Validate() error {

filebeat/harvester/reader/multiline_test.go

@@ -73,6 +73,40 @@ func TestMultilineBeforeNegateOK(t *testing.T) {
 	)
 }
 
+func TestMultilineAfterNegateOKFlushPattern(t *testing.T) {
+	flushMatcher := match.MustCompile(`EventEnd`)
+
+	testMultilineOK(t,
+		MultilineConfig{
+			Pattern:      match.MustCompile(`EventStart`),
+			Negate:       true,
+			Match:        "after",
+			FlushPattern: &flushMatcher,
+		},
+		3,
+		"EventStart\nEventId: 1\nEventEnd\n",
+		"OtherThingInBetween\n", // this should be a seperate event..
+		"EventStart\nEventId: 2\nEventEnd\n",
+	)
+}
+
+func TestMultilineAfterNegateOKFlushPatternWhereTheFirstLinesDosentMatchTheStartPattern(t *testing.T) {
+	flushMatcher := match.MustCompile(`EventEnd`)
+
+	testMultilineOK(t,
+		MultilineConfig{
+			Pattern:      match.MustCompile(`EventStart`),
+			Negate:       true,
+			Match:        "after",
+			FlushPattern: &flushMatcher,
+		},
+		3, //first two non-matching lines, will be merged to one event
+		"StartLineThatDosentMatchTheEvent\nOtherThingInBetween\n",
+		"EventStart\nEventId: 2\nEventEnd\n",
+		"EventStart\nEventId: 3\nEventEnd\n",
+	)
+}
+
 func TestMultilineBeforeNegateOKWithEmptyLine(t *testing.T) {
 	testMultilineOK(t,
 		MultilineConfig{
@@ -96,6 +130,7 @@ func testMultilineOK(t *testing.T, cfg MultilineConfig, events int, expected ...
 		if err != nil {
 			break
 		}
+
 		messages = append(messages, message)
 	}