lib/logstorage: follow-up for the commit d378ec6

- Make sure that the JSONParser lifetime is bigger than the lifetime of the @Cee fields parsed with it.
- Add support for parsing @Cee message inside RFC5424 protocol.
- Document the added functionality in the Syslog data ingestion docs - https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#cee
- Mention the added functionality in the https://docs.victoriametrics.com/victorialogs/changelog/

Updates #842

Author: valyala

docs/victorialogs/CHANGELOG.md

@@ -21,6 +21,8 @@ according to the follosing docs:
 
 ## tip
 
+* FEATURE: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): add support for automatic parsing of [`@cee` messages](http://cee.mitre.org/language/1.0-beta1/clt.html#syslog). Thanks to @exherb for the pull request [#842](https://github.com/VictoriaMetrics/VictoriaLogs/pull/842).
+
 * BUGFIX: [delete API](https://docs.victoriametrics.com/victorialogs/#how-to-delete-logs): prevent from possible fatal error (panic) at `block_stream_merger.go:237` during the deletion of the logs. See [#825](https://github.com/VictoriaMetrics/VictoriaLogs/issues/825).
 
 ## [v1.38.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.38.0)

docs/victorialogs/data-ingestion/syslog.md

@@ -197,6 +197,11 @@ VictoriaLogs automatically parses [CEF Syslog messages for SIEM](https://www.mic
 
 An optional `extension` fields are parsed into `cef.extension.<key>=<value>` fields.
 
+## CEE
+
+VictoriaLogs automatically parses [`@cee` messages](http://cee.mitre.org/language/1.0-beta1/clt.html#syslog) into distinct log fields.
+See [these docs](https://www.rsyslog.com/json-elasticsearch/) for more details.
+
 ## Dropping fields
 
 VictoriaLogs supports `-syslog.ignoreFields.tcp`, `-syslog.ignoreFields.udp` and `-syslog.ignoreFields.unix` command-line flags for skipping

lib/logstorage/json_parser.go

@@ -12,12 +12,12 @@ import (
 //
 // See https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model
 //
-// Use GetParser() for obtaining the parser.
+// Use GetJSONParser() for obtaining the parser.
 type JSONParser struct {
 	// Fields contains the parsed JSON line after Parse() call
 	//
 	// The Fields are valid until the next call to ParseLogMessage()
-	// or until the parser is returned to the pool with PutParser() call.
+	// or until the parser is returned to the pool with PutJSONParser() call.
 	Fields []Field
 
 	// p is used for fast JSON parsing

lib/logstorage/syslog_parser.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
 
 // GetSyslogParser returns syslog parser from the pool.
@@ -61,6 +60,11 @@ type SyslogParser struct {
 	// See https://datatracker.ietf.org/doc/html/rfc5424#section-6.3
 	sdParser logfmtParser
 
+	// jsonParser is used for parsing CEE messages.
+	//
+	// See https://cee.mitre.org/language/1.0-beta1/clt.html#syslog
+	jsonParser *JSONParser
+
 	// currentYear is used as the current year for rfc3164 messages.
 	currentYear int
 
@@ -72,6 +76,11 @@ type SyslogParser struct {
 }
 
 func (p *SyslogParser) reset() {
+	if p.jsonParser != nil {
+		PutJSONParser(p.jsonParser)
+		p.jsonParser = nil
+	}
+
 	p.currentYear = 0
 	p.timezone = nil
 	p.resetFields()
@@ -83,30 +92,44 @@ func (p *SyslogParser) resetFields() {
 
 	p.buf = p.buf[:0]
 	p.sdParser.reset()
+
+	if p.jsonParser != nil {
+		p.jsonParser.reset()
+	}
 }
 
 func (p *SyslogParser) AddMessageField(s string) {
-	if strings.HasPrefix(s, "CEF:") {
-		fields := p.Fields
-		if p.parseCEFMessage(strings.TrimPrefix(s, "CEF:")) {
-			return
-		}
+	fields := p.Fields
+	if !p.parseSpecialMessage(s) {
 		p.Fields = fields
+		p.AddField("message", s)
 	}
+}
 
-	if strings.HasPrefix(s, "@cee:") {
-		jsonP := GetJSONParser()
-		defer PutJSONParser(jsonP)
-		jsonStr := strings.TrimPrefix(s, "@cee:")
-		if err := jsonP.ParseLogMessage(bytesutil.ToUnsafeBytes(jsonStr)); err == nil {
-			p.Fields = append(p.Fields, jsonP.Fields...)
-			return
-		} else {
-			logger.Warnf("parse @cee json fail, %s", jsonStr)
-		}
+func (p *SyslogParser) parseSpecialMessage(s string) bool {
+	switch {
+	case strings.HasPrefix(s, "CEF:"):
+		cefStr := strings.TrimPrefix(s, "CEF:")
+		return p.parseCEFMessage(cefStr)
+	case strings.HasPrefix(s, "@cee:"):
+		ceeStr := strings.TrimPrefix(s, "@cee:")
+		return p.parseCEEMessage(ceeStr)
+	default:
+		return false
 	}
+}
 
-	p.AddField("message", s)
+// parseCEEMessage parses CEE message. See https://cee.mitre.org/language/1.0-beta1/clt.html#syslog
+func (p *SyslogParser) parseCEEMessage(s string) bool {
+	if p.jsonParser == nil {
+		p.jsonParser = GetJSONParser()
+	}
+	err := p.jsonParser.ParseLogMessage(bytesutil.ToUnsafeBytes(s))
+	if err != nil {
+		return false
+	}
+	p.Fields = append(p.Fields, p.jsonParser.Fields...)
+	return true
 }
 
 // AddField adds name=value log field to p.Fields.
@@ -328,7 +351,10 @@ func (p *SyslogParser) parseRFC5424(s string) {
 
 func (p *SyslogParser) parseRFC5424SD(s string) (string, bool) {
 	if strings.HasPrefix(s, "- ") {
-		return s[2:], true
+		return s[len("- "):], true
+	}
+	if strings.HasPrefix(s, "@cee:") {
+		return s, true
 	}
 
 	for {

lib/logstorage/syslog_parser_test.go

@@ -123,9 +123,9 @@ func TestSyslogParser(t *testing.T) {
 	f(`Sep 29 08:26:10 host`, time.UTC, `format=rfc3164 timestamp=2024-09-29T08:26:10Z hostname=host`)
 	f(`Sep 29 08:26:10`, time.UTC, `format=rfc3164 timestamp=2024-09-29T08:26:10Z`)
 
-	// @cee
-	f(`Jun  3 12:08:33 abcd systemd[1]: @cee: {"k":"v","message":"test"}`, time.UTC,
-		`format=rfc3164 timestamp=2024-06-03T12:08:33Z hostname=abcd app_name=systemd proc_id=1 k=v message=test`)
-	f(`Jun  3 12:08:33 abcd systemd[1]: @cee: {"k":"v","message":"two words"}`, time.UTC,
-		`format=rfc3164 timestamp=2024-06-03T12:08:33Z hostname=abcd app_name=systemd proc_id=1 k=v message="two words"`)
+	// @cee - https://cee.mitre.org/language/1.0-beta1/clt.html#syslog
+	f(`Jun  3 12:08:33 abcd systemd[1]: @cee: {"k":"v","message":"test"}`, time.UTC, `format=rfc3164 timestamp=2024-06-03T12:08:33Z hostname=abcd app_name=systemd proc_id=1 k=v message=test`)
+	f(`Jun  3 12:08:33 abcd systemd[1]: @cee: {"k":"v","message":"two words"}`, time.UTC, `format=rfc3164 timestamp=2024-06-03T12:08:33Z hostname=abcd app_name=systemd proc_id=1 k=v message="two words"`)
+	f(`<0>Dec 20 12:42:20 syslog-relay process[35]: @cee: {"crit":123,"id":"abc","appname":"application","pname":"auth","pid":123,"host":"system.example.com","pri":10,"time":"2011-12-20T12:38:05.123456-05:00","action":"login","domain":"app","object":"account","service":"web","status":"success"}`, time.UTC, `priority=0 facility_keyword=kern level=emerg facility=0 severity=0 format=rfc3164 timestamp=2024-12-20T12:42:20Z hostname=syslog-relay app_name=process proc_id=35 crit=123 id=abc appname=application pname=auth pid=123 host=system.example.com pri=10 time=2011-12-20T12:38:05.123456-05:00 action=login domain=app object=account service=web status=success`)
+	f(`<165>1 2011-12-20T12:38:06Z 10.10.0.1 process - example-event-1 @cee:{"pname":"auth","host":"system.example.com","time":"2011-12-20T12:38:05.123456-05:00"}`, time.UTC, `priority=165 facility_keyword=local4 level=notice facility=20 severity=5 format=rfc5424 timestamp=2011-12-20T12:38:06Z hostname=10.10.0.1 app_name=process proc_id=- msg_id=example-event-1 pname=auth host=system.example.com time=2011-12-20T12:38:05.123456-05:00`)
 }