in_aegisbpf: add AegisBPF runtime-security input plugin - #12272
Conversation
Streams runtime-security events from a co-located AegisBPF agent (BPF-LSM enforcement) over its opt-in root-only Unix control socket (GET /events -> newline-delimited JSON/OCSF) into the pipeline. Event-driven collector with reconnect; drains promptly because the agent drops slow readers. Signed-off-by: Eren Arı <erenari27@gmail.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds the ChangesAegisBPF input integration
Estimated code review effort: 4 (Complex) | ~45 minutes Mergeability Score: ⚪ Minimal · up to This PR adds the AegisBPF input plugin and its build registration; no actionable merge-blocking risk remains in the supplied evidence, so it is merge-ready after normal checks and review. Sequence Diagram(s)sequenceDiagram
participant in_aegisbpf
participant AegisBPF as AegisBPF Unix socket
participant Pipeline as Fluent Bit pipeline
in_aegisbpf->>AegisBPF: Request /events stream
AegisBPF-->>in_aegisbpf: Send newline-delimited JSON
in_aegisbpf->>Pipeline: Append encoded valid events
Possibly related PRs
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: 4
🧹 Nitpick comments (9)
plugins/in_aegisbpf/in_aegisbpf.c (8)
65-79: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRename
write_allfor consistency.Every other static function in this file uses an
aegisbpf_orin_aegisbpf_prefix. Renamewrite_alltoaegisbpf_write_all.The coding guidelines require descriptive
snake_casenames with a component prefix.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 65 - 79, Rename the static helper write_all to aegisbpf_write_all and update every call site in the file to use the new component-prefixed name, preserving its behavior unchanged.Source: Coding guidelines
83-95: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove declarations to the top of the function.
The function uses two anonymous nested blocks so that
line,line_len,mp,mp_size,root_type,consumed, andretcan be declared mid-function. The coding guidelines state: "Declare variables at the start of functions rather than mid-block." Declare all of these at the top ofaegisbpf_process_linesand remove the nested blocks. This also removes two indentation levels.The same pattern appears in
write_all(line 69) andin_aegisbpf_read(lines 172, 183-184).Also applies to: 113-118
🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 83 - 95, Update aegisbpf_process_lines, write_all, and in_aegisbpf_read to declare all local variables at the start of each function, including line, line_len, mp, mp_size, root_type, consumed, and ret where applicable. Remove the anonymous nested blocks introduced solely for declarations and adjust indentation while preserving the existing control flow and behavior.Source: Coding guidelines
249-280: 🩺 Stability & Availability | 🔵 Trivial | ⚖️ Poor tradeoff
connectandwrite_allrun blocking on the engine thread.The code sets
O_NONBLOCKonly afterconnectandwrite_allcomplete. Both calls therefore block the engine thread. For anAF_UNIXstream socket with a listening peer both normally return at once, and the request is 12 bytes. If the agent's accept backlog is full,connectcan still block.Set
O_NONBLOCKimmediately aftersocket(), and handleEINPROGRESSfromconnectandEAGAINfromwrite.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 249 - 280, Set O_NONBLOCK immediately after socket creation in the connection setup, before connect and write_all. Update connect handling to accept EINPROGRESS as an in-progress connection, and ensure the stream request write path handles EAGAIN without blocking, preserving cleanup and failure behavior for other errors.
445-456: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftSet
.cb_collecttoNULLand add runtime tests.Two points:
.cb_collectpoints toin_aegisbpf_reconnect, andin_aegisbpf_initalready registers the same function throughflb_input_set_collector_time. The duplicate registration is harmless becausein_aegisbpf_reconnectreturns early whenctx->connectedis set, but the intent is unclear. Set.cb_collecttoNULL, as plugins that register their own collectors do.- No test file accompanies this plugin. The coding guidelines require validation of "both success and failure paths, including invalid payloads, boundary sizes, and null or missing fields." Add a runtime test with a mock Unix-socket agent that covers the acknowledgement-line skip, a malformed JSON line, a line at
FLB_IN_AEGISBPF_BUF_MAX, a non-object JSON root, EOF, and reconnection.I can generate the runtime test skeleton. Tell me if you want it.
🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 445 - 456, Set .cb_collect to NULL in in_aegisbpf_plugin because in_aegisbpf_init already registers in_aegisbpf_reconnect via flb_input_set_collector_time. Add a runtime test using a mock Unix-socket agent covering acknowledgement-line skipping, malformed JSON, FLB_IN_AEGISBPF_BUF_MAX boundary input, non-object JSON, EOF, and reconnection.Source: Coding guidelines
131-146: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog encoder failures.
If
flb_log_event_encoder_begin_recordfails, the code drops the event with no log entry and no metric. Ifset_body_from_raw_msgpackfails, the rollback is also silent. Add aflb_plg_errorcall on each failure path so that record loss is observable.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 131 - 146, Add flb_plg_error logging to both failure paths in the record-encoding block: when flb_log_event_encoder_begin_record fails and when flb_log_event_encoder_set_body_from_raw_msgpack fails before rollback. Preserve the existing commit and rollback behavior while making each dropped record observable.
402-428: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winDocument or handle event loss across pause.
The file header on lines 29-30 states that the agent drops slow readers.
in_aegisbpf_pausestops the read collector, so the socket is no longer drained. The agent then drops this reader and closes the connection. On resume the read collector observes EOF,aegisbpf_disconnectruns, and the reconnect timer rebuilds the connection. The plugin recovers, but every event produced during the pause is lost and no message reports the loss.Consider calling
aegisbpf_disconnectinin_aegisbpf_pauseand logging a warning about the gap. That makes the loss explicit and avoids the stale-fd resume path.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 402 - 428, Update in_aegisbpf_pause to call aegisbpf_disconnect before pausing collectors, then log a warning describing that events may be lost during the pause. Ensure the disconnect clears the read connection so in_aegisbpf_resume does not resume a stale file descriptor, while preserving the existing collector pause behavior.
346-348: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe
reconnect_secguard cannot trigger for a negative value the operator sets.The config map on line 437 already supplies the default
"2". This guard only rewrites explicit values of0or less. It silently replaces an invalid operator value instead of reporting it.Reject a non-positive
reconnect_secwithflb_plg_errorand return-1, so the operator learns that the value is invalid.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 346 - 348, The reconnect_sec handling in the plugin initialization path should reject operator-provided values less than or equal to zero instead of replacing them with FLB_IN_AEGISBPF_DEFAULT_RECONN. Log the invalid value with flb_plg_error, return -1 from the surrounding initialization function, and preserve valid positive values unchanged.
306-318: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winA permanently failing connection is silent at the default log level.
aegisbpf_connectlogs connect failures withflb_plg_debug.in_aegisbpf_reconnectdiscards the return value. Ifsocket_pathis wrong or the agent is absent, the plugin emits no records and no visible message at the default log level.Log the first failure, or every Nth failure, with
flb_plg_warn. Keep the repeated attempts at debug level so the log does not flood.🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 306 - 318, The reconnect path in in_aegisbpf_reconnect currently discards aegisbpf_connect failures, leaving permanent connection problems invisible. Track the connection failure result and emit a flb_plg_warn message for the first failure or an appropriate periodic failure interval, while keeping subsequent retry messages at flb_plg_debug to avoid log flooding.plugins/in_aegisbpf/in_aegisbpf.h (1)
41-42: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRename the collector id fields.
coll_fd_reconnectandcoll_fd_readhold collector ids returned byflb_input_set_collector_timeandflb_input_set_collector_socket, not file descriptors. The_fdsuffix conflicts withfdon line 38, which is a real descriptor. Usecoll_id_reconnectandcoll_id_readto match the convention inplugins/in_docker_events/docker_events.c.🤖 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/in_aegisbpf/in_aegisbpf.h` around lines 41 - 42, Rename the collector ID fields coll_fd_reconnect and coll_fd_read to coll_id_reconnect and coll_id_read, and update every reference to these fields throughout the plugin, including their flb_input_set_collector_time/socket setup and cleanup paths. Leave the actual fd field unchanged.
🤖 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 `@cmake/plugins_options.cmake`:
- Line 14: Default FLB_IN_AEGISBPF to OFF on Windows while preserving ON for
other platforms in cmake/plugins_options.cmake:14-14. In
plugins/CMakeLists.txt:248-248, wrap REGISTER_IN_PLUGIN("in_aegisbpf") in a
platform condition excluding Windows, matching existing Unix-only plugin guards.
In `@plugins/in_aegisbpf/in_aegisbpf.c`:
- Around line 171-227: Bound the per-invocation drain work in the callback’s
recv loop by adding FLB_IN_AEGISBPF_DRAIN_MAX in in_aegisbpf.h and tracking
bytes received via a drained counter after each successful recv. Stop processing
once the byte or record limit is reached, flush the accumulated encoder output
through flb_input_log_append, and return so the event-driven collector can
re-arm and continue on the next wake.
- Around line 174-196: Track oversized-line handling with a new discarding state
in struct flb_in_aegisbpf instead of resetting buf_len and forcing
handshake_done in the buffer-growth path. Update aegisbpf_process_lines to
discard bytes through the next newline, then clear the state and resume normal
parsing; leave handshake_done unchanged. Reset discarding alongside buf_len in
aegisbpf_connect and aegisbpf_disconnect.
- Around line 113-129: Update the flb_pack_json handling in the in_aegisbpf
input path to use its record-count API and accept only exactly one
FLB_PACK_JSON_OBJECT. Reject arrays, scalar roots, multiple JSON values, and any
non-whitespace trailing bytes after consumed; retain the existing
debug-and-continue behavior for invalid lines and free allocated output before
rejection.
---
Nitpick comments:
In `@plugins/in_aegisbpf/in_aegisbpf.c`:
- Around line 65-79: Rename the static helper write_all to aegisbpf_write_all
and update every call site in the file to use the new component-prefixed name,
preserving its behavior unchanged.
- Around line 83-95: Update aegisbpf_process_lines, write_all, and
in_aegisbpf_read to declare all local variables at the start of each function,
including line, line_len, mp, mp_size, root_type, consumed, and ret where
applicable. Remove the anonymous nested blocks introduced solely for
declarations and adjust indentation while preserving the existing control flow
and behavior.
- Around line 249-280: Set O_NONBLOCK immediately after socket creation in the
connection setup, before connect and write_all. Update connect handling to
accept EINPROGRESS as an in-progress connection, and ensure the stream request
write path handles EAGAIN without blocking, preserving cleanup and failure
behavior for other errors.
- Around line 445-456: Set .cb_collect to NULL in in_aegisbpf_plugin because
in_aegisbpf_init already registers in_aegisbpf_reconnect via
flb_input_set_collector_time. Add a runtime test using a mock Unix-socket agent
covering acknowledgement-line skipping, malformed JSON, FLB_IN_AEGISBPF_BUF_MAX
boundary input, non-object JSON, EOF, and reconnection.
- Around line 131-146: Add flb_plg_error logging to both failure paths in the
record-encoding block: when flb_log_event_encoder_begin_record fails and when
flb_log_event_encoder_set_body_from_raw_msgpack fails before rollback. Preserve
the existing commit and rollback behavior while making each dropped record
observable.
- Around line 402-428: Update in_aegisbpf_pause to call aegisbpf_disconnect
before pausing collectors, then log a warning describing that events may be lost
during the pause. Ensure the disconnect clears the read connection so
in_aegisbpf_resume does not resume a stale file descriptor, while preserving the
existing collector pause behavior.
- Around line 346-348: The reconnect_sec handling in the plugin initialization
path should reject operator-provided values less than or equal to zero instead
of replacing them with FLB_IN_AEGISBPF_DEFAULT_RECONN. Log the invalid value
with flb_plg_error, return -1 from the surrounding initialization function, and
preserve valid positive values unchanged.
- Around line 306-318: The reconnect path in in_aegisbpf_reconnect currently
discards aegisbpf_connect failures, leaving permanent connection problems
invisible. Track the connection failure result and emit a flb_plg_warn message
for the first failure or an appropriate periodic failure interval, while keeping
subsequent retry messages at flb_plg_debug to avoid log flooding.
In `@plugins/in_aegisbpf/in_aegisbpf.h`:
- Around line 41-42: Rename the collector ID fields coll_fd_reconnect and
coll_fd_read to coll_id_reconnect and coll_id_read, and update every reference
to these fields throughout the plugin, including their
flb_input_set_collector_time/socket setup and cleanup paths. Leave the actual fd
field unchanged.
🪄 Autofix
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: e2f1e24e-01f9-4613-b365-5dcb2430c2a0
📒 Files selected for processing (5)
cmake/plugins_options.cmakeplugins/CMakeLists.txtplugins/in_aegisbpf/CMakeLists.txtplugins/in_aegisbpf/in_aegisbpf.cplugins/in_aegisbpf/in_aegisbpf.h
|
|
||
| # Inputs (sources, data collectors) | ||
| # ================================= | ||
| DEFINE_OPTION(FLB_IN_AEGISBPF "Enable AegisBPF input plugin" ON) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
The POSIX-only plugin is enabled and registered without a platform guard. plugins/in_aegisbpf/in_aegisbpf.c includes <sys/un.h>, <sys/socket.h>, and <unistd.h>, and it calls fcntl and recv. Neither the build option nor the registration excludes Windows, so a Windows build attempts to compile the plugin and fails.
cmake/plugins_options.cmake#L14-L14: defaultFLB_IN_AEGISBPFtoOFFon Windows instead ofONfor all platforms.plugins/CMakeLists.txt#L248-L248: wrapREGISTER_IN_PLUGIN("in_aegisbpf")in a platform condition that excludes Windows, matching how the repository gates other Unix-only inputs.
📍 Affects 2 files
cmake/plugins_options.cmake#L14-L14(this comment)plugins/CMakeLists.txt#L248-L248
🤖 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 `@cmake/plugins_options.cmake` at line 14, Default FLB_IN_AEGISBPF to OFF on
Windows while preserving ON for other platforms in
cmake/plugins_options.cmake:14-14. In plugins/CMakeLists.txt:248-248, wrap
REGISTER_IN_PLUGIN("in_aegisbpf") in a platform condition excluding Windows,
matching existing Unix-only plugin guards.
| while (1) { | ||
| ssize_t n; | ||
|
|
||
| if (ctx->buf_len == ctx->buf_size) { | ||
| if (ctx->buf_size >= FLB_IN_AEGISBPF_BUF_MAX) { | ||
| /* A single line exceeded the cap; drop it defensively. */ | ||
| flb_plg_warn(ins, "line exceeded %d bytes, dropping", | ||
| FLB_IN_AEGISBPF_BUF_MAX); | ||
| ctx->buf_len = 0; | ||
| ctx->handshake_done = 1; | ||
| } | ||
| else { | ||
| size_t new_size = ctx->buf_size * 2; | ||
| char *tmp; | ||
| if (new_size > FLB_IN_AEGISBPF_BUF_MAX) { | ||
| new_size = FLB_IN_AEGISBPF_BUF_MAX; | ||
| } | ||
| tmp = flb_realloc(ctx->buf, new_size); | ||
| if (tmp == NULL) { | ||
| flb_errno(); | ||
| break; | ||
| } | ||
| ctx->buf = tmp; | ||
| ctx->buf_size = new_size; | ||
| } | ||
| } | ||
|
|
||
| n = recv(ctx->fd, ctx->buf + ctx->buf_len, | ||
| ctx->buf_size - ctx->buf_len, 0); | ||
| if (n > 0) { | ||
| ctx->buf_len += (size_t) n; | ||
| aegisbpf_process_lines(ctx); | ||
| continue; | ||
| } | ||
| else if (n == 0) { | ||
| flb_plg_info(ins, "agent closed the connection"); | ||
| disconnected = 1; | ||
| break; | ||
| } | ||
| else { | ||
| if (errno == EAGAIN || errno == EWOULDBLOCK) { | ||
| break; /* drained */ | ||
| } | ||
| if (errno == EINTR) { | ||
| continue; | ||
| } | ||
| flb_plg_warn(ins, "read error: %s", strerror(errno)); | ||
| disconnected = 1; | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| if (ctx->encoder->output_length > 0) { | ||
| flb_input_log_append(ins, NULL, 0, | ||
| ctx->encoder->output_buffer, | ||
| ctx->encoder->output_length); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Bound the drain loop and the accumulated encoder output.
The loop runs until recv returns EAGAIN. A continuously writing agent can hold the engine thread in this callback for an unbounded time. All records accumulate in one encoder buffer, and the single flb_input_log_append at the end can be arbitrarily large. This blocks other collectors and delays backpressure.
Add a per-invocation bound. Flush the encoder and return when the loop reaches a byte or record limit. The event-driven collector re-arms and continues on the next wake.
♻️ Suggested shape
static int in_aegisbpf_read(struct flb_input_instance *ins,
struct flb_config *config, void *data)
{
struct flb_in_aegisbpf *ctx = data;
int disconnected = 0;
+ size_t drained = 0;
(void) config;
flb_log_event_encoder_reset(ctx->encoder);
while (1) {
ssize_t n;
+
+ if (drained >= FLB_IN_AEGISBPF_DRAIN_MAX) {
+ break;
+ }Increment drained by n after each successful recv, and define FLB_IN_AEGISBPF_DRAIN_MAX in plugins/in_aegisbpf/in_aegisbpf.h.
🤖 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/in_aegisbpf/in_aegisbpf.c` around lines 171 - 227, Bound the
per-invocation drain work in the callback’s recv loop by adding
FLB_IN_AEGISBPF_DRAIN_MAX in in_aegisbpf.h and tracking bytes received via a
drained counter after each successful recv. Stop processing once the byte or
record limit is reached, flush the accumulated encoder output through
flb_input_log_append, and return so the event-driven collector can re-arm and
continue on the next wake.
There was a problem hiding this comment.
^ @ErenAri
We need to address this issue.
ref: #12272 (comment).
![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: 1f2f05d429
ℹ️ 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".
|
Follow-ups on the checklist: Documentation PR: fluent/fluent-bit-docs#2670 (adds Valgrind (built with No leaks, no errors, and no On "why not a generic input": Fluent Bit has no generic input that connects out as a client to a Unix stream socket and performs a request/handshake ( |
Docs + example config for streaming AegisBPF OCSF events into any Fluent Bit output via the native `aegisbpf` input plugin. Links the upstream plugin PR (fluent/fluent-bit#12272) and docs PR (fluent/fluent-bit-docs#2670), with a generic-input fallback until the plugin ships in a release. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
There was a problem hiding this comment.
As you might be noticed that UNIX socket with the normal creation procedure is only work on non Windows environment. So, could you add a conditional clause on CMakeLists.txt for definition of in_agisbpf plugin?
It seems that this plugin is only working on Linux so we need to add a restriction to compile this plugin.
It would be better to put it inside of the following clause:
# These plugins works only on Linux
if(${CMAKE_SYSTEM_NAME} MATCHES "Linux")
# Other Linux only plugins...
REGISTER_IN_PLUGIN("in_aegisbpf")
endif()
|
Thanks for the review @cosmo0920! Fixed in the latest commit:
While I was at it I also addressed the automated-review findings:
Rebuilt and re-tested end-to-end (mock agent → records forwarded; array/scalar lines correctly rejected). For the docs-required label: the docs PR is up at fluent/fluent-bit-docs#2670 ( |
…loop Address review feedback: - Register the plugin only inside the Linux-only block (POSIX Unix socket / fcntl/recv); fixes the Windows build. (cosmo0920) - Accept only a single whole JSON object per line (reject arrays, scalars, and trailing roots from flb_pack_json). - On an over-length line, skip the tail to the next newline instead of corrupting the next line / handshake state. - Bound bytes drained per collector wake so a busy agent can't hold the engine thread or grow the append arbitrarily. Signed-off-by: Eren Arı <erenari27@gmail.com>
ec849e5 to
152ccc2
Compare
|
Heads up on CI: the previous unit-test failures were transient infrastructure flakes, not the plugin — the jobs died during environment setup (repeated |
cosmo0920
left a comment
cosmo0920
left a comment
There was a problem hiding this comment.
I found style issues and an unbounded condition in an infinite loop under a certain reason.
| char *line = ctx->buf + start; | ||
| size_t line_len = i - start; |
There was a problem hiding this comment.
We don't recommend to define variables in the middle of functions.
| char *mp = NULL; | ||
| size_t mp_size = 0; | ||
| int root_type = 0; | ||
| size_t consumed = 0; | ||
| int ret; |
| flb_log_event_encoder_reset(ctx->encoder); | ||
|
|
||
| while (1) { | ||
| ssize_t n; |
| while (1) { | ||
| ssize_t n; | ||
|
|
||
| if (ctx->buf_len == ctx->buf_size) { | ||
| if (ctx->buf_size >= FLB_IN_AEGISBPF_BUF_MAX) { | ||
| /* A single line exceeded the cap; drop it defensively. */ | ||
| flb_plg_warn(ins, "line exceeded %d bytes, dropping", | ||
| FLB_IN_AEGISBPF_BUF_MAX); | ||
| ctx->buf_len = 0; | ||
| ctx->handshake_done = 1; | ||
| } | ||
| else { | ||
| size_t new_size = ctx->buf_size * 2; | ||
| char *tmp; | ||
| if (new_size > FLB_IN_AEGISBPF_BUF_MAX) { | ||
| new_size = FLB_IN_AEGISBPF_BUF_MAX; | ||
| } | ||
| tmp = flb_realloc(ctx->buf, new_size); | ||
| if (tmp == NULL) { | ||
| flb_errno(); | ||
| break; | ||
| } | ||
| ctx->buf = tmp; | ||
| ctx->buf_size = new_size; | ||
| } | ||
| } | ||
|
|
||
| n = recv(ctx->fd, ctx->buf + ctx->buf_len, | ||
| ctx->buf_size - ctx->buf_len, 0); | ||
| if (n > 0) { | ||
| ctx->buf_len += (size_t) n; | ||
| aegisbpf_process_lines(ctx); | ||
| continue; | ||
| } | ||
| else if (n == 0) { | ||
| flb_plg_info(ins, "agent closed the connection"); | ||
| disconnected = 1; | ||
| break; | ||
| } | ||
| else { | ||
| if (errno == EAGAIN || errno == EWOULDBLOCK) { | ||
| break; /* drained */ | ||
| } | ||
| if (errno == EINTR) { | ||
| continue; | ||
| } | ||
| flb_plg_warn(ins, "read error: %s", strerror(errno)); | ||
| disconnected = 1; | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| if (ctx->encoder->output_length > 0) { | ||
| flb_input_log_append(ins, NULL, 0, | ||
| ctx->encoder->output_buffer, | ||
| ctx->encoder->output_length); | ||
| } |
There was a problem hiding this comment.
^ @ErenAri
We need to address this issue.
ref: #12272 (comment).
Address review style feedback: move all local declarations to the top of write_all, aegisbpf_process_lines and in_aegisbpf_read (C89 style, no declarations in the middle of functions), removing the nested blocks. Signed-off-by: Eren Arı <erenari27@gmail.com>
|
Thanks @cosmo0920 addressed both:
Rebuilt and re-tested end-to-end (records forwarded, non-object lines rejected, clean EOF). DCO-signed. |
Summary
Adds
in_aegisbpf, an input plugin that streams runtime-security events from a co-located AegisBPF agent into the Fluent Bit pipeline.AegisBPF is a BPF-LSM enforcement agent. It exposes an opt-in, root-only Unix control socket; sending
GET /eventsturns the connection into a newline-delimited stream of JSON (OCSF) security events. This plugin connects out to that socket, forwards each event as a record, and reconnects if the agent restarts.Design notes:
flb_input_set_collector_socket) and drains all available data on each wake. This matters because the agent uses non-blocking broadcast and drops slow readers — polling would risk being dropped under load.flb_pack_jsonand emitted via the log-event encoder. A per-line size cap guards against pathological input.Configuration
socket_path/var/run/aegisbpf/aegisbpf.sockreconnect_sec2(The AegisBPF agent must run with
AEGIS_API_SOCKETset; Fluent Bit must run as the socket owner, i.e. root.)Testing
Built into
fluent-bitand run against a mock agent (unix socket that sends the ack then JSON event lines). Debug log:The handshake ack is skipped, events become records with timestamps, and EOF/reconnect/pause paths work.
fluent/fluent-bit-docs— will follow if the plugin is acceptedEnabled by default (
FLB_IN_AEGISBPF); registered inplugins/CMakeLists.txtandcmake/plugins_options.cmake.Summary by CodeRabbit