out_nats: handshake once per connection, write-completeness checks, respect server max_payload#12159
out_nats: handshake once per connection, write-completeness checks, respect server max_payload#12159ChpaMing wants to merge 3 commits into
Conversation
The flush callback wrote the NATS CONNECT handshake on every flush, even when the upstream layer handed back a recycled keepalive connection. The server only needs CONNECT right after the TCP connection is established, so resending it on every flush just adds noise to the protocol stream. Track the handshake state in the connection's own user_data field (unused for upstream connections): connections are zero-initialized on creation, so a fresh connection always runs the handshake while a recycled one keeps the mark and skips it. Connections that fail I/O are destroyed by the upstream layer instead of being recycled, and the keepalive queue destroys connections dropped by the server, so a stale mark can never skip the handshake on a new connection. Signed-off-by: ChpaMing <cham@users.noreply.github.com>
The CONNECT and PUB writes only checked whether flb_io_net_write() returned -1. A short write (bytes_sent < requested length) was treated as success, leaving the protocol stream misaligned: the server then parses the next frame in the middle of the truncated one and the connection degrades into protocol errors. Fail the flush with FLB_RETRY whenever fewer bytes than requested were written, for both the CONNECT handshake and the PUB frame. Signed-off-by: ChpaMing <cham@users.noreply.github.com>
The plugin used to publish a whole chunk as a single PUB message. Under load a chunk can grow past the nats-server max_payload limit (1MB by default), in which case the server rejects the message with -ERR 'Maximum Payload Violation' and closes the connection: the flush fails deterministically no matter how many times it is retried. Convert and queue records one by one and emit a PUB whenever the pending payload would cross a fixed 1MB limit. Every message keeps the usual '[[ts, record], ...]' JSON array shape, so the message semantics are unchanged for downstream consumers. A fixed limit is used on purpose instead of parsing max_payload from the server INFO line, to keep the plugin simple; it matches the server default. A single record whose own payload exceeds the limit would be rejected no matter how it is batched and would retry-loop the chunk forever: log an error, drop it, and continue with the remaining records. Signed-off-by: ChpaMing <cham@users.noreply.github.com>
📝 WalkthroughWalkthroughThe NATS output plugin now converts records individually, batches JSON payloads below a 1 MB limit, sends framed PUB requests with write validation, reuses CONNECT handshakes per connection, drops oversized records, and retries failed flushes. ChangesNATS payload flow
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant cb_nats_flush
participant record_to_json
participant nats_pub
participant NATSConnection
cb_nats_flush->>NATSConnection: reuse connection and send CONNECT once
cb_nats_flush->>record_to_json: convert decoded event
record_to_json-->>cb_nats_flush: return JSON element
cb_nats_flush->>nats_pub: publish size-limited batch
nats_pub-->>cb_nats_flush: return write status
cb_nats_flush-->>cb_nats_flush: return FLB_RETRY on failure
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
plugins/out_nats/nats.c (1)
353-363: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider counting dropped records in the plugin metrics.
Records discarded at Line 309 are only logged; there is no drop accounting, so operators see a silent gap between ingested and delivered records. Emitting a drop metric (or at least rate-limiting the error log) would keep drop state observable for this route.
As per coding guidelines: "Preserve route-specific success, retry, and drop state".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@plugins/out_nats/nats.c` around lines 353 - 363, Update the record-discard path around the existing Line 309 handling in the NATS output route to increment the plugin’s dropped-record metric whenever a record is discarded, while preserving the existing success and retry accounting. Reuse the established output-plugin metric or drop-accounting symbol rather than adding a separate mechanism; keep the current error logging behavior unless required by the metric implementation.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@plugins/out_nats/nats.c`:
- Around line 309-319: Update the oversized-record check in the batch-building
flow around elem_len and batch_count to account for the two array wrapper bytes
added to every published payload. Reject records whose serialized size plus '['
and ']' exceeds NATS_MAX_PAYLOAD, ensuring they are dropped before the
empty-batch path can publish an oversized message; keep the existing
batch-splitting behavior for records that fit individually.
- Around line 288-355: Guard every flb_sds_cat call in the batching flow around
json_msg, including initialisation, delimiters, record appends, and final
closing, so allocation failure is detected before json_msg is overwritten.
Preserve and destroy the original buffer on failure, clean up elem, the decoder,
and upstream connection as appropriate, then return FLB_RETRY; a small append
helper may centralize this handling.
- Around line 241-254: Add an error log in the NATS CONNECT write-failure branch
inside the u_conn->user_data initialization block, before releasing the
connection and returning FLB_RETRY. Use flb_plg_error with the plugin context
ctx->ins and include the failed or incomplete CONNECT write details.
---
Nitpick comments:
In `@plugins/out_nats/nats.c`:
- Around line 353-363: Update the record-discard path around the existing Line
309 handling in the NATS output route to increment the plugin’s dropped-record
metric whenever a record is discarded, while preserving the existing success and
retry accounting. Reuse the established output-plugin metric or drop-accounting
symbol rather than adding a separate mechanism; keep the current error logging
behavior unless required by the metric implementation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 93d49b3d-8dd4-42fc-b299-e78adfde965f
📒 Files selected for processing (1)
plugins/out_nats/nats.c
| if (u_conn->user_data == NULL) { | ||
| ret = flb_io_net_write(u_conn, | ||
| NATS_CONNECT, | ||
| sizeof(NATS_CONNECT) - 1, | ||
| &bytes_sent); | ||
| if (ret == -1 || bytes_sent != sizeof(NATS_CONNECT) - 1) { | ||
| /* an incomplete CONNECT leaves the protocol stream misaligned */ | ||
| flb_upstream_conn_release(u_conn); | ||
| FLB_OUTPUT_RETURN(FLB_RETRY); | ||
| } | ||
|
|
||
| flb_plg_debug(ctx->ins, "sent NATS CONNECT handshake"); | ||
| u_conn->user_data = u_conn; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect flb_connection definition and all user_data usages on connections
fd -t f 'flb_connection.h' | xargs -r cat -n
rg -nP '\buser_data\b' --type=c --type=h -g '!plugins/out_nats/**' -C2 | rg -n 'conn|connection' -C2Repository: fluent/fluent-bit
Length of output: 8361
🏁 Script executed:
#!/bin/bash
set -u
echo "== user_data usages (excluding plugins/out_nats/nats.c) =="
rg -n '\buser_data\b' --type=c --type=h plugins include src tests | grep -v 'plugins/out_nats/nats.c' || true
echo
echo "== NATS plugin snippet around handshakes =="
sed -n '220,265p' plugins/out_nats/nats.c | cat -n
echo
echo "== upstream conn setup/release snippets =="
rg -n 'flb_connection_setup|flb_connection_create|flb_upstream_conn_release|av_queue|busy_queue|user_data|cleanup_connection|destroy_conn|keepalive' src include plugins plugins/out include/fluent-bit/server -g '*.{c,h}' | head -n 240Repository: fluent/fluent-bit
Length of output: 36345
🏁 Script executed:
#!/bin/bash
set -u
echo "== src/flb_connection.c =="
cat -n src/flb_connection.c
echo
echo "== src/flb_upstream.c focused sections =="
rg -n 'flb_upstream_conn_(get|release|cleanup|destroy)|_available|av_queue|user_data|flb_connection_setup|flb_connection_create|prepare_destroy_conn|cleanup_connection|keepalive' src/flb_upstream.c src/flb_upstream.h src/flb_io.c include -g '*.{c,h}' -C3 | sed -n '1,260p'
echo
echo "== exact user_data writes in core C source excluding generated vmlinux headers =="
rg -n '->user_data\s*=|\.user_data\s*=' src plugins/include include/fluent-bit/*.h include/fluent-bit/*/*.h --type=c --type=h | grep -v 'plugins/in_ebpf/traces/includes' || trueRepository: fluent/fluent-bit
Length of output: 24704
🏁 Script executed:
#!/bin/bash
set -u
echo "== exact user_data assignment/write references in C/H excluding eBPF generated headers and nats.c =="
rg -n -e '\->user_data\s*=\s*' -e '\.user_data\s*=\s*' src plugins include/fluent-bit --type=c --type=h | grep -v 'plugins/in_ebpf/traces/includes' || true
echo
echo "== flb_upstream_conn_get/release body =="
sed -n '711,810p' src/flb_upstream.c | cat -n
sed -n '862,925p' src/flb_upstream.c | cat -nRepository: fluent/fluent-bit
Length of output: 9463
Log the incomplete NATS CONNECT write.
Lines 228-253 document that only new connections send CONNECT, but the write failure path at line 227 has no flb_plg_error, so this may retry silently.
🛠️ Add the missing error log
if (ret == -1 || bytes_sent != sizeof(NATS_CONNECT) - 1) {
/* an incomplete CONNECT leaves the protocol stream misaligned */
+ flb_plg_error(ctx->ins, "CONNECT write incomplete (%zu/%zu bytes)",
+ ret == -1 ? (size_t) 0 : bytes_sent,
+ sizeof(NATS_CONNECT) - 1);
flb_upstream_conn_release(u_conn);
FLB_OUTPUT_RETURN(FLB_RETRY);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if (u_conn->user_data == NULL) { | |
| ret = flb_io_net_write(u_conn, | |
| NATS_CONNECT, | |
| sizeof(NATS_CONNECT) - 1, | |
| &bytes_sent); | |
| if (ret == -1 || bytes_sent != sizeof(NATS_CONNECT) - 1) { | |
| /* an incomplete CONNECT leaves the protocol stream misaligned */ | |
| flb_upstream_conn_release(u_conn); | |
| FLB_OUTPUT_RETURN(FLB_RETRY); | |
| } | |
| flb_plg_debug(ctx->ins, "sent NATS CONNECT handshake"); | |
| u_conn->user_data = u_conn; | |
| } | |
| if (u_conn->user_data == NULL) { | |
| ret = flb_io_net_write(u_conn, | |
| NATS_CONNECT, | |
| sizeof(NATS_CONNECT) - 1, | |
| &bytes_sent); | |
| if (ret == -1 || bytes_sent != sizeof(NATS_CONNECT) - 1) { | |
| /* an incomplete CONNECT leaves the protocol stream misaligned */ | |
| flb_plg_error(ctx->ins, "CONNECT write incomplete (%zu/%zu bytes)", | |
| ret == -1 ? (size_t) 0 : bytes_sent, | |
| sizeof(NATS_CONNECT) - 1); | |
| flb_upstream_conn_release(u_conn); | |
| FLB_OUTPUT_RETURN(FLB_RETRY); | |
| } | |
| flb_plg_debug(ctx->ins, "sent NATS CONNECT handshake"); | |
| u_conn->user_data = u_conn; | |
| } |
🧰 Tools
🪛 Cppcheck (2.21.0)
[error] 243-243: There is an unknown macro here somewhere. Configuration is required. If FLB_VERSION_STR is a macro then please configure it.
(unknownMacro)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@plugins/out_nats/nats.c` around lines 241 - 254, Add an error log in the NATS
CONNECT write-failure branch inside the u_conn->user_data initialization block,
before releasing the connection and returning FLB_RETRY. Use flb_plg_error with
the plugin context ctx->ins and include the failed or incomplete CONNECT write
details.
| json_msg = flb_sds_cat(json_msg, "[", 1); | ||
|
|
||
| req_len = snprintf(request, flb_sds_len(event_chunk->tag)+ 32, | ||
| "PUB %s %zu\r\n", | ||
| event_chunk->tag, json_len); | ||
| while ((ret = flb_log_event_decoder_next( | ||
| &log_decoder, | ||
| &log_event)) == FLB_EVENT_DECODER_SUCCESS) { | ||
| elem = record_to_json(ctx, &log_event, | ||
| event_chunk->tag, tag_len, config); | ||
| if (!elem) { | ||
| flb_plg_error(ctx->ins, "failed to convert record to JSON"); | ||
| flb_sds_destroy(json_msg); | ||
| flb_log_event_decoder_destroy(&log_decoder); | ||
| flb_upstream_conn_release(u_conn); | ||
| FLB_OUTPUT_RETURN(FLB_ERROR); | ||
| } | ||
| elem_len = flb_sds_len(elem); | ||
|
|
||
| /* | ||
| * A single record larger than the limit would be rejected no | ||
| * matter how it is batched and would retry-loop the chunk | ||
| * forever: drop it and continue with the next records. | ||
| */ | ||
| if (elem_len > NATS_MAX_PAYLOAD) { | ||
| flb_plg_error(ctx->ins, | ||
| "dropping record of %zu bytes: exceeds NATS " | ||
| "max_payload (%d)", elem_len, NATS_MAX_PAYLOAD); | ||
| flb_sds_destroy(elem); | ||
| continue; | ||
| } | ||
|
|
||
| /* Append JSON message and ending CRLF */ | ||
| memcpy(request + req_len, json_msg, json_len); | ||
| req_len += json_len; | ||
| request[req_len++] = '\r'; | ||
| request[req_len++] = '\n'; | ||
| flb_sds_destroy(json_msg); | ||
| /* send the current batch if this record would push it over the limit */ | ||
| if (batch_count > 0 && | ||
| flb_sds_len(json_msg) + 1 + elem_len + 1 > NATS_MAX_PAYLOAD) { | ||
| json_msg = flb_sds_cat(json_msg, "]", 1); | ||
| if (nats_pub(ctx, u_conn, event_chunk->tag, tag_len, | ||
| json_msg) == -1) { | ||
| flb_sds_destroy(elem); | ||
| flb_sds_destroy(json_msg); | ||
| flb_log_event_decoder_destroy(&log_decoder); | ||
| flb_upstream_conn_release(u_conn); | ||
| FLB_OUTPUT_RETURN(FLB_RETRY); | ||
| } | ||
| flb_sds_destroy(json_msg); | ||
|
|
||
| json_msg = flb_sds_create_size(4096); | ||
| if (!json_msg) { | ||
| flb_errno(); | ||
| flb_sds_destroy(elem); | ||
| flb_log_event_decoder_destroy(&log_decoder); | ||
| flb_upstream_conn_release(u_conn); | ||
| FLB_OUTPUT_RETURN(FLB_RETRY); | ||
| } | ||
| json_msg = flb_sds_cat(json_msg, "[", 1); | ||
| batch_count = 0; | ||
| } | ||
|
|
||
| ret = flb_io_net_write(u_conn, request, req_len, &bytes_sent); | ||
| if (ret == -1) { | ||
| flb_errno(); | ||
| flb_free(request); | ||
| flb_upstream_conn_release(u_conn); | ||
| FLB_OUTPUT_RETURN(FLB_RETRY); | ||
| if (batch_count > 0) { | ||
| json_msg = flb_sds_cat(json_msg, ",", 1); | ||
| } | ||
| json_msg = flb_sds_cat(json_msg, elem, elem_len); | ||
| flb_sds_destroy(elem); | ||
| batch_count++; | ||
| } | ||
|
|
||
| flb_free(request); | ||
| flb_log_event_decoder_destroy(&log_decoder); | ||
|
|
||
| /* send the final batch */ | ||
| if (batch_count > 0) { | ||
| json_msg = flb_sds_cat(json_msg, "]", 1); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unchecked flb_sds_cat results: NULL-deref and leak on allocation failure.
flb_sds_cat returns NULL and leaves the original buffer allocated when the growth realloc fails. Every call site here (Lines 288, 320, 339, 344, 346, 355) assigns the result straight back to json_msg, so an OOM leaks the old buffer and the next flb_sds_len(json_msg) / nats_pub() dereferences NULL. This is reachable precisely on the large-chunk path this PR targets, since batches grow toward 1 MB.
Suggest a small helper that appends and flags failure, then bail out with FLB_RETRY on error.
🛡️ Sketch of a guarded append
+static int nats_sds_cat(flb_sds_t *buf, const char *str, size_t len)
+{
+ flb_sds_t tmp;
+
+ tmp = flb_sds_cat(*buf, str, len);
+ if (!tmp) {
+ return -1;
+ }
+ *buf = tmp;
+ return 0;
+}- json_msg = flb_sds_cat(json_msg, elem, elem_len);
- flb_sds_destroy(elem);
+ ret = nats_sds_cat(&json_msg, elem, elem_len);
+ flb_sds_destroy(elem);
+ if (ret == -1) {
+ flb_errno();
+ flb_sds_destroy(json_msg);
+ flb_log_event_decoder_destroy(&log_decoder);
+ flb_upstream_conn_release(u_conn);
+ FLB_OUTPUT_RETURN(FLB_RETRY);
+ }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@plugins/out_nats/nats.c` around lines 288 - 355, Guard every flb_sds_cat call
in the batching flow around json_msg, including initialisation, delimiters,
record appends, and final closing, so allocation failure is detected before
json_msg is overwritten. Preserve and destroy the original buffer on failure,
clean up elem, the decoder, and upstream connection as appropriate, then return
FLB_RETRY; a small append helper may centralize this handling.
| if (elem_len > NATS_MAX_PAYLOAD) { | ||
| flb_plg_error(ctx->ins, | ||
| "dropping record of %zu bytes: exceeds NATS " | ||
| "max_payload (%d)", elem_len, NATS_MAX_PAYLOAD); | ||
| flb_sds_destroy(elem); | ||
| continue; | ||
| } | ||
|
|
||
| /* Append JSON message and ending CRLF */ | ||
| memcpy(request + req_len, json_msg, json_len); | ||
| req_len += json_len; | ||
| request[req_len++] = '\r'; | ||
| request[req_len++] = '\n'; | ||
| flb_sds_destroy(json_msg); | ||
| /* send the current batch if this record would push it over the limit */ | ||
| if (batch_count > 0 && | ||
| flb_sds_len(json_msg) + 1 + elem_len + 1 > NATS_MAX_PAYLOAD) { |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Off-by-two: a record between NATS_MAX_PAYLOAD - 1 and NATS_MAX_PAYLOAD still produces an oversized payload.
The drop threshold ignores the [/] wrapper that is always added. A record of exactly NATS_MAX_PAYLOAD bytes passes Line 309, then lands in an empty batch (batch_count == 0, so the split check at Line 318 is skipped) and is published as 1 + elem_len + 1 bytes — a Maximum Payload Violation and the infinite retry loop this branch is meant to prevent.
🐛 Account for the array delimiters in the drop check
- if (elem_len > NATS_MAX_PAYLOAD) {
+ /* the element is always wrapped in '[' ... ']' */
+ if (elem_len + 2 > NATS_MAX_PAYLOAD) {
flb_plg_error(ctx->ins,
"dropping record of %zu bytes: exceeds NATS "
"max_payload (%d)", elem_len, NATS_MAX_PAYLOAD);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if (elem_len > NATS_MAX_PAYLOAD) { | |
| flb_plg_error(ctx->ins, | |
| "dropping record of %zu bytes: exceeds NATS " | |
| "max_payload (%d)", elem_len, NATS_MAX_PAYLOAD); | |
| flb_sds_destroy(elem); | |
| continue; | |
| } | |
| /* Append JSON message and ending CRLF */ | |
| memcpy(request + req_len, json_msg, json_len); | |
| req_len += json_len; | |
| request[req_len++] = '\r'; | |
| request[req_len++] = '\n'; | |
| flb_sds_destroy(json_msg); | |
| /* send the current batch if this record would push it over the limit */ | |
| if (batch_count > 0 && | |
| flb_sds_len(json_msg) + 1 + elem_len + 1 > NATS_MAX_PAYLOAD) { | |
| /* the element is always wrapped in '[' ... ']' */ | |
| if (elem_len + 2 > NATS_MAX_PAYLOAD) { | |
| flb_plg_error(ctx->ins, | |
| "dropping record of %zu bytes: exceeds NATS " | |
| "max_payload (%d)", elem_len, NATS_MAX_PAYLOAD); | |
| flb_sds_destroy(elem); | |
| continue; | |
| } | |
| /* send the current batch if this record would push it over the limit */ | |
| if (batch_count > 0 && | |
| flb_sds_len(json_msg) + 1 + elem_len + 1 > NATS_MAX_PAYLOAD) { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@plugins/out_nats/nats.c` around lines 309 - 319, Update the oversized-record
check in the batch-building flow around elem_len and batch_count to account for
the two array wrapper bytes added to every published payload. Reject records
whose serialized size plus '[' and ']' exceeds NATS_MAX_PAYLOAD, ensuring they
are dropped before the empty-batch path can publish an oversized message; keep
the existing batch-splitting behavior for records that fit individually.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 39c3b14d87
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| * INFO line was considered and deliberately left out to keep the plugin | ||
| * simple: the fixed 1MB limit matches the server default. | ||
| */ | ||
| #define NATS_MAX_PAYLOAD (1024 * 1024) |
There was a problem hiding this comment.
When a NATS server is configured with max_payload below 1 MiB, this hard-coded value still permits oversized PUB frames. Because nats_pub only verifies that the socket write completed and never reads the server's -ERR, a single oversized final batch can be rejected while the flush returns FLB_OK, permanently releasing and losing the chunk; parse the connection's INFO line and batch against its advertised limit instead.
AGENTS.md reference: AGENTS.md:L267-L271
Useful? React with 👍 / 👎.
| * matter how it is batched and would retry-loop the chunk | ||
| * forever: drop it and continue with the next records. | ||
| */ | ||
| if (elem_len > NATS_MAX_PAYLOAD) { |
There was a problem hiding this comment.
For a record whose serialized element is 1,048,575 or 1,048,576 bytes, this check accepts the element even though the actual PUB payload adds the opening and closing array brackets, producing 1,048,577 or 1,048,578 bytes. A default NATS server rejects that frame, while a completed socket write can still make this flush return FLB_OK; the single-record threshold must reserve both delimiter bytes.
AGENTS.md reference: AGENTS.md:L267-L271
Useful? React with 👍 / 👎.
| if (batch_count > 0) { | ||
| json_msg = flb_sds_cat(json_msg, ",", 1); | ||
| } | ||
| json_msg = flb_sds_cat(json_msg, elem, elem_len); |
There was a problem hiding this comment.
When this append must grow the initial 4 KiB buffer and realloc fails under memory pressure, flb_sds_cat returns NULL while leaving the original allocation intact. Assigning that result directly to json_msg loses the allocation and then lets the next flb_sds_len, flb_sds_cat, or nats_pub dereference NULL, crashing Fluent Bit instead of returning FLB_RETRY; retain the old pointer and check the append result.
Useful? React with 👍 / 👎.
2 participants
Three independent fixes to the NATS output plugin, one commit each:
1.
out_nats: send CONNECT once per connection, not on every flushcb_nats_flushcurrently writesNATS_CONNECTbefore every PUB. With keepalive enabled (the default), every flush on a reused connection repeats the client handshake. The NATS protocol expects CONNECT once per connection; nats-server tolerates the repeats, but it is unnecessary protocol traffic and breaks any consumer/proxy that assumes one CONNECT per connection. Handshake state is now tracked per upstream connection. The flag cannot go stale: connections that errored are never recycled (flb_upstream.ckeepalive release path), and connections dropped by the server are destroyed viacb_upstream_conn_ka_dropped.2.
out_nats: treat incomplete writes as flush failureflb_io_net_writewas only checked forret == -1. A short write (bytes_sent < len) leaves the server with a truncated frame, desynchronizing the protocol stream; the server then discards/closes the connection and the chunk is lost without any error surfaced. Incomplete CONNECT or PUB writes now fail the flush (FLB_RETRY).3.
out_nats: split chunks exceeding server max_payloadThe whole chunk is currently published as a single PUB message. nats-server's default
max_payloadis 1 MB; a larger publish is rejected (-ERR 'Maximum Payload Violation') and the server closes the connection, so the flush can never succeed. We hit this in production collecting Windows event logs (winevtlog replaying channel history), with the server logging e.g.maximum payload exceeded: 1178357 vs 1048576. The flush now batches records into multiple PUB messages, each kept under a fixed 1 MB — deliberately conservative; parsing the server INFOmax_payloadis noted as a possible follow-up in a code comment. Message shape is unchanged: each message is still a JSON array of[timestamp, record]pairs. A single record larger than the limit is dropped with an error log instead of causing an infinite retry loop.Out of scope (deliberately not included): NATS authentication (user/pass/token) and platform-specific connection handling. We maintain a downstream fork with those and can propose them separately if there is interest.
N/A
Testing
Example configuration used for the test below:
Debug log output (12 s run, ~100 records/s, keepalive connection reused across flushes):
CONNECT appears exactly once for the whole run (previously: once per flush), zero
[error]lines, zero retries.All records were persisted by the server's JetStream stream (
raw_logs, filterlogs.raw.>); stream message count grew monotonically during the run, and reading messages back confirmed the unchanged[[ts, {"tag": "logs.raw.sysmon", ...}], ...]shape.The max_payload batching logic is the same code we run in a downstream fork feeding Windows Sysmon/Security events into JetStream in production.
Build: MSVC on Windows,
plugins/out_nats/nats.ccompiles with zero warnings.Example configuration file for the change
Debug log output from testing the change
Attached Valgrind output that shows no leaks or memory corruption was found
Valgrind: N/A — development and testing for this PR were done on Windows (MSVC), where Valgrind is not available. The changes add one heap allocation per PUB frame (freed on all paths) and reuse the existing sds/json conversion helpers.
Documentation
N/A — no configuration changes; all three fixes preserve existing behavior and message format.
Backporting
These are bug fixes and would be reasonable backport candidates, but we leave that call to the maintainers.
Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.
Summary by CodeRabbit
New Features
Bug Fixes