app/vlinsert/syslog: preserve multiline messages in datagrams

app/vlinsert/syslog/syslog.go

@@ -42,6 +42,13 @@ var (
 	listenAddrUnix = flagutil.NewArrayString("syslog.listenAddr.unix", "Comma-separated list of Unix socket filepaths to listen to for Syslog messages. "+
 		"Filepaths may be prepended with 'unixgram:'  for listening for SOCK_DGRAM sockets. By default SOCK_STREAM sockets are used. "+
 		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/")
+	framingUDP = flagutil.NewArrayString("syslog.framing.udp", "Message framing for the corresponding -syslog.listenAddr.udp. "+
+		"Supported values: datagram, newline. The default datagram mode treats every UDP packet as a single Syslog message. "+
+		"The newline mode splits every UDP packet into messages delimited by newline characters. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#message-framing")
+	framingUnix = flagutil.NewArrayString("syslog.framing.unix", "Message framing for SOCK_DGRAM Unix sockets at the corresponding -syslog.listenAddr.unix. "+
+		"Supported values: datagram, newline. The default datagram mode treats every packet as a single Syslog message. "+
+		"The newline mode splits every packet into messages delimited by newline characters. This flag doesn't apply to SOCK_STREAM Unix sockets. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#message-framing")
 
 	tlsEnable = flagutil.NewArrayBool("syslog.tls", "Whether to enable TLS for receiving syslog messages at the corresponding -syslog.listenAddr.tcp. "+
 		"The corresponding -syslog.tlsCertFile and -syslog.tlsKeyFile must be set if -syslog.tls is set. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#security")
@@ -193,11 +200,14 @@ func runUnixListener(addr string, argIdx int) {
 	if err != nil {
 		logger.Fatalf("cannot parse configs for -syslog.listenAddr.unix=%q: %s", addr, err)
 	}
-
 	laddr := getUnixSocketNetworkAndPath(addr)
 	if laddr.Net == "unix" {
 		runUnixStreamListener(laddr, cfg)
 	} else {
+		cfg.packetFraming, err = parsePacketFraming(framingUnix.GetOptionalArg(argIdx))
+		if err != nil {
+			logger.Fatalf("cannot parse -syslog.framing.unix for -syslog.listenAddr.unix=%q: %s", addr, err)
+		}
 		runUnixPacketListener(laddr, cfg)
 	}
 }
@@ -269,6 +279,10 @@ func runUDPListener(addr string, argIdx int) {
 	if err != nil {
 		logger.Fatalf("cannot parse configs for -syslog.listenAddr.udp=%q: %s", addr, err)
 	}
+	cfg.packetFraming, err = parsePacketFraming(framingUDP.GetOptionalArg(argIdx))
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.framing.udp for -syslog.listenAddr.udp=%q: %s", addr, err)
+	}
 
 	doneCh := make(chan struct{})
 	go func() {
@@ -356,7 +370,7 @@ func servePacketListener(ln net.PacketConn, cfg *configs) {
 
 				remoteIP := getRemoteIP(remoteAddr, cfg.useRemoteIP)
 
-				if err := processStream(cfg.typ, bb.NewReader(), cfg.compressMethod, cfg.useLocalTimestamp, remoteIP, cp); err != nil {
+				if err := processPacket(cfg.typ, bb.NewReader(), cfg.compressMethod, cfg.packetFraming, cfg.useLocalTimestamp, remoteIP, cp); err != nil {
 					logger.Errorf("syslog: cannot process %s data from %s at %s: %s", cfg.typ, remoteAddr, localAddr, err)
 				}
 			}
@@ -411,7 +425,7 @@ func serveStreamListener(ln net.Listener, cfg *configs) {
 	wg.Wait()
 }
 
-// processStream parses a stream of syslog messages from r and ingests them into vlstorage.
+// processStream ingests syslog messages from r using octet-counting or newline-delimited framing.
 func processStream(protocol string, r io.Reader, compressMethod string, useLocalTimestamp bool, remoteIP string, cp *insertutil.CommonParams) error {
 	if err := insertutil.CanWriteData(); err != nil {
 		return err
@@ -462,6 +476,42 @@ func processUncompressedStream(r io.Reader, useLocalTimestamp bool, remoteIP str
 	return slr.Error()
 }
 
+// processPacket ingests a datagram as a single syslog message instead of splitting it by newline,
+// unless newline framing is configured.
+func processPacket(protocol string, r io.Reader, compressMethod, packetFraming string, useLocalTimestamp bool, remoteIP string, cp *insertutil.CommonParams) error {
+	if err := insertutil.CanWriteData(); err != nil {
+		return err
+	}
+
+	isStreamMode := packetFraming == packetFramingNewline
+	lmp := cp.NewLogMessageProcessor("syslog_"+protocol, isStreamMode)
+	err := processPacketInternal(r, compressMethod, packetFraming, useLocalTimestamp, remoteIP, lmp)
+	lmp.MustClose()
+
+	return err
+}
+
+func processPacketInternal(r io.Reader, compressMethod, packetFraming string, useLocalTimestamp bool, remoteIP string, lmp insertutil.LogMessageProcessor) error {
+	if packetFraming == packetFramingNewline {
+		return processStreamInternal(r, compressMethod, useLocalTimestamp, remoteIP, lmp)
+	}
+	return processDatagramInternal(r, compressMethod, useLocalTimestamp, remoteIP, lmp)
+}
+
+func processDatagramInternal(r io.Reader, compressMethod string, useLocalTimestamp bool, remoteIP string, lmp insertutil.LogMessageProcessor) error {
+	return protoparserutil.ReadUncompressedData(r, compressMethod, insertutil.MaxLineSizeBytes, func(data []byte) error {
+		if len(data) == 0 {
+			return nil
+		}
+		currentYear := int(globalCurrentYear.Load())
+		err := processLine(data, currentYear, globalTimezone, useLocalTimestamp, remoteIP, lmp)
+		if err != nil {
+			errorsTotal.Inc()
+		}
+		return err
+	})
+}
+
 type syslogLineReader struct {
 	line []byte
 
@@ -674,10 +724,28 @@ type configs struct {
 	extraFields       []logstorage.Field
 	tenantID          logstorage.TenantID
 	compressMethod    string
+	packetFraming     string
 	useLocalTimestamp bool
 	useRemoteIP       bool
 }
 
+const (
+	packetFramingDatagram = "datagram"
+	packetFramingNewline  = "newline"
+)
+
+func parsePacketFraming(s string) (string, error) {
+	if s == "" {
+		return packetFramingDatagram, nil
+	}
+	switch s {
+	case packetFramingDatagram, packetFramingNewline:
+		return s, nil
+	default:
+		return "", fmt.Errorf("unsupported framing %q; supported values: %q, %q", s, packetFramingDatagram, packetFramingNewline)
+	}
+}
+
 func getConfigs(typ string, argIdx int, streamFieldsArg, ignoreFieldsArg, decolorizeFieldsArg, extraFieldsArg, tenantIDArg, compressMethodArg *flagutil.ArrayString,
 	useLocalTimestampArg, useRemoteIPArg *flagutil.ArrayBool,
 ) (*configs, error) {

app/vlinsert/syslog/syslog_test.go

@@ -9,6 +9,39 @@ import (
 	"github.com/VictoriaMetrics/VictoriaLogs/app/vlinsert/insertutil"
 )
 
+func TestProcessPacketInternal_Success(t *testing.T) {
+	f := func(data, packetFraming string, currentYear int, timestampsExpected []int64, resultExpected string) {
+		t.Helper()
+
+		MustInit()
+		defer MustStop()
+
+		globalTimezone = time.UTC
+		globalCurrentYear.Store(int64(currentYear))
+
+		tlp := &insertutil.TestLogMessageProcessor{}
+		if err := processPacketInternal(bytes.NewBufferString(data), "", packetFraming, false, "1.2.3.4", tlp); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// packet framing datagram
+	timestamp1 := time.Date(2026, 7, 3, 16, 36, 17, 0, time.UTC).UnixNano()
+	f("<133>1 2026-07-03T16:36:17Z switch.example.com - - - - \r\n  ALARM 12345\r\n  --- END", packetFramingDatagram, 2026, []int64{timestamp1},
+		`{"priority":"133","facility_keyword":"local0","level":"notice","facility":"16","severity":"5","format":"rfc5424","hostname":"switch.example.com","app_name":"-","proc_id":"-","msg_id":"-","_msg":"\r\n  ALARM 12345\r\n  --- END","remote_ip":"1.2.3.4"}`)
+
+	// packet framing newline
+	data := `<133>1 2026-07-03T16:36:17Z host-1 app - - - message-1
+<134>1 2026-07-03T16:36:18Z host-2 app - - - message-2`
+	timestamp2 := time.Date(2026, 7, 3, 16, 36, 18, 0, time.UTC).UnixNano()
+	f(data, packetFramingNewline, 2026, []int64{timestamp1, timestamp2},
+		`{"priority":"133","facility_keyword":"local0","level":"notice","facility":"16","severity":"5","format":"rfc5424","hostname":"host-1","app_name":"app","proc_id":"-","msg_id":"-","_msg":"message-1","remote_ip":"1.2.3.4"}
+{"priority":"134","facility_keyword":"local0","level":"info","facility":"16","severity":"6","format":"rfc5424","hostname":"host-2","app_name":"app","proc_id":"-","msg_id":"-","_msg":"message-2","remote_ip":"1.2.3.4"}`)
+}
+
 func TestSyslogLineReader_Success(t *testing.T) {
 	f := func(data string, linesExpected []string) {
 		t.Helper()

docs/victorialogs/CHANGELOG.md

@@ -46,6 +46,7 @@ according to the following docs:
 * BUGFIX: [vlselect](https://docs.victoriametrics.com/victorialogs/cluster/): properly register `vmalert.proxyURL` as a secret flag. Previously, this flag was registered after logger initialization and printed in plaintext on application startup.
 * BUGFIX: [multi-level cluster setup](https://docs.victoriametrics.com/victorialogs/cluster/#multi-level-cluster-setup): properly return `502 Bad Gateway` HTTP response code when a `vlselect` node queries other `vlselect` nodes and the underlying `vlstorage` is unavailable, as described at [high availability](https://docs.victoriametrics.com/victorialogs/cluster/#high-availability) docs. This allows configuring proper failover schemes to a healthy cluster.
 * BUGFIX: [`/select/tenant_ids`](https://docs.victoriametrics.com/victorialogs/querying/#querying-tenants): return tenant ids in a deterministic (sorted) order. Previously the order was random across requests. See [#1575](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1575).
+* BUGFIX: [Syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#message-framing): treat every UDP and `SOCK_DGRAM` Unix packet as a single Syslog message by default, so RFC5424 message bodies can contain newline characters. Set `-syslog.framing.udp=newline` or `-syslog.framing.unix=newline` to retain the previous newline-delimited packet behavior. See [#1573](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1573).
 
 ## [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0)
 

docs/victorialogs/data-ingestion/syslog.md

@@ -42,11 +42,15 @@ VictoriaLogs can accept logs from the following syslog collectors:
 - [Rsyslog](https://www.rsyslog.com/). See [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#rsyslog).
 - [Syslog-ng](https://www.syslog-ng.com/). See [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#syslog-ng).
 
-Multiple logs in Syslog format can be ingested via a single TCP connection or via a single UDP packet - just put every log on a separate line
-and delimit them with `\n` char.
+Multiple logs in Syslog format can be ingested via a single TCP connection by putting every log on a separate line and delimiting them with `\n` char.
+
+By default, every UDP packet is treated as a single Syslog message according to [RFC 5426](https://www.rfc-editor.org/rfc/rfc5426.html#section-3.1).
+Every `SOCK_DGRAM` Unix packet is handled in the same way. This preserves newline characters inside the message.
+See [message framing](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#message-framing)
+for how to ingest multiple newline-delimited messages from a single packet.
 
 VictoriaLogs automatically extracts the following [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model)
-from the received Syslog lines:
+from the received Syslog messages:
 
 - [`_time`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field) - log timestamp. See also [log timestamps](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#log-timestamps)
 - [`_msg`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field) - the `MESSAGE` field from the supported syslog formats above
@@ -55,8 +59,8 @@ from the received Syslog lines:
 - `level` - string representation of the log level according to the `<PRI>` field value
 - `priority`, `facility` and `severity` - these fields are extracted from `<PRI>` field
 - `facility_keyword` - string representation of the `facility` field according to [these docs](https://en.wikipedia.org/wiki/Syslog#Facility)
-- `format` - this field is set to either `rfc3164` or `rfc5424` depending on the format of the parsed syslog line
-- `msg_id` - `MSGID` field from log line in `RFC5424` format.
+- `format` - this field is set to either `rfc3164` or `rfc5424` depending on the parsed Syslog message format
+- `msg_id` - `MSGID` field from a Syslog message in `RFC5424` format.
 
 The `[STRUCTURED-DATA]` is parsed into fields with the `SD-ID.param1`, `SD-ID.param2`, ..., `SD-ID.paramN` names and the corresponding values
 according to [the specification](https://datatracker.ietf.org/doc/html/rfc5424#section-6.3).
@@ -82,6 +86,7 @@ curl http://localhost:9428/select/logsql/query -d 'query=_time:5m'
 See also:
 
 - [Log timestamps](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#log-timestamps)
+- [Message framing](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#message-framing)
 - [Security](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#security)
 - [Compression](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#compression)
 - [Multitenancy](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#multitenancy)
@@ -93,6 +98,16 @@ See also:
 - [Data ingestion troubleshooting](https://docs.victoriametrics.com/victorialogs/data-ingestion/#troubleshooting).
 - [How to query VictoriaLogs](https://docs.victoriametrics.com/victorialogs/querying/).
 
+## Message framing
+
+By default, every packet received at `-syslog.listenAddr.udp` or at a `unixgram:` address in `-syslog.listenAddr.unix` is treated as a single Syslog message.
+This allows the message to contain newline characters.
+
+Set `-syslog.framing.udp=newline` for the corresponding UDP listen address in order to split every received UDP packet into multiple Syslog messages
+at newline characters. Set `-syslog.framing.unix=newline` to enable the same behavior for the corresponding `unixgram:` address.
+The framing flags are arrays, so each listen address can use an independent mode. The supported values are `datagram` and `newline`.
+The framing flags do not apply to TCP or `SOCK_STREAM` Unix sockets.
+
 ## Log timestamps
 
 By default VictoriaLogs uses the timestamp from the parsed Syslog message as [`_time` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field).