From bf5a3f8d63ebc1f9c847851b1a9f27f771fb4dad Mon Sep 17 00:00:00 2001 From: hovavza <147598197+hovavza@users.noreply.github.com> Date: Fri, 10 Nov 2023 17:01:17 +0200 Subject: [PATCH] udp-receiver async - fix data corruption (with buffer pools) (#28898) **Description:** Fixing a bug in udp async mode only (but didn't affect the default non-async mode). Udp-receiver reuses the same buffer when each packet is processed. While that's working fine when running it without async config, it cause a significant amount of duplicate packets and corrupted packets being sent downstream. The reader-async thread is reading a packet from the udp port into the buffer, places that buffer in the channel, and reads another packet into the same buffer and pushes it to the channel. Let's say that the processor-async thread was a bit slow, so it only tries to read from the channel after the 2 items were placed in the channel. In that case, the processor thread will read 2 items from the channel, but it will be the same 2nd packet (since the 1st one was overwritten). In some cases, it seems the processor is reading a corrupted buffer (since the reader is currently writing into it). We can't fix it by having the reader allocate a new buffer before each time it reads a packet from the udp port, since that hurts performance significantly (reducing it up to ~50%). Instead, use a pool so the buffers are reused. Before reading a packet, the reader get a buffer from the pool. The processor returns it back to the pool after it has been successfully processed **Link to tracking Issue:** 27613 **Testing:** Ran existing unitests. Ran ran stress tests (sending 250k udp packets per second) duplicate/corruption issue didn't happen; performance wasn't hurt. **Documentation:** None --- ...a-input-udp-async-fix-data-corruption.yaml | 27 +++++++++ pkg/stanza/operator/input/udp/udp.go | 59 +++++++++++++------ 2 files changed, 67 insertions(+), 19 deletions(-) create mode 100644 .chloggen/pkg-stanza-input-udp-async-fix-data-corruption.yaml diff --git a/.chloggen/pkg-stanza-input-udp-async-fix-data-corruption.yaml b/.chloggen/pkg-stanza-input-udp-async-fix-data-corruption.yaml new file mode 100644 index 000000000000..0bcf8f4a5e1f --- /dev/null +++ b/.chloggen/pkg-stanza-input-udp-async-fix-data-corruption.yaml @@ -0,0 +1,27 @@ +# Use this changelog template to create an entry for release notes. + +# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' +change_type: bug_fix + +# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver) +component: pkg/stanza + +# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). +note: Fix data-corruption/race-condition issue in udp async (reuse of buffer); use buffer pool isntead. + +# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. +issues: [27613] + +# (Optional) One or more lines of additional information to render under the primary note. +# These lines will be padded with 2 spaces and then inserted directly into the document. +# Use pipe (|) for multiline entries. +subtext: + +# If your change doesn't affect end users or the exported elements of any package, +# you should instead start your pull request title with [chore] or use the "Skip Changelog" label. +# Optional: The change log or logs in which this entry should be included. +# e.g. '[user]' or '[user, api]' +# Include 'user' if the change is relevant to end users. +# Include 'api' if there is a change to a library API. +# Default: '[user]' +change_logs: [] diff --git a/pkg/stanza/operator/input/udp/udp.go b/pkg/stanza/operator/input/udp/udp.go index 4593f9ead5e5..ac7126f0eec8 100644 --- a/pkg/stanza/operator/input/udp/udp.go +++ b/pkg/stanza/operator/input/udp/udp.go @@ -138,6 +138,12 @@ func (c Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) { if c.AsyncConfig != nil { udpInput.messageQueue = make(chan messageAndAddress, c.AsyncConfig.MaxQueueLength) + udpInput.readBufferPool = sync.Pool{ + New: func() interface{} { + buffer := make([]byte, MaxUDPSize) + return &buffer + }, + } } return udpInput, nil } @@ -159,13 +165,15 @@ type Input struct { splitFunc bufio.SplitFunc resolver *helper.IPResolver - messageQueue chan messageAndAddress - stopOnce sync.Once + messageQueue chan messageAndAddress + readBufferPool sync.Pool + stopOnce sync.Once } type messageAndAddress struct { - Message []byte - RemoteAddr net.Addr + Message *[]byte + RemoteAddr net.Addr + MessageLength int } // Start will start listening for messages on a socket. @@ -206,9 +214,12 @@ func (u *Input) readAndProcessMessages(ctx context.Context) { defer u.wg.Done() dec := decode.New(u.encoding) - buf := make([]byte, 0, MaxUDPSize) + readBuffer := make([]byte, MaxUDPSize) + scannerBuffer := make([]byte, 0, MaxUDPSize) for { - message, remoteAddr, err := u.readMessage() + message, remoteAddr, bufferLength, err := u.readMessage(readBuffer) + message = u.removeTrailingCharactersAndNULsFromBuffer(message, bufferLength) + if err != nil { select { case <-ctx.Done(): @@ -219,11 +230,11 @@ func (u *Input) readAndProcessMessages(ctx context.Context) { break } - u.processMessage(ctx, message, remoteAddr, dec, buf) + u.processMessage(ctx, message, remoteAddr, dec, scannerBuffer) } } -func (u *Input) processMessage(ctx context.Context, message []byte, remoteAddr net.Addr, dec *decode.Decoder, buf []byte) { +func (u *Input) processMessage(ctx context.Context, message []byte, remoteAddr net.Addr, dec *decode.Decoder, scannerBuffer []byte) { if u.OneLogPerPacket { log := truncateMaxLog(message) u.handleMessage(ctx, remoteAddr, dec, log) @@ -231,7 +242,7 @@ func (u *Input) processMessage(ctx context.Context, message []byte, remoteAddr n } scanner := bufio.NewScanner(bytes.NewReader(message)) - scanner.Buffer(buf, MaxUDPSize) + scanner.Buffer(scannerBuffer, MaxUDPSize) scanner.Split(u.splitFunc) @@ -247,8 +258,10 @@ func (u *Input) readMessagesAsync(ctx context.Context) { defer u.wg.Done() for { - message, remoteAddr, err := u.readMessage() + readBuffer := u.readBufferPool.Get().(*[]byte) // Can't reuse the same buffer since same references would be written multiple times to the messageQueue (and cause data override of previous entries) + message, remoteAddr, bufferLength, err := u.readMessage(*readBuffer) if err != nil { + u.readBufferPool.Put(readBuffer) select { case <-ctx.Done(): return @@ -259,8 +272,9 @@ func (u *Input) readMessagesAsync(ctx context.Context) { } messageAndAddr := messageAndAddress{ - Message: message, - RemoteAddr: remoteAddr, + Message: &message, + MessageLength: bufferLength, + RemoteAddr: remoteAddr, } // Send the message to the message queue for processing @@ -272,7 +286,7 @@ func (u *Input) processMessagesAsync(ctx context.Context) { defer u.wg.Done() dec := decode.New(u.encoding) - buf := make([]byte, 0, MaxUDPSize) + scannerBuffer := make([]byte, 0, MaxUDPSize) for { // Read a message from the message queue. @@ -281,7 +295,9 @@ func (u *Input) processMessagesAsync(ctx context.Context) { return // Channel closed, exit the goroutine. } - u.processMessage(ctx, messageAndAddr.Message, messageAndAddr.RemoteAddr, dec, buf) + trimmedMessage := u.removeTrailingCharactersAndNULsFromBuffer(*messageAndAddr.Message, messageAndAddr.MessageLength) + u.processMessage(ctx, trimmedMessage, messageAndAddr.RemoteAddr, dec, scannerBuffer) + u.readBufferPool.Put(messageAndAddr.Message) } } @@ -331,17 +347,22 @@ func (u *Input) handleMessage(ctx context.Context, remoteAddr net.Addr, dec *dec } // readMessage will read log messages from the connection. -func (u *Input) readMessage() ([]byte, net.Addr, error) { - n, addr, err := u.connection.ReadFrom(u.buffer) +func (u *Input) readMessage(buffer []byte) ([]byte, net.Addr, int, error) { + n, addr, err := u.connection.ReadFrom(buffer) if err != nil { - return nil, nil, err + return nil, nil, 0, err } + return buffer, addr, n, nil +} + +// This will remove trailing characters and NULs from the buffer +func (u *Input) removeTrailingCharactersAndNULsFromBuffer(buffer []byte, n int) []byte { // Remove trailing characters and NULs - for ; (n > 0) && (u.buffer[n-1] < 32); n-- { // nolint + for ; (n > 0) && (buffer[n-1] < 32); n-- { // nolint } - return u.buffer[:n], addr, nil + return buffer[:n] } // Stop will stop listening for udp messages.