From 1f2f05d42945160d91bd0b2434eddafbf4777e6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Eren=20Ar=C4=B1?= Date: Wed, 12 Aug 2026 10:42:56 +0000 Subject: [PATCH 1/3] in_aegisbpf: add AegisBPF runtime-security input plugin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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ı --- cmake/plugins_options.cmake | 1 + plugins/CMakeLists.txt | 1 + plugins/in_aegisbpf/CMakeLists.txt | 4 + plugins/in_aegisbpf/in_aegisbpf.c | 456 +++++++++++++++++++++++++++++ plugins/in_aegisbpf/in_aegisbpf.h | 53 ++++ 5 files changed, 515 insertions(+) create mode 100644 plugins/in_aegisbpf/CMakeLists.txt create mode 100644 plugins/in_aegisbpf/in_aegisbpf.c create mode 100644 plugins/in_aegisbpf/in_aegisbpf.h diff --git a/cmake/plugins_options.cmake b/cmake/plugins_options.cmake index e6e12e1aee9..6a9249f1c86 100644 --- a/cmake/plugins_options.cmake +++ b/cmake/plugins_options.cmake @@ -11,6 +11,7 @@ option(FLB_MINIMAL "Enable minimal build configuration" No) # Inputs (sources, data collectors) # ================================= +DEFINE_OPTION(FLB_IN_AEGISBPF "Enable AegisBPF input plugin" ON) DEFINE_OPTION(FLB_IN_BLOB "Enable Blob input plugin" ON) DEFINE_OPTION(FLB_IN_CALYPTIA_FLEET "Enable Calyptia Fleet input plugin" ON) DEFINE_OPTION(FLB_IN_COLLECTD "Enable Collectd input plugin" ON) diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index af43a8fc2e4..45602c3eac0 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -245,6 +245,7 @@ endmacro() # Custom Plugins REGISTER_CUSTOM_PLUGIN("custom_calyptia") +REGISTER_IN_PLUGIN("in_aegisbpf") REGISTER_IN_PLUGIN("in_blob") # These plugins works only on Linux diff --git a/plugins/in_aegisbpf/CMakeLists.txt b/plugins/in_aegisbpf/CMakeLists.txt new file mode 100644 index 00000000000..dc0ef6d679e --- /dev/null +++ b/plugins/in_aegisbpf/CMakeLists.txt @@ -0,0 +1,4 @@ +set(src + in_aegisbpf.c) + +FLB_PLUGIN(in_aegisbpf "${src}" "") diff --git a/plugins/in_aegisbpf/in_aegisbpf.c b/plugins/in_aegisbpf/in_aegisbpf.c new file mode 100644 index 00000000000..0bc73c6c5b5 --- /dev/null +++ b/plugins/in_aegisbpf/in_aegisbpf.c @@ -0,0 +1,456 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* AegisBPF input plugin + * --------------------- + * Streams runtime-security events from a co-located AegisBPF agent into the + * Fluent Bit pipeline. AegisBPF (https://github.com/ErenAri/Aegis-BPF) is a + * BPF-LSM enforcement agent that exposes an opt-in, root-only Unix control + * socket; a "GET /events" request turns 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. + * + * The agent drops slow readers (its broadcast uses non-blocking sends), so the + * plugin drains the socket in an event-driven collector rather than polling. + */ + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "in_aegisbpf.h" + +static void aegisbpf_disconnect(struct flb_in_aegisbpf *ctx) +{ + if (ctx->coll_fd_read >= 0) { + flb_input_collector_delete(ctx->coll_fd_read, ctx->ins); + ctx->coll_fd_read = -1; + } + if (ctx->fd >= 0) { + close(ctx->fd); + ctx->fd = -1; + } + ctx->connected = 0; + ctx->handshake_done = 0; + ctx->buf_len = 0; +} + +static int write_all(int fd, const char *buf, size_t len) +{ + size_t off = 0; + while (off < len) { + ssize_t w = write(fd, buf + off, len - off); + if (w < 0) { + if (errno == EINTR) { + continue; + } + return -1; + } + off += (size_t) w; + } + return 0; +} + +/* Encode every complete newline-terminated JSON line held in ctx->buf into the + * event encoder, then compact the leftover partial line to the front. */ +static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx) +{ + size_t start = 0; + size_t i; + + for (i = 0; i < ctx->buf_len; i++) { + if (ctx->buf[i] != '\n') { + continue; + } + + { + char *line = ctx->buf + start; + size_t line_len = i - start; + + /* strip a trailing CR if present */ + if (line_len > 0 && line[line_len - 1] == '\r') { + line_len--; + } + + start = i + 1; + + /* The agent's first line is the streaming ack, not an event. */ + if (!ctx->handshake_done) { + ctx->handshake_done = 1; + continue; + } + if (line_len == 0) { + continue; + } + + { + char *mp = NULL; + size_t mp_size = 0; + int root_type = 0; + size_t consumed = 0; + int ret; + + ret = flb_pack_json(line, line_len, &mp, &mp_size, + &root_type, &consumed); + if (ret != 0 || mp == NULL) { + flb_plg_debug(ctx->ins, "skipping non-JSON line (%zu bytes)", + line_len); + if (mp != NULL) { + flb_free(mp); + } + continue; + } + + if (flb_log_event_encoder_begin_record(ctx->encoder) == + FLB_EVENT_ENCODER_SUCCESS) { + struct flb_time tm; + int r; + + flb_time_get(&tm); + flb_log_event_encoder_set_timestamp(ctx->encoder, &tm); + r = flb_log_event_encoder_set_body_from_raw_msgpack( + ctx->encoder, mp, mp_size); + if (r == FLB_EVENT_ENCODER_SUCCESS) { + flb_log_event_encoder_commit_record(ctx->encoder); + } + else { + flb_log_event_encoder_rollback_record(ctx->encoder); + } + } + flb_free(mp); + } + } + } + + if (start > 0) { + if (start < ctx->buf_len) { + memmove(ctx->buf, ctx->buf + start, ctx->buf_len - start); + } + ctx->buf_len -= start; + } +} + +/* Socket collector: drain all currently-available bytes, then flush records. */ +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; + + (void) config; + + flb_log_event_encoder_reset(ctx->encoder); + + 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); + } + + if (disconnected) { + aegisbpf_disconnect(ctx); + } + + return 0; +} + +static int aegisbpf_connect(struct flb_in_aegisbpf *ctx, + struct flb_config *config) +{ + struct sockaddr_un addr; + int fd; + int flags; + static const char req[] = "GET /events\n"; + + if (flb_sds_len(ctx->socket_path) >= sizeof(addr.sun_path)) { + flb_plg_error(ctx->ins, "socket_path too long: %s", ctx->socket_path); + return -1; + } + + fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) { + flb_errno(); + return -1; + } + + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, ctx->socket_path, sizeof(addr.sun_path) - 1); + + if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { + flb_plg_debug(ctx->ins, "connect(%s) failed: %s", + ctx->socket_path, strerror(errno)); + close(fd); + return -1; + } + + /* Request the event stream (blocking write; the request is tiny). */ + if (write_all(fd, req, sizeof(req) - 1) < 0) { + flb_plg_warn(ctx->ins, "failed to send stream request: %s", + strerror(errno)); + close(fd); + return -1; + } + + /* Non-blocking reads so the collector never stalls the engine. */ + flags = fcntl(fd, F_GETFL, 0); + if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) { + flb_errno(); + close(fd); + return -1; + } + + ctx->fd = fd; + ctx->connected = 1; + ctx->handshake_done = 0; + ctx->buf_len = 0; + + ctx->coll_fd_read = flb_input_set_collector_socket(ctx->ins, + in_aegisbpf_read, + fd, config); + if (ctx->coll_fd_read < 0) { + flb_plg_error(ctx->ins, "could not register read collector"); + aegisbpf_disconnect(ctx); + return -1; + } + if (flb_input_collector_start(ctx->coll_fd_read, ctx->ins) < 0) { + flb_plg_error(ctx->ins, "could not start read collector"); + aegisbpf_disconnect(ctx); + return -1; + } + + flb_plg_info(ctx->ins, "connected to AegisBPF at %s", ctx->socket_path); + return 0; +} + +/* Time collector: (re)establish the connection while disconnected. */ +static int in_aegisbpf_reconnect(struct flb_input_instance *ins, + struct flb_config *config, void *data) +{ + struct flb_in_aegisbpf *ctx = data; + + (void) ins; + + if (ctx->connected) { + return 0; + } + aegisbpf_connect(ctx, config); + return 0; +} + +static int in_aegisbpf_init(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + struct flb_in_aegisbpf *ctx; + int ret; + + (void) data; + + ctx = flb_calloc(1, sizeof(struct flb_in_aegisbpf)); + if (ctx == NULL) { + flb_errno(); + return -1; + } + ctx->ins = in; + ctx->fd = -1; + ctx->coll_fd_read = -1; + ctx->coll_fd_reconnect = -1; + + flb_input_set_context(in, ctx); + + if (flb_input_config_map_set(in, (void *) ctx) < 0) { + flb_plg_error(in, "unable to load configuration"); + flb_free(ctx); + return -1; + } + + if (ctx->reconnect_sec <= 0) { + ctx->reconnect_sec = FLB_IN_AEGISBPF_DEFAULT_RECONN; + } + + ctx->buf_size = FLB_IN_AEGISBPF_BUF_INIT; + ctx->buf = flb_malloc(ctx->buf_size); + if (ctx->buf == NULL) { + flb_errno(); + flb_free(ctx); + return -1; + } + + ctx->encoder = flb_log_event_encoder_create(FLB_LOG_EVENT_FORMAT_DEFAULT); + if (ctx->encoder == NULL) { + flb_plg_error(in, "could not initialize event encoder"); + flb_free(ctx->buf); + flb_free(ctx); + return -1; + } + + /* Drive (re)connection from a timer; the read collector is registered once + * a connection is established. */ + ret = flb_input_set_collector_time(in, in_aegisbpf_reconnect, + ctx->reconnect_sec, 0, config); + if (ret < 0) { + flb_plg_error(in, "could not register reconnect collector"); + flb_log_event_encoder_destroy(ctx->encoder); + flb_free(ctx->buf); + flb_free(ctx); + return -1; + } + ctx->coll_fd_reconnect = ret; + + return 0; +} + +static int in_aegisbpf_exit(void *data, struct flb_config *config) +{ + struct flb_in_aegisbpf *ctx = data; + + (void) config; + + if (ctx == NULL) { + return 0; + } + aegisbpf_disconnect(ctx); + if (ctx->encoder != NULL) { + flb_log_event_encoder_destroy(ctx->encoder); + } + if (ctx->buf != NULL) { + flb_free(ctx->buf); + } + flb_free(ctx); + return 0; +} + +static void in_aegisbpf_pause(void *data, struct flb_config *config) +{ + struct flb_in_aegisbpf *ctx = data; + + (void) config; + + if (ctx->coll_fd_reconnect >= 0) { + flb_input_collector_pause(ctx->coll_fd_reconnect, ctx->ins); + } + if (ctx->coll_fd_read >= 0) { + flb_input_collector_pause(ctx->coll_fd_read, ctx->ins); + } +} + +static void in_aegisbpf_resume(void *data, struct flb_config *config) +{ + struct flb_in_aegisbpf *ctx = data; + + (void) config; + + if (ctx->coll_fd_reconnect >= 0) { + flb_input_collector_resume(ctx->coll_fd_reconnect, ctx->ins); + } + if (ctx->coll_fd_read >= 0) { + flb_input_collector_resume(ctx->coll_fd_read, ctx->ins); + } +} + +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "socket_path", FLB_IN_AEGISBPF_DEFAULT_SOCKET, + 0, FLB_TRUE, offsetof(struct flb_in_aegisbpf, socket_path), + "Path to the AegisBPF control socket (root-only Unix stream socket)." + }, + { + FLB_CONFIG_MAP_INT, "reconnect_sec", "2", + 0, FLB_TRUE, offsetof(struct flb_in_aegisbpf, reconnect_sec), + "Interval in seconds between reconnection attempts." + }, + /* EOF */ + {0} +}; + +struct flb_input_plugin in_aegisbpf_plugin = { + .name = "aegisbpf", + .description = "AegisBPF runtime-security events", + .cb_init = in_aegisbpf_init, + .cb_pre_run = NULL, + .cb_collect = in_aegisbpf_reconnect, + .cb_flush_buf = NULL, + .config_map = config_map, + .cb_pause = in_aegisbpf_pause, + .cb_resume = in_aegisbpf_resume, + .cb_exit = in_aegisbpf_exit +}; diff --git a/plugins/in_aegisbpf/in_aegisbpf.h b/plugins/in_aegisbpf/in_aegisbpf.h new file mode 100644 index 00000000000..d96fe689cbe --- /dev/null +++ b/plugins/in_aegisbpf/in_aegisbpf.h @@ -0,0 +1,53 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FLB_IN_AEGISBPF_H +#define FLB_IN_AEGISBPF_H + +#include +#include +#include + +#define FLB_IN_AEGISBPF_DEFAULT_SOCKET "/var/run/aegisbpf/aegisbpf.sock" +#define FLB_IN_AEGISBPF_DEFAULT_RECONN 2 /* seconds */ +#define FLB_IN_AEGISBPF_BUF_INIT 16384 /* initial line-assembly buffer */ +#define FLB_IN_AEGISBPF_BUF_MAX (1024 * 1024) /* cap: drop a pathological line */ + +struct flb_in_aegisbpf { + /* config */ + flb_sds_t socket_path; /* AegisBPF control socket path */ + int reconnect_sec; /* reconnect interval */ + + /* connection state */ + int fd; /* stream socket fd, -1 when disconnected */ + int connected; + int handshake_done; /* the agent's first line is a streaming ack; skip it */ + int coll_fd_reconnect; /* time collector: (re)connect */ + int coll_fd_read; /* socket collector: drain events */ + + /* line assembly */ + char *buf; + size_t buf_size; + size_t buf_len; + + struct flb_log_event_encoder *encoder; + struct flb_input_instance *ins; +}; + +#endif From 152ccc2dbe37fc097478073e3d0c63935ada2f3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Eren=20Ar=C4=B1?= Date: Wed, 12 Aug 2026 14:29:35 +0000 Subject: [PATCH 2/3] in_aegisbpf: gate to Linux, validate single JSON object, bound drain loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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ı --- plugins/CMakeLists.txt | 2 +- plugins/in_aegisbpf/in_aegisbpf.c | 40 +++++++++++++++++++++++++++---- plugins/in_aegisbpf/in_aegisbpf.h | 2 ++ 3 files changed, 38 insertions(+), 6 deletions(-) diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index 45602c3eac0..914094d485f 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -245,11 +245,11 @@ endmacro() # Custom Plugins REGISTER_CUSTOM_PLUGIN("custom_calyptia") -REGISTER_IN_PLUGIN("in_aegisbpf") REGISTER_IN_PLUGIN("in_blob") # These plugins works only on Linux if(${CMAKE_SYSTEM_NAME} MATCHES "Linux") + REGISTER_IN_PLUGIN("in_aegisbpf") REGISTER_IN_PLUGIN("in_cpu") REGISTER_IN_PLUGIN("in_mem") REGISTER_IN_PLUGIN("in_thermal") diff --git a/plugins/in_aegisbpf/in_aegisbpf.c b/plugins/in_aegisbpf/in_aegisbpf.c index 0bc73c6c5b5..fc33b86fdd1 100644 --- a/plugins/in_aegisbpf/in_aegisbpf.c +++ b/plugins/in_aegisbpf/in_aegisbpf.c @@ -59,6 +59,7 @@ static void aegisbpf_disconnect(struct flb_in_aegisbpf *ctx) } ctx->connected = 0; ctx->handshake_done = 0; + ctx->skipping_line = 0; ctx->buf_len = 0; } @@ -85,7 +86,20 @@ static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx) size_t start = 0; size_t i; - for (i = 0; i < ctx->buf_len; i++) { + /* If a previous read dropped an oversized line, discard bytes up to and + * including the next newline so the tail of that line is never parsed as a + * (truncated) event. */ + if (ctx->skipping_line) { + char *nl = memchr(ctx->buf, '\n', ctx->buf_len); + if (nl == NULL) { + ctx->buf_len = 0; + return; + } + ctx->skipping_line = 0; + start = (size_t) (nl - ctx->buf) + 1; + } + + for (i = start; i < ctx->buf_len; i++) { if (ctx->buf[i] != '\n') { continue; } @@ -119,8 +133,13 @@ static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx) ret = flb_pack_json(line, line_len, &mp, &mp_size, &root_type, &consumed); - if (ret != 0 || mp == NULL) { - flb_plg_debug(ctx->ins, "skipping non-JSON line (%zu bytes)", + /* Accept only a single, whole JSON object per line: reject parse + * errors, arrays/scalars, and any trailing bytes after the object + * (which flb_pack_json would otherwise pack as extra roots). */ + if (ret != 0 || mp == NULL || + root_type != FLB_PACK_JSON_OBJECT || consumed != line_len) { + flb_plg_debug(ctx->ins, + "skipping line: not a single JSON object (%zu bytes)", line_len); if (mp != NULL) { flb_free(mp); @@ -163,6 +182,7 @@ static int in_aegisbpf_read(struct flb_input_instance *ins, { struct flb_in_aegisbpf *ctx = data; int disconnected = 0; + size_t drained = 0; (void) config; @@ -173,11 +193,14 @@ static int in_aegisbpf_read(struct flb_input_instance *ins, 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. */ + /* A single line exceeded the cap; drop the buffered head and mark + * the line for skipping so its remaining tail (still in the + * socket) is discarded up to the next newline rather than parsed + * as a truncated event. */ flb_plg_warn(ins, "line exceeded %d bytes, dropping", FLB_IN_AEGISBPF_BUF_MAX); ctx->buf_len = 0; - ctx->handshake_done = 1; + ctx->skipping_line = 1; } else { size_t new_size = ctx->buf_size * 2; @@ -200,6 +223,13 @@ static int in_aegisbpf_read(struct flb_input_instance *ins, if (n > 0) { ctx->buf_len += (size_t) n; aegisbpf_process_lines(ctx); + /* Bound work per wake so a continuously-writing agent can't hold the + * engine thread or grow the append arbitrarily large. The socket + * collector re-arms and continues on the next wake. */ + drained += (size_t) n; + if (drained >= FLB_IN_AEGISBPF_DRAIN_MAX) { + break; + } continue; } else if (n == 0) { diff --git a/plugins/in_aegisbpf/in_aegisbpf.h b/plugins/in_aegisbpf/in_aegisbpf.h index d96fe689cbe..f1fbc6e87f0 100644 --- a/plugins/in_aegisbpf/in_aegisbpf.h +++ b/plugins/in_aegisbpf/in_aegisbpf.h @@ -28,6 +28,7 @@ #define FLB_IN_AEGISBPF_DEFAULT_RECONN 2 /* seconds */ #define FLB_IN_AEGISBPF_BUF_INIT 16384 /* initial line-assembly buffer */ #define FLB_IN_AEGISBPF_BUF_MAX (1024 * 1024) /* cap: drop a pathological line */ +#define FLB_IN_AEGISBPF_DRAIN_MAX (4 * 1024 * 1024) /* max bytes drained per collector wake */ struct flb_in_aegisbpf { /* config */ @@ -38,6 +39,7 @@ struct flb_in_aegisbpf { int fd; /* stream socket fd, -1 when disconnected */ int connected; int handshake_done; /* the agent's first line is a streaming ack; skip it */ + int skipping_line; /* discarding the tail of an over-length line */ int coll_fd_reconnect; /* time collector: (re)connect */ int coll_fd_read; /* socket collector: drain events */ From abfa952a10768f3c861a2509409380bf43ced20e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Eren=20Ar=C4=B1?= Date: Thu, 13 Aug 2026 11:27:14 +0000 Subject: [PATCH 3/3] in_aegisbpf: declare variables at the top of functions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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ı --- plugins/in_aegisbpf/in_aegisbpf.c | 124 +++++++++++++++--------------- 1 file changed, 63 insertions(+), 61 deletions(-) diff --git a/plugins/in_aegisbpf/in_aegisbpf.c b/plugins/in_aegisbpf/in_aegisbpf.c index fc33b86fdd1..5d6cf1c17a2 100644 --- a/plugins/in_aegisbpf/in_aegisbpf.c +++ b/plugins/in_aegisbpf/in_aegisbpf.c @@ -66,8 +66,10 @@ static void aegisbpf_disconnect(struct flb_in_aegisbpf *ctx) static int write_all(int fd, const char *buf, size_t len) { size_t off = 0; + ssize_t w; + while (off < len) { - ssize_t w = write(fd, buf + off, len - off); + w = write(fd, buf + off, len - off); if (w < 0) { if (errno == EINTR) { continue; @@ -85,12 +87,22 @@ static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx) { size_t start = 0; size_t i; + char *nl; + char *line; + size_t line_len; + char *mp; + size_t mp_size; + int root_type; + size_t consumed; + int ret; + int r; + struct flb_time tm; /* If a previous read dropped an oversized line, discard bytes up to and * including the next newline so the tail of that line is never parsed as a * (truncated) event. */ if (ctx->skipping_line) { - char *nl = memchr(ctx->buf, '\n', ctx->buf_len); + nl = memchr(ctx->buf, '\n', ctx->buf_len); if (nl == NULL) { ctx->buf_len = 0; return; @@ -104,68 +116,58 @@ static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx) continue; } - { - char *line = ctx->buf + start; - size_t line_len = i - start; + line = ctx->buf + start; + line_len = i - start; - /* strip a trailing CR if present */ - if (line_len > 0 && line[line_len - 1] == '\r') { - line_len--; - } + /* strip a trailing CR if present */ + if (line_len > 0 && line[line_len - 1] == '\r') { + line_len--; + } - start = i + 1; + start = i + 1; - /* The agent's first line is the streaming ack, not an event. */ - if (!ctx->handshake_done) { - ctx->handshake_done = 1; - continue; - } - if (line_len == 0) { - continue; - } - - { - char *mp = NULL; - size_t mp_size = 0; - int root_type = 0; - size_t consumed = 0; - int ret; - - ret = flb_pack_json(line, line_len, &mp, &mp_size, - &root_type, &consumed); - /* Accept only a single, whole JSON object per line: reject parse - * errors, arrays/scalars, and any trailing bytes after the object - * (which flb_pack_json would otherwise pack as extra roots). */ - if (ret != 0 || mp == NULL || - root_type != FLB_PACK_JSON_OBJECT || consumed != line_len) { - flb_plg_debug(ctx->ins, - "skipping line: not a single JSON object (%zu bytes)", - line_len); - if (mp != NULL) { - flb_free(mp); - } - continue; - } + /* The agent's first line is the streaming ack, not an event. */ + if (!ctx->handshake_done) { + ctx->handshake_done = 1; + continue; + } + if (line_len == 0) { + continue; + } - if (flb_log_event_encoder_begin_record(ctx->encoder) == - FLB_EVENT_ENCODER_SUCCESS) { - struct flb_time tm; - int r; - - flb_time_get(&tm); - flb_log_event_encoder_set_timestamp(ctx->encoder, &tm); - r = flb_log_event_encoder_set_body_from_raw_msgpack( - ctx->encoder, mp, mp_size); - if (r == FLB_EVENT_ENCODER_SUCCESS) { - flb_log_event_encoder_commit_record(ctx->encoder); - } - else { - flb_log_event_encoder_rollback_record(ctx->encoder); - } - } + mp = NULL; + mp_size = 0; + root_type = 0; + consumed = 0; + ret = flb_pack_json(line, line_len, &mp, &mp_size, &root_type, &consumed); + /* Accept only a single, whole JSON object per line: reject parse errors, + * arrays/scalars, and any trailing bytes after the object (which + * flb_pack_json would otherwise pack as extra roots). */ + if (ret != 0 || mp == NULL || + root_type != FLB_PACK_JSON_OBJECT || consumed != line_len) { + flb_plg_debug(ctx->ins, + "skipping line: not a single JSON object (%zu bytes)", + line_len); + if (mp != NULL) { flb_free(mp); } + continue; } + + if (flb_log_event_encoder_begin_record(ctx->encoder) == + FLB_EVENT_ENCODER_SUCCESS) { + flb_time_get(&tm); + flb_log_event_encoder_set_timestamp(ctx->encoder, &tm); + r = flb_log_event_encoder_set_body_from_raw_msgpack(ctx->encoder, + mp, mp_size); + if (r == FLB_EVENT_ENCODER_SUCCESS) { + flb_log_event_encoder_commit_record(ctx->encoder); + } + else { + flb_log_event_encoder_rollback_record(ctx->encoder); + } + } + flb_free(mp); } if (start > 0) { @@ -183,14 +185,15 @@ static int in_aegisbpf_read(struct flb_input_instance *ins, struct flb_in_aegisbpf *ctx = data; int disconnected = 0; size_t drained = 0; + ssize_t n; + size_t new_size; + char *tmp; (void) config; flb_log_event_encoder_reset(ctx->encoder); 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 the buffered head and mark @@ -203,8 +206,7 @@ static int in_aegisbpf_read(struct flb_input_instance *ins, ctx->skipping_line = 1; } else { - size_t new_size = ctx->buf_size * 2; - char *tmp; + new_size = ctx->buf_size * 2; if (new_size > FLB_IN_AEGISBPF_BUF_MAX) { new_size = FLB_IN_AEGISBPF_BUF_MAX; }