[libbeat] Improve syslog parser/processor error handling

CHANGELOG.next.asciidoc

@@ -40,6 +40,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 *Affecting all Beats*
 
 - Allow loading secrets that contain commas from the keystore {pull}31694{pull}.
+- Improve syslog parser/processor error handling. {issue}31246[31246] {pull}31798[31798]
 - Fix Windows service timeouts when the "TCP/IP NetBIOS Helper" service is disabled. {issue}31810[31810] {pull}31835[31835]
 - Expand fields in `decode_json_fields` if target is set. {issue}31712[31712] {pull}32010[32010]
 

NOTICE.txt

@@ -15187,11 +15187,11 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/stretchr/testify
-Version: v1.7.0
+Version: v1.7.1
 Licence type (autodetected): MIT
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/stretchr/testify@v1.7.0/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/stretchr/testify@v1.7.1/LICENSE:
 
 MIT License
 

go.mod

@@ -120,7 +120,7 @@ require (
 	github.com/shopspring/decimal v1.2.0
 	github.com/spf13/cobra v1.3.0
 	github.com/spf13/pflag v1.0.5
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.7.1
 	github.com/tsg/go-daemon v0.0.0-20200207173439-e704b93fd89b
 	github.com/ugorji/go/codec v1.1.8
 	github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71

go.sum

@@ -1605,8 +1605,9 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.5.0/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/syndtr/gocapability v0.0.0-20170704070218-db04d3cc01c8/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
 github.com/syndtr/gocapability v0.0.0-20180916011248-d98352740cb2/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=

libbeat/processors/syslog/docs/syslog.asciidoc

@@ -121,6 +121,14 @@ The RFC 5424 format accepts the following forms of timestamps:
 
 Formats with an asterisk (*) are a non-standard allowance.
 
+[float]
+==== Structured Data
+
+For RFC 5424-formatted logs, if the structured data cannot be parsed according
+to RFC standards, the original structured data text will be prepended to the message
+field, separated by a space.
+
+
 ifndef::serverless[]
 
 [float]

libbeat/processors/syslog/syslog.go

@@ -172,14 +172,16 @@ func (p *processor) run(event *beat.Event) error {
 	fields, ts, err := syslog.ParseMessage(data, p.Format, p.TimeZone.Location())
 	if err != nil {
 		p.stats.Failure.Inc()
-		return err
+	} else {
+		p.stats.Success.Inc()
 	}
 
 	jsontransform.WriteJSONKeys(event, fields, false, p.OverwriteKeys, !p.IgnoreFailure)
-	event.Timestamp = ts
-	p.stats.Success.Inc()
+	if !ts.IsZero() {
+		event.Timestamp = ts
+	}
 
-	return nil
+	return err
 }
 
 // String will return a string representation of this processor (the configuration).

libbeat/processors/syslog/syslog_test.go

@@ -57,20 +57,20 @@ func mustParseTime(layout, value string, loc *time.Location) time.Time {
 }
 
 var syslogCases = map[string]struct {
-	Cfg      *conf.C
-	In       mapstr.M
-	Want     mapstr.M
-	WantTime time.Time
-	WantErr  bool
+	cfg      *conf.C
+	in       mapstr.M
+	want     mapstr.M
+	wantTime time.Time
+	wantErr  bool
 }{
 	"rfc-3164": {
-		Cfg: conf.MustNewConfigFrom(mapstr.M{
+		cfg: conf.MustNewConfigFrom(mapstr.M{
 			"timezone": "America/Chicago",
 		}),
-		In: mapstr.M{
+		in: mapstr.M{
 			"message": `<13>Oct 11 22:14:15 test-host su[1024]: this is the message`,
 		},
-		Want: mapstr.M{
+		want: mapstr.M{
 			"log": mapstr.M{
 				"syslog": mapstr.M{
 					"priority": 13,
@@ -89,14 +89,14 @@ var syslogCases = map[string]struct {
 			},
 			"message": "this is the message",
 		},
-		WantTime: mustParseTime(time.Stamp, "Oct 11 22:14:15", cfgtype.MustNewTimezone("America/Chicago").Location()),
+		wantTime: mustParseTime(time.Stamp, "Oct 11 22:14:15", cfgtype.MustNewTimezone("America/Chicago").Location()),
 	},
 	"rfc-5424": {
-		Cfg: conf.MustNewConfigFrom(mapstr.M{}),
-		In: mapstr.M{
+		cfg: conf.MustNewConfigFrom(mapstr.M{}),
+		in: mapstr.M{
 			"message": `<165>1 2003-10-11T22:14:15.003Z mymachine.example.com evntslog 1024 ID47 [exampleSDID@32473 iut="3" eventSource="Application" eventID="1011"][examplePriority@32473 class="high"] this is the message`,
 		},
-		Want: mapstr.M{
+		want: mapstr.M{
 			"log": mapstr.M{
 				"syslog": mapstr.M{
 					"priority": 165,
@@ -113,11 +113,11 @@ var syslogCases = map[string]struct {
 					"procid":   "1024",
 					"msgid":    "ID47",
 					"version":  "1",
-					"structured_data": map[string]map[string]string{
-						"examplePriority@32473": {
+					"structured_data": map[string]interface{}{
+						"examplePriority@32473": map[string]interface{}{
 							"class": "high",
 						},
-						"exampleSDID@32473": {
+						"exampleSDID@32473": map[string]interface{}{
 							"eventID":     "1011",
 							"eventSource": "Application",
 							"iut":         "3",
@@ -127,7 +127,7 @@ var syslogCases = map[string]struct {
 			},
 			"message": "this is the message",
 		},
-		WantTime: mustParseTime(time.RFC3339Nano, "2003-10-11T22:14:15.003Z", nil),
+		wantTime: mustParseTime(time.RFC3339Nano, "2003-10-11T22:14:15.003Z", nil),
 	},
 }
 
@@ -137,22 +137,22 @@ func TestSyslog(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
 
-			p, err := New(tc.Cfg)
+			p, err := New(tc.cfg)
 			if err != nil {
 				panic(err)
 			}
 			event := &beat.Event{
-				Fields: tc.In,
+				Fields: tc.in,
 			}
 
 			got, gotErr := p.Run(event)
-			if tc.WantErr {
+			if tc.wantErr {
 				assert.Error(t, gotErr)
 			} else {
 				assert.NoError(t, gotErr)
 			}
 
-			assert.Equal(t, tc.Want, got.Fields)
+			assert.Equal(t, tc.want, got.Fields)
 		})
 	}
 }
@@ -164,9 +164,9 @@ func BenchmarkSyslog(b *testing.B) {
 			b.ReportAllocs()
 			for i := 0; i < b.N; i++ {
 
-				p, _ := New(bc.Cfg)
+				p, _ := New(bc.cfg)
 				event := &beat.Event{
-					Fields: bc.In,
+					Fields: bc.in,
 				}
 
 				_, _ = p.Run(event)
@@ -177,46 +177,46 @@ func BenchmarkSyslog(b *testing.B) {
 
 func TestAppendStringField(t *testing.T) {
 	tests := map[string]struct {
-		InMap   mapstr.M
-		InField string
-		InValue string
-		Want    mapstr.M
+		inMap   mapstr.M
+		inField string
+		inValue string
+		want    mapstr.M
 	}{
 		"nil": {
-			InMap:   mapstr.M{},
-			InField: "error",
-			InValue: "foo",
-			Want: mapstr.M{
+			inMap:   mapstr.M{},
+			inField: "error",
+			inValue: "foo",
+			want: mapstr.M{
 				"error": "foo",
 			},
 		},
 		"string": {
-			InMap: mapstr.M{
+			inMap: mapstr.M{
 				"error": "foo",
 			},
-			InField: "error",
-			InValue: "bar",
-			Want: mapstr.M{
+			inField: "error",
+			inValue: "bar",
+			want: mapstr.M{
 				"error": []string{"foo", "bar"},
 			},
 		},
 		"string-slice": {
-			InMap: mapstr.M{
+			inMap: mapstr.M{
 				"error": []string{"foo", "bar"},
 			},
-			InField: "error",
-			InValue: "some value",
-			Want: mapstr.M{
+			inField: "error",
+			inValue: "some value",
+			want: mapstr.M{
 				"error": []string{"foo", "bar", "some value"},
 			},
 		},
 		"interface-slice": {
-			InMap: mapstr.M{
+			inMap: mapstr.M{
 				"error": []interface{}{"foo", "bar"},
 			},
-			InField: "error",
-			InValue: "some value",
-			Want: mapstr.M{
+			inField: "error",
+			inValue: "some value",
+			want: mapstr.M{
 				"error": []interface{}{"foo", "bar", "some value"},
 			},
 		},
@@ -226,9 +226,9 @@ func TestAppendStringField(t *testing.T) {
 		tc := tc
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
-			appendStringField(tc.InMap, tc.InField, tc.InValue)
+			appendStringField(tc.inMap, tc.inField, tc.inValue)
 
-			assert.Equal(t, tc.Want, tc.InMap)
+			assert.Equal(t, tc.want, tc.inMap)
 		})
 	}
 }

libbeat/reader/syslog/docs/syslog.asciidoc

@@ -69,3 +69,10 @@ The RFC 5424 format accepts the following forms of timestamps:
   ** `2003-10-11T22:14:15.123456-06:00`
 
 Formats with an asterisk (*) are a non-standard allowance.
+
+[float]
+==== Structured Data
+
+For RFC 5424-formatted logs, if the structured data cannot be parsed according
+to RFC standards, the original structured data text will be prepended to the message
+field, separated by a space.