From e9e909577c5bad56a3c2346e6b296d5094560084 Mon Sep 17 00:00:00 2001 From: func25 Date: Fri, 10 Jul 2026 08:49:48 +0700 Subject: [PATCH] fix --- app/vlinsert/syslog/syslog.go | 74 +++++++++++++++++++++- app/vlinsert/syslog/syslog_test.go | 33 ++++++++++ docs/victorialogs/CHANGELOG.md | 1 + docs/victorialogs/data-ingestion/syslog.md | 25 ++++++-- 4 files changed, 125 insertions(+), 8 deletions(-) diff --git a/app/vlinsert/syslog/syslog.go b/app/vlinsert/syslog/syslog.go index 7735eee341..184c5a12d9 100644 --- a/app/vlinsert/syslog/syslog.go +++ b/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) { diff --git a/app/vlinsert/syslog/syslog_test.go b/app/vlinsert/syslog/syslog_test.go index 6f0f4525a3..fe27a00791 100644 --- a/app/vlinsert/syslog/syslog_test.go +++ b/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() diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index a5869b16fe..6ac7a40922 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/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) diff --git a/docs/victorialogs/data-ingestion/syslog.md b/docs/victorialogs/data-ingestion/syslog.md index 3fa042b4ae..85d78a36e8 100644 --- a/docs/victorialogs/data-ingestion/syslog.md +++ b/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 `` field value - `priority`, `facility` and `severity` - these fields are extracted from `` 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).