Skip to content

input: fix threaded input conditional routing for logs #11783

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 15 commits into from
May 7, 2026
Merged

input: fix threaded input conditional routing for logs #11783

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
4e6ae9f
input_chunk: add local append path for threaded routing
edsiper May 7, 2026
bd30053
input_log: enable threaded conditional routing
edsiper May 7, 2026
9b7cfd1
input_chunk: process routed logs from ring buffer
edsiper May 7, 2026
4915f8e
tests: integration: cover threaded conditional routing
edsiper May 7, 2026
361eed7
input_log: preserve base direct routes after splitting
edsiper May 7, 2026
1326f37
tests: integration: extend conditional routing coverage
edsiper May 7, 2026
41e500d
out_opentelemetry: fix oauth2 config ownership
edsiper May 7, 2026
22086e7
tests: integration: cover grouped conditional routing
edsiper May 7, 2026
e1828b9
in_opentelemetry: fix log ingestion edge cases
edsiper May 7, 2026
31e77e1
out_opentelemetry: fix grouped log reassembly
edsiper May 7, 2026
c0fb084
opentelemetry: fix JSON log scope handling
edsiper May 7, 2026
04510b6
tests: integration: harden OpenTelemetry coverage
edsiper May 7, 2026
514cad5
in_opentelemetry: preserve dropped scope attributes count
edsiper May 7, 2026
8f861df
out_opentelemetry: handle group hash failures
edsiper May 7, 2026
83b1710
tests: integration: stabilize grouped OTLP matching
edsiper May 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions include/fluent-bit/flb_input_chunk.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,11 +124,21 @@ int flb_input_chunk_append_raw(struct flb_input_instance *in,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size);
int flb_input_chunk_append_raw_local(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size);
int flb_input_chunk_ring_buffer_enqueue(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size);
int flb_input_chunk_ring_buffer_enqueue_log_routing(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size);

const void *flb_input_chunk_flush(struct flb_input_chunk *ic, size_t *size);
int flb_input_chunk_release_lock(struct flb_input_chunk *ic);
Expand Down
7 changes: 7 additions & 0 deletions include/fluent-bit/flb_input_log.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,4 +38,11 @@ int flb_input_log_append_skip_processor_stages(struct flb_input_instance *ins,
size_t tag_len,
const void *buf,
size_t buf_size);

int flb_input_log_append_processed(struct flb_input_instance *ins,
size_t records,
const char *tag,
size_t tag_len,
const void *buf,
size_t buf_size);
#endif
44 changes: 39 additions & 5 deletions src/flb_input_chunk.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_input_chunk.h>
#include <fluent-bit/flb_input_log.h>
#include <fluent-bit/flb_input_plugin.h>
#include <fluent-bit/flb_storage.h>
#include <fluent-bit/flb_time.h>
Expand All @@ -53,9 +54,12 @@
#define FLB_INPUT_CHUNK_RELEASE_SCOPE_LOCAL 0
#define FLB_INPUT_CHUNK_RELEASE_SCOPE_GLOBAL 1

#define FLB_INPUT_CHUNK_RAW_LOG_ROUTING (1 << 0)

struct input_chunk_raw {
struct flb_input_instance *ins;
int event_type;
int flags;
size_t records;
flb_sds_t tag;
void *buf_data;
Expand Down Expand Up @@ -2977,6 +2981,7 @@ static void destroy_chunk_raw(struct input_chunk_raw *cr)

static int append_to_ring_buffer(struct flb_input_instance *ins,
int event_type,
int flags,
size_t records,
const char *tag,
size_t tag_len,
Expand All @@ -3001,6 +3006,7 @@ static int append_to_ring_buffer(struct flb_input_instance *ins,
}
cr->ins = ins;
cr->event_type = event_type;
cr->flags = flags;

if (tag && tag_len > 0) {
cr->tag = flb_sds_create_len(tag, tag_len);
Expand Down Expand Up @@ -3123,9 +3129,17 @@ void flb_input_chunk_ring_buffer_collector(struct flb_config *ctx, void *data)
tag_len = 0;
}

input_chunk_append_raw(cr->ins, cr->event_type, cr->records,
cr->tag, tag_len,
cr->buf_data, cr->buf_size);
if (cr->event_type == FLB_INPUT_LOGS &&
(cr->flags & FLB_INPUT_CHUNK_RAW_LOG_ROUTING) != 0) {
flb_input_log_append_processed(cr->ins, cr->records,
cr->tag, tag_len,
cr->buf_data, cr->buf_size);
}
else {
input_chunk_append_raw(cr->ins, cr->event_type, cr->records,
cr->tag, tag_len,
cr->buf_data, cr->buf_size);
}
Comment on lines +3132 to +3142

@coderabbitai coderabbitai Bot May 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Handle append failures in ring-buffer collector to avoid silent data loss.

The collector discards cr regardless of append result. If append fails, dequeued records are lost without explicit failure handling.

🛠️ Suggested fix 
 void flb_input_chunk_ring_buffer_collector(struct flb_config *ctx, void *data)
 {
     int ret;
+    int append_ret;
     int tag_len = 0;
@@
-                if (cr->event_type == FLB_INPUT_LOGS &&
-                    (cr->flags & FLB_INPUT_CHUNK_RAW_LOG_ROUTING) != 0) {
-                    flb_input_log_append_processed(cr->ins, cr->records,
-                                                   cr->tag, tag_len,
-                                                   cr->buf_data, cr->buf_size);
-                }
-                else {
-                    input_chunk_append_raw(cr->ins, cr->event_type, cr->records,
-                                           cr->tag, tag_len,
-                                           cr->buf_data, cr->buf_size);
-                }
+                if (cr->event_type == FLB_INPUT_LOGS &&
+                    (cr->flags & FLB_INPUT_CHUNK_RAW_LOG_ROUTING) != 0) {
+                    append_ret = flb_input_log_append_processed(cr->ins, cr->records,
+                                                                cr->tag, tag_len,
+                                                                cr->buf_data, cr->buf_size);
+                }
+                else {
+                    append_ret = input_chunk_append_raw(cr->ins, cr->event_type, cr->records,
+                                                        cr->tag, tag_len,
+                                                        cr->buf_data, cr->buf_size);
+                }
+
+                if (append_ret != 0) {
+                    flb_plg_error(ins, "failed to append dequeued ring-buffer payload");
+                }
                 destroy_chunk_raw(cr);
🤖 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 `@src/flb_input_chunk.c` around lines 3132 - 3142, The code unconditionally
drops cr even if appending fails; modify the calls to
flb_input_log_append_processed(...) and input_chunk_append_raw(...) to capture
their return values, and if the return indicates failure (non-zero), log an
error and re-enqueue cr back onto the collector ring using the same ring enqueue
API used where chunks are originally queued (do not free or discard cr on
failure); only free/acknowledge cr when the append succeeds.

destroy_chunk_raw(cr);
}
cr = NULL;
Expand All @@ -3148,7 +3162,7 @@ int flb_input_chunk_append_raw(struct flb_input_instance *in,
* add the data reference to the ring buffer.
*/
if (flb_input_is_threaded(in)) {
ret = append_to_ring_buffer(in, event_type, records,
ret = append_to_ring_buffer(in, event_type, 0, records,
tag, tag_len,
buf, buf_size);
}
Expand All @@ -3160,17 +3174,37 @@ int flb_input_chunk_append_raw(struct flb_input_instance *in,
return ret;
}

int flb_input_chunk_append_raw_local(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size)
{
return input_chunk_append_raw(in, event_type, records,
tag, tag_len, buf, buf_size);
}

int flb_input_chunk_ring_buffer_enqueue(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size)
{
return append_to_ring_buffer(in, event_type, records,
return append_to_ring_buffer(in, event_type, 0, records,
tag, tag_len,
buf, buf_size);
}

int flb_input_chunk_ring_buffer_enqueue_log_routing(struct flb_input_instance *in,
int event_type,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size)
{
return append_to_ring_buffer(in, event_type, FLB_INPUT_CHUNK_RAW_LOG_ROUTING,
records, tag, tag_len, buf, buf_size);
}

/* Retrieve a raw buffer from a dyntag node */
const void *flb_input_chunk_flush(struct flb_input_chunk *ic, size_t *size)
{
Expand Down
181 changes: 118 additions & 63 deletions src/flb_input_log.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -980,7 +980,8 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins,
const char *tag,
size_t tag_len,
const void *buf,
size_t buf_size)
size_t buf_size,
int local_append)
{
int ret;
int appended;
Expand Down Expand Up @@ -1018,13 +1019,6 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins,
return 0;
}

/* Conditional routing not supported for threaded inputs */
if (flb_input_is_threaded(ins)) {
flb_plg_warn(ins, "conditional routing not supported for threaded inputs, "
"falling back to normal routing");
return 0;
}

cfl_list_init(&payloads);
cfl_list_foreach(head, &ins->routes_direct) {
route_path = cfl_list_entry(head, struct flb_router_path, _head);
Expand Down Expand Up @@ -1249,13 +1243,24 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins,
continue;
}

ret = flb_input_chunk_append_raw(ins,
FLB_INPUT_LOGS,
payload->total_records,
payload->tag,
flb_sds_len(payload->tag),
payload->data,
payload->size);
if (local_append == FLB_TRUE) {
ret = flb_input_chunk_append_raw_local(ins,
FLB_INPUT_LOGS,
payload->total_records,
payload->tag,
flb_sds_len(payload->tag),
payload->data,
payload->size);
}
else {
ret = flb_input_chunk_append_raw(ins,
FLB_INPUT_LOGS,
payload->total_records,
payload->tag,
flb_sds_len(payload->tag),
payload->data,
payload->size);
}
if (ret != 0) {
flb_router_chunk_context_destroy(&context);
route_payload_list_destroy(&payloads);
Expand Down Expand Up @@ -1305,56 +1310,20 @@ static int split_and_append_route_payloads(struct flb_input_instance *ins,
return 0;
}

static int input_log_append(struct flb_input_instance *ins,
size_t processor_starting_stage,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size)
static int input_log_append_processed_internal(struct flb_input_instance *ins,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size,
int local_append)
{
int ret;
int conditional_result;
int conditional_handled = FLB_FALSE;
int processor_is_active;
void *out_buf = (void *) buf;
size_t dummy = 0;
size_t out_size = buf_size;
const char *base_tag = tag;
size_t base_tag_len = tag_len;
struct flb_input_chunk *chunk = NULL;

processor_is_active = flb_processor_is_active(ins->processor);
if (processor_is_active) {
if (!tag) {
if (ins->tag && ins->tag_len > 0) {
tag = ins->tag;
tag_len = ins->tag_len;
}
else {
tag = ins->name;
tag_len = strlen(ins->name);
}
}

ret = flb_processor_run(ins->processor,
processor_starting_stage,
FLB_PROCESSOR_LOGS,
tag, tag_len,
(char *) buf, buf_size,
&out_buf, &out_size);
if (ret == -1) {
return -1;
}

if (out_size == 0) {
return 0;
}

if (buf != out_buf) {
/* a new buffer was created, re-count the number of records */
records = flb_mp_count_log_records(out_buf, out_size);
}
}

if (!base_tag) {
if (ins->tag && ins->tag_len > 0) {
base_tag = ins->tag;
Expand All @@ -1370,11 +1339,9 @@ static int input_log_append(struct flb_input_instance *ins,
}

conditional_result = split_and_append_route_payloads(ins, records, tag, tag_len,
out_buf, out_size);
buf, buf_size,
local_append);
if (conditional_result < 0) {
if (processor_is_active && buf != out_buf) {
flb_free(out_buf);
}
return -1;
}

Expand All @@ -1387,8 +1354,14 @@ static int input_log_append(struct flb_input_instance *ins,
* receive data even when conditional routes exist. The conditional routing
* should be additive, not exclusive.
*/
ret = flb_input_chunk_append_raw(ins, FLB_INPUT_LOGS, records,
tag, tag_len, out_buf, out_size);
if (local_append == FLB_TRUE) {
ret = flb_input_chunk_append_raw_local(ins, FLB_INPUT_LOGS, records,
tag, tag_len, buf, buf_size);
}
else {
ret = flb_input_chunk_append_raw(ins, FLB_INPUT_LOGS, records,
tag, tag_len, buf, buf_size);
}

if (ret == 0 && conditional_handled == FLB_TRUE && base_tag) {
chunk = NULL;
Expand All @@ -1403,9 +1376,75 @@ static int input_log_append(struct flb_input_instance *ins,
}
}

return ret;
}

static int input_log_append(struct flb_input_instance *ins,
size_t processor_starting_stage,
size_t records,
const char *tag, size_t tag_len,
const void *buf, size_t buf_size)
{
int ret;
int processor_is_active;
void *out_buf = (void *) buf;
size_t out_size = buf_size;

processor_is_active = flb_processor_is_active(ins->processor);
if (processor_is_active) {
if (!tag) {
if (ins->tag && ins->tag_len > 0) {
tag = ins->tag;
tag_len = ins->tag_len;
}
else {
tag = ins->name;
tag_len = strlen(ins->name);
}
}

ret = flb_processor_run(ins->processor,
processor_starting_stage,
FLB_PROCESSOR_LOGS,
tag, tag_len,
(char *) buf, buf_size,
&out_buf, &out_size);
if (ret == -1) {
return -1;
}

if (out_size == 0) {
if (buf != out_buf) {
flb_free(out_buf);
}
return 0;
}

if (buf != out_buf) {
/* a new buffer was created, re-count the number of records */
records = flb_mp_count_log_records(out_buf, out_size);
}
}

if (flb_input_is_threaded(ins) == FLB_TRUE &&
input_has_conditional_routes(ins) == FLB_TRUE) {
ret = flb_input_chunk_ring_buffer_enqueue_log_routing(ins,
FLB_INPUT_LOGS,
records,
tag, tag_len,
out_buf, out_size);
}
Comment on lines +1585 to +1592

@coderabbitai coderabbitai Bot May 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

per_record_routing is being sent through a path that never builds payloads.

input_has_conditional_routes() also returns true for route->per_record_routing, so threaded inputs with that flag now take this enqueue path. But the splitter only creates route payloads when route->condition is present, which means per_record_routing records can be marked as handled here and then have their base route bits stripped later without ever producing a routed chunk. Please exclude per_record_routing from this new threaded path until it is explicitly supported here, or add the missing split logic.

🤖 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 `@src/flb_input_log.c` around lines 1585 - 1592, Threaded inputs are using the
enqueue-path guarded by flb_input_is_threaded and input_has_conditional_routes
but input_has_conditional_routes also returns true for route->per_record_routing
even though the splitter only builds payloads when route->condition is present;
this lets per_record_routing records be marked handled without ever being
split/routed. Fix by excluding per-record routing from this threaded path:
change the condition around flb_input_chunk_ring_buffer_enqueue_log_routing
(where flb_input_is_threaded(ins) and input_has_conditional_routes(ins) are
checked) to additionally ensure per-record routing is not present (e.g., add a
check like "&& !input_has_per_record_routing(ins)" or modify
input_has_conditional_routes to ignore route->per_record_routing), or
alternatively implement the missing split logic used by the splitter so
route->per_record_routing paths produce payloads before calling
flb_input_chunk_ring_buffer_enqueue_log_routing; update references to
flb_input_is_threaded, input_has_conditional_routes,
flb_input_chunk_ring_buffer_enqueue_log_routing, and route->per_record_routing
accordingly.

else {
ret = input_log_append_processed_internal(ins, records,
tag, tag_len,
out_buf, out_size,
FLB_FALSE);
}

if (processor_is_active && buf != out_buf) {
flb_free(out_buf);
}

return ret;
}

Expand All @@ -1422,6 +1461,23 @@ int flb_input_log_append(struct flb_input_instance *ins,
return ret;
}

int flb_input_log_append_processed(struct flb_input_instance *ins,
size_t records,
const char *tag,
size_t tag_len,
const void *buf,
size_t buf_size)
{
if (records == 0) {
records = flb_mp_count_log_records(buf, buf_size);
}

return input_log_append_processed_internal(ins, records,
tag, tag_len,
buf, buf_size,
FLB_TRUE);
}

/* Take a msgpack serialized record and enqueue it as a chunk */
int flb_input_log_append_skip_processor_stages(struct flb_input_instance *ins,
size_t processor_starting_stage,
Expand Down Expand Up @@ -1450,4 +1506,3 @@ int flb_input_log_append_records(struct flb_input_instance *ins,
ret = input_log_append(ins, 0, records, tag, tag_len, buf, buf_size);
return ret;
}

Loading
Loading