From 19b2cd18cb6490371ad16a62bde86d3de4dda6e2 Mon Sep 17 00:00:00 2001
From: Pier-Hugues Pellerin <phpellerin@gmail.com>
Date: Tue, 30 Oct 2018 07:58:06 -0400
Subject: [PATCH] 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
(cherry picked from commit 8dbfed23ef8aac327007ed66b62519929fdd2f16)
---
 CHANGELOG.asciidoc                           |  1 +
 libbeat/beat/event.go                        |  3 ++
 libbeat/processors/dissect/processor.go      | 10 ++++
 libbeat/processors/dissect/processor_test.go | 56 ++++++++++++++++++++
 4 files changed, 70 insertions(+)

diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 585b9a90c6e9..053e3580c2ee 100644
--- a/CHANGELOG.asciidoc
+++ b/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*
 
diff --git a/libbeat/beat/event.go b/libbeat/beat/event.go
index 8b25349eb1eb..1191dff23d8d 100644
--- a/libbeat/beat/event.go
+++ b/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.
diff --git a/libbeat/processors/dissect/processor.go b/libbeat/processors/dissect/processor.go
index 613ea38dc104..eb6f29240b2c 100644
--- a/libbeat/processors/dissect/processor.go
+++ b/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
 	}
 
diff --git a/libbeat/processors/dissect/processor_test.go b/libbeat/processors/dissect/processor_test.go
index 132dae078729..7f7fbbb1b44d 100644
--- a/libbeat/processors/dissect/processor_test.go
+++ b/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)
+	})
+}