Cherry-pick #8751 to 6.x: Dissect tag on parsing error

CHANGELOG.asciidoc

@@ -119,6 +119,7 @@ https://github.com/elastic/beats/compare/v6.4.0...6.x[Check the HEAD diff]
 - Add Beats Central Management {pull}8559[8559]
 - Report configured queue type. {pull}8091[8091]
 - 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)
+	})
+}