Include log.source.address for unparseable syslog messages (elastic#1…

…5453) Source address was being included in syslog events, but if the syslog message failed to be parsed the event was being generated without this information. Add it in any case. Co-authored-by: Brian Candler <[email protected]>
leweafan · Jan 13, 2020 · f16fdb1 · f16fdb1
1 parent e7d4976
commit f16fdb1
Show file tree

Hide file tree

Showing 4 changed files with 116 additions and 26 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -482,6 +482,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add expand_event_list_from_field support in s3 input for reading json format AWS logs. {issue}15357[15357] {pull}15370[15370]
 - Add azure-eventhub input which will use the azure eventhub go sdk. {issue}14092[14092] {pull}14882[14882]
 - Expose more metrics of harvesters (e.g. `read_offset`, `start_time`). {pull}13395[13395]
+- Include log.source.address for unparseable syslog messages. {issue}13268[13268] {pull}15453[15453]
 - Release aws elb fileset as GA. {pull}15426[15426] {issue}15380[15380]
 - Release aws s3access fileset to GA. {pull}15431[15431] {issue}15430[15430]
 

diff --git a/filebeat/input/syslog/input.go b/filebeat/input/syslog/input.go
@@ -128,24 +128,8 @@ func NewInput(
 
 	forwarder := harvester.NewForwarder(out)
 	cb := func(data []byte, metadata inputsource.NetworkMetadata) {
-		ev := newEvent()
-		Parse(data, ev)
-		if !ev.IsValid() {
-			log.Errorw("can't parse event as syslog rfc3164", "message", string(data))
-			// On error revert to the raw bytes content, we need a better way to communicate this kind of
-			// error upstream this should be a global effort.
-			forwarder.Send(beat.Event{
-				Timestamp: time.Now(),
-				Meta: common.MapStr{
-					"truncated": metadata.Truncated,
-				},
-				Fields: common.MapStr{
-					"message": string(data),
-				},
-			})
-		} else {
-			forwarder.Send(createEvent(ev, metadata, time.Local, log))
-		}
+		ev := parseAndCreateEvent(data, metadata, time.Local, log)
+		forwarder.Send(ev)
 	}
 
 	server, err := factory(cb, config.Protocol)
@@ -201,11 +185,6 @@ func (p *Input) Wait() {
 func createEvent(ev *event, metadata inputsource.NetworkMetadata, timezone *time.Location, log *logp.Logger) beat.Event {
 	f := common.MapStr{
 		"message": strings.TrimRight(ev.Message(), "\n"),
-		"log": common.MapStr{
-			"source": common.MapStr{
-				"address": metadata.RemoteAddr.String(),
-			},
-		},
 	}
 
 	syslog := common.MapStr{}
@@ -254,13 +233,31 @@ func createEvent(ev *event, metadata inputsource.NetworkMetadata, timezone *time
 		f["event.sequence"] = ev.Sequence()
 	}
 
-	return beat.Event{
-		Timestamp: ev.Timestamp(timezone),
+	return newBeatEvent(ev.Timestamp(timezone), metadata, f)
+}
+
+func parseAndCreateEvent(data []byte, metadata inputsource.NetworkMetadata, timezone *time.Location, log *logp.Logger) beat.Event {
+	ev := newEvent()
+	Parse(data, ev)
+	if !ev.IsValid() {
+		log.Errorw("can't parse event as syslog rfc3164", "message", string(data))
+		return newBeatEvent(time.Now(), metadata, common.MapStr{
+			"message": string(data),
+		})
+	}
+	return createEvent(ev, metadata, time.Local, log)
+}
+
+func newBeatEvent(timestamp time.Time, metadata inputsource.NetworkMetadata, fields common.MapStr) beat.Event {
+	event := beat.Event{
+		Timestamp: timestamp,
 		Meta: common.MapStr{
 			"truncated": metadata.Truncated,
 		},
-		Fields: f,
+		Fields: fields,
 	}
+	event.Fields.Put("log.source.address", metadata.RemoteAddr.String())
+	return event
 }
 
 func mapValueToName(v int, m mapper) (string, error) {

diff --git a/filebeat/input/syslog/input_test.go b/filebeat/input/syslog/input_test.go
@@ -193,6 +193,58 @@ func TestSequence(t *testing.T) {
 	})
 }
 
+func TestParseAndCreateEvent(t *testing.T) {
+	cases := map[string]struct {
+		data     []byte
+		expected common.MapStr
+	}{
+		"valid data": {
+			data: []byte("<34>Oct 11 22:14:15 mymachine su[230]: 'su root' failed for lonvick on /dev/pts/8"),
+			expected: common.MapStr{
+				"event":    common.MapStr{"severity": 2},
+				"hostname": "mymachine",
+				"log": common.MapStr{
+					"source": common.MapStr{
+						"address": "127.0.0.1",
+					},
+				},
+				"message": "'su root' failed for lonvick on /dev/pts/8",
+				"process": common.MapStr{"pid": 230, "program": "su"},
+				"syslog": common.MapStr{
+					"facility":       4,
+					"facility_label": "security/authorization",
+					"priority":       34,
+					"severity_label": "Critical",
+				},
+			},
+		},
+
+		"invalid data": {
+			data: []byte("invalid"),
+			expected: common.MapStr{
+				"log": common.MapStr{
+					"source": common.MapStr{
+						"address": "127.0.0.1",
+					},
+				},
+				"message": "invalid",
+			},
+		},
+	}
+
+	tz := time.Local
+	log := logp.NewLogger("syslog")
+	metadata := dummyMetadata()
+
+	for title, c := range cases {
+		t.Run(title, func(t *testing.T) {
+			event := parseAndCreateEvent(c.data, metadata, tz, log)
+			assert.Equal(t, c.expected, event.Fields)
+			assert.Equal(t, metadata.Truncated, event.Meta["truncated"])
+		})
+	}
+}
+
 func dummyMetadata() inputsource.NetworkMetadata {
 	ip := "127.0.0.1"
 	parsedIP := net.ParseIP(ip)

diff --git a/filebeat/tests/system/test_syslog.py b/filebeat/tests/system/test_syslog.py
@@ -49,6 +49,46 @@ def test_syslog_with_tcp(self):
         self.assert_syslog(output[0])
         sock.close()
 
+    def test_syslog_with_tcp_invalid_message(self):
+        """
+        Test syslog input with invalid events from TCP.
+        """
+        host = "127.0.0.1"
+        port = 8080
+        input_raw = """
+- type: syslog
+  protocol:
+    tcp:
+        host: "{}:{}"
+"""
+
+        input_raw = input_raw.format(host, port)
+        self.render_config_template(
+            input_raw=input_raw,
+            inputs=False,
+        )
+
+        filebeat = self.start_beat()
+
+        self.wait_until(lambda: self.log_contains("Started listening for TCP connection"))
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # TCP
+        sock.connect((host, port))
+
+        for n in range(0, 2):
+            sock.send("invalid\n")
+
+        self.wait_until(lambda: self.output_count(lambda x: x >= 2))
+
+        filebeat.check_kill_and_wait()
+
+        output = self.read_output()
+
+        assert len(output) == 2
+        assert output[0]["message"] == "invalid"
+        assert len(output[0]["log.source.address"]) > 0
+        sock.close()
+
     def test_syslog_with_udp(self):
         """
         Test syslog input with events from TCP.