app/vlinsert/syslog: fix embedded newline in TCP syslog corrupting the next message

app/vlinsert/syslog/syslog.go

@@ -483,6 +483,22 @@ func (slr *syslogLineReader) Error() error {
 	return slr.err
 }
 
+// peekByte returns the next byte from slr without consuming it.
+//
+// false is returned on error. io.EOF is stored as is, so it is reported as a clean
+// stream end by Error(); any other error is wrapped for consistent diagnostics.
+func (slr *syslogLineReader) peekByte() (byte, bool) {
+	b, err := slr.br.Peek(1)
+	if err != nil {
+		if err != io.EOF {
+			err = fmt.Errorf("cannot read syslog frame: %w", err)
+		}
+		slr.err = err
+		return 0, false
+	}
+	return b[0], true
+}
+
 // nextLine reads the next syslog line from slr and stores it at slr.line.
 //
 // false is returned if the next line cannot be read. Error() must be called in this case
@@ -492,29 +508,32 @@ func (slr *syslogLineReader) nextLine() bool {
 		return false
 	}
 
-again:
-	prefix, err := slr.br.ReadSlice(' ')
-	if err != nil {
-		if err != io.EOF {
-			slr.err = fmt.Errorf("cannot read message frame prefix: %w", err)
+	// Skip frame delimiters (LF) between messages, including empty lines.
+	for {
+		b, ok := slr.peekByte()
+		if !ok {
 			return false
 		}
-		if len(prefix) == 0 {
-			slr.err = err
-			return false
+		if b != '\n' {
+			break
 		}
+		// The byte is buffered after a successful peek, so Discard cannot fail.
+		_, _ = slr.br.Discard(1)
 	}
-	// skip empty lines
-	for len(prefix) > 0 && prefix[0] == '\n' {
-		prefix = prefix[1:]
-	}
-	if len(prefix) == 0 {
-		// An empty prefix or a prefix with empty lines - try reading yet another prefix.
-		goto again
+
+	// Detect the framing method by the first byte without consuming it.
+	b, ok := slr.peekByte()
+	if !ok {
+		return false
 	}
 
-	if prefix[0] >= '0' && prefix[0] <= '9' {
+	if b >= '0' && b <= '9' {
 		// This is octet-counting method. See https://www.ietf.org/archive/id/draft-gerhards-syslog-plain-tcp-07.html#msgxfer
+		prefix, err := slr.br.ReadSlice(' ')
+		if err != nil {
+			slr.err = fmt.Errorf("cannot read message frame prefix: %w", err)
+			return false
+		}
 		msgLenStr := bytesutil.ToUnsafeString(prefix[:len(prefix)-1])
 		msgLen, err := strconv.ParseUint(msgLenStr, 10, 64)
 		if err != nil {
@@ -534,7 +553,7 @@ again:
 	}
 
 	// This is octet-stuffing method. See https://www.ietf.org/archive/id/draft-gerhards-syslog-plain-tcp-07.html#octet-stuffing-legacy
-	slr.line = append(slr.line[:0], prefix...)
+	slr.line = slr.line[:0]
 	for {
 		line, err := slr.br.ReadSlice('\n')
 		if err == nil {

app/vlinsert/syslog/syslog_test.go

@@ -34,8 +34,16 @@ func TestSyslogLineReader_Success(t *testing.T) {
 	f("\n\n\n", nil)
 
 	f("foobar", []string{"foobar"})
-	f("foobar\n", []string{"foobar\n"})
-	f("\n\nfoo\n\nbar\n\n", []string{"foo\n\nbar\n\n"})
+	f("foobar\n", []string{"foobar"})
+	f("\n\nfoo\n\nbar\n\n", []string{"foo", "bar"})
+
+	// An embedded LF splits the message into an extra fragment (")"), which must not
+	// leak into the following message. See https://github.com/VictoriaMetrics/VictoriaLogs/issues/1365
+	f("<78>foo bar 0x0001\n)\n<11>next message\n", []string{
+		"<78>foo bar 0x0001",
+		")",
+		"<11>next message",
+	})
 
 	f(`Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches...`, []string{"Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches..."})
 

docs/victorialogs/CHANGELOG.md

@@ -38,6 +38,7 @@ according to the following docs:
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): avoid `cannot connect to storage node at ...: EOF` errors after `vlselect` or `vlinsert` was idle for more than 60 seconds. See [#1440](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1440).
 * BUGFIX: [vlselect](https://docs.victoriametrics.com/victorialogs/cluster/): return `502 Bad Gateway` HTTP response code for incoming queries when one of the `vlstorage` nodes runs a VictoriaLogs version with an incompatible internal API instead of `400 Bad Request`. This is consistent with the `502 Bad Gateway` response returned when a `vlstorage` node is unavailable, and it allows building a proper failover scheme in high-availability setups. See [these docs](https://docs.victoriametrics.com/victorialogs/cluster/#high-availability).
 * 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: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): split syslog messages containing newline characters into separate log entries at every newline when they are ingested via TCP using octet-stuffing framing, so a message with an embedded newline no longer corrupts the parsing of the following messages. See [#1365](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1365).
 
 ## [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0)