Add RFC5424 format support for syslog input (#23954) (#25119)

filebeat/input/syslog/config.go

@@ -36,13 +36,31 @@ import (
 
 type config struct {
 	harvester.ForwarderConfig `config:",inline"`
+	Format                    syslogFormat           `config:"format"`
 	Protocol                  common.ConfigNamespace `config:"protocol"`
 }
 
+type syslogFormat int
+
+const (
+	syslogFormatRFC3164 = iota
+	syslogFormatRFC5424
+	syslogFormatAuto
+)
+
+var (
+	syslogFormats = map[string]syslogFormat{
+		"rfc3164": syslogFormatRFC3164,
+		"rfc5424": syslogFormatRFC5424,
+		"auto":    syslogFormatAuto,
+	}
+)
+
 var defaultConfig = config{
 	ForwarderConfig: harvester.ForwarderConfig{
 		Type: "syslog",
 	},
+	Format: syslogFormatRFC3164,
 }
 
 type syslogTCP struct {
@@ -122,3 +140,12 @@ func factory(
 		return nil, fmt.Errorf("you must choose between TCP or UDP")
 	}
 }
+
+func (f *syslogFormat) Unpack(value string) error {
+	format, ok := syslogFormats[value]
+	if !ok {
+		return fmt.Errorf("invalid format '%s'", value)
+	}
+	*f = format
+	return nil
+}

filebeat/input/syslog/event.go

@@ -18,6 +18,7 @@
 package syslog
 
 import (
+	"bytes"
 	"math"
 	"time"
 )
@@ -72,8 +73,17 @@ type event struct {
 	year       int
 	loc        *time.Location
 	sequence   int
+
+	// RFC 5424
+	version   int
+	appName   string
+	msgID     string
+	processID string
+	data      EventData
 }
 
+type EventData map[string]map[string]string
+
 // newEvent() return a new event.
 func newEvent() *event {
 	return &event{
@@ -86,6 +96,7 @@ func newEvent() *event {
 		second:   -1,
 		year:     time.Now().Year(),
 		sequence: -1,
+		version:  -1,
 	}
 }
 
@@ -198,7 +209,12 @@ func (s *event) Year() int {
 
 // SetMessage sets the message.
 func (s *event) SetMessage(b []byte) {
-	s.message = string(b)
+	// remove BOM
+	if b[0] == 0xef && b[1] == 0xbb && b[2] == 0xbf {
+		s.message = string(b[3:])
+	} else {
+		s.message = string(b)
+	}
 }
 
 // Message returns the message.
@@ -298,6 +314,39 @@ func (s *event) Nanosecond() int {
 	return s.nanosecond
 }
 
+// SetVersion sets the version.
+func (s *event) SetVersion(version []byte) {
+	s.version = bytesToInt(version)
+}
+
+func (s *event) Version() int {
+	return s.version
+}
+
+func (s *event) SetAppName(appname []byte) {
+	s.appName = string(appname)
+}
+
+func (s *event) AppName() string {
+	return s.appName
+}
+
+func (s *event) SetMsgID(msgID []byte) {
+	s.msgID = string(msgID)
+}
+
+func (s *event) MsgID() string {
+	return s.msgID
+}
+
+func (s *event) SetProcID(processID []byte) {
+	s.processID = string(processID)
+}
+
+func (s *event) ProcID() string {
+	return s.processID
+}
+
 // Timestamp return the timestamp in UTC.
 func (s *event) Timestamp(timezone *time.Location) time.Time {
 	var t *time.Location
@@ -319,9 +368,39 @@ func (s *event) Timestamp(timezone *time.Location) time.Time {
 	).UTC()
 }
 
+func (s *event) IsDataEmpty() bool {
+	if s.data == nil {
+		return true
+	}
+	return len(s.data) == 0
+}
+
 // IsValid returns true if the date and the message are present.
 func (s *event) IsValid() bool {
-	return s.day != -1 && s.hour != -1 && s.minute != -1 && s.second != -1 && s.message != ""
+	return s.day != -1 && s.hour != -1 && s.minute != -1 && s.second != -1 && (s.message != "" || !s.IsDataEmpty())
+}
+
+func (s *event) SetData(id string, key string, data []byte, start int, end int, bs []int) {
+	var v string
+
+	// param value escape
+	// https://tools.ietf.org/html/rfc5424#section-6.3.3
+	if len(bs) > 0 {
+		buf := bytes.NewBufferString("")
+		for _, i := range bs {
+			buf.Write(data[start:i])
+			start = i + 1
+		}
+		if start <= end {
+			buf.Write(data[start:end])
+		}
+		v = buf.String()
+	} else {
+		v = string(data[start:end])
+	}
+	if element, ok := s.data[id]; ok {
+		element[key] = v
+	}
 }
 
 // BytesToInt takes a variable length of bytes and assume ascii chars and convert it to int, this is