Dissect tag on parsing error (#8751)

Before when a parsing error occurred the events was returned untouched
and an error was logged, if you don't look at your logs you have no
the idea that the tokenizer was not able to match your string.

Instead, when a parsing error occurs in the Dissect processor, we will now
add a tag named 'dissect_parsing_error' to the 'log.flags' field.
With that information, you are now able to reprocess your data or do
filtering on the UI.

Fixes: #8123

CHANGELOG.asciidoc

@@ -130,6 +130,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Add Beats Central Management {pull}8559[8559]
 - Allow Bus to buffer events in case listeners are not configured. {pull}8527[8527]
 - Enable `host` and `cloud` metadata processors by default. {pull}8596[8596]
+- Dissect will now flag event on parsing error. {pull}8751[8751]
 
 *Auditbeat*
 

libbeat/beat/event.go

@@ -24,6 +24,9 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 )
 
+// FlagField fields used to keep information or errors when events are parsed.
+const FlagField = "log.flags"
+
 // Event is the common event format shared by all beats.
 // Every event must have a timestamp and provide encodable Fields in `Fields`.
 // The `Meta`-fields can be used to pass additional meta-data to the outputs.

libbeat/processors/dissect/processor.go

@@ -27,6 +27,8 @@ import (
 	"github.com/elastic/beats/libbeat/processors"
 )
 
+const flagParsingError = "dissect_parsing_error"
+
 type processor struct {
 	config config
 }
@@ -60,6 +62,14 @@ func (p *processor) Run(event *beat.Event) (*beat.Event, error) {
 
 	m, err := p.config.Tokenizer.Dissect(s)
 	if err != nil {
+		if err := common.AddTagsWithKey(
+			event.Fields,
+			beat.FlagField,
+			[]string{flagParsingError},
+		); err != nil {
+			return event, errors.Wrap(err, "cannot add new flag the event")
+		}
+
 		return event, err
 	}
 

libbeat/processors/dissect/processor_test.go

@@ -176,3 +176,59 @@ func TestFieldAlreadyExist(t *testing.T) {
 		})
 	}
 }
+
+func TestErrorFlagging(t *testing.T) {
+	t.Run("when the parsing fails add a flag", func(t *testing.T) {
+		c, err := common.NewConfigFrom(map[string]interface{}{
+			"tokenizer": "%{ok} - %{notvalid}",
+		})
+
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		processor, err := newProcessor(c)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		e := beat.Event{Fields: common.MapStr{"message": "hello world"}}
+		event, err := processor.Run(&e)
+
+		if !assert.Error(t, err) {
+			return
+		}
+
+		flags, err := event.GetValue(beat.FlagField)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		assert.Contains(t, flags, flagParsingError)
+	})
+
+	t.Run("when the parsing is succesful do not add a flag", func(t *testing.T) {
+		c, err := common.NewConfigFrom(map[string]interface{}{
+			"tokenizer": "%{ok} %{valid}",
+		})
+
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		processor, err := newProcessor(c)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		e := beat.Event{Fields: common.MapStr{"message": "hello world"}}
+		event, err := processor.Run(&e)
+
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		_, err = event.GetValue(beat.FlagField)
+		assert.Error(t, err)
+	})
+}