Skip to content

router_config: wire every route merging into a shared output #12142

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

Draft
zshuang0316 wants to merge 3 commits into
base: master
Choose a base branch
from
Draft

router_config: wire every route merging into a shared output #12142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ed24170
router_config: wire every route merging into a shared output
zshuang0316 Jul 23, 2026
d468bf5
task: keep one task route per output for merged routes
zshuang0316 Jul 23, 2026
7188d83
tests: internal: cover routes merging into a shared output
zshuang0316 Jul 23, 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
14 changes: 11 additions & 3 deletions src/flb_router_config.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1418,7 +1418,8 @@ static struct flb_output_instance *find_output_instance(struct flb_config *confi
}

static int input_has_direct_route(struct flb_input_instance *in,
struct flb_output_instance *out)
struct flb_output_instance *out,
struct flb_route *route)
{
struct cfl_list *head;
struct flb_router_path *path;
Expand All @@ -1427,9 +1428,16 @@ static int input_has_direct_route(struct flb_input_instance *in,
return FLB_FALSE;
}

/*
* A direct route is identified by the (route, output) pair, not the output
* alone: several routes may target the same output (e.g. different
* conditions merging into one output), and each is evaluated independently
* at runtime via its own flb_router_path. Deduplicating on the output only
* would drop all but the first route, so compare the route too.
*/
cfl_list_foreach(head, &in->routes_direct) {
path = cfl_list_entry(head, struct flb_router_path, _head);
if (path->ins == out) {
if (path->ins == out && path->route == route) {

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 23, 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.

P2 Badge Prevent merged routes from duplicating output flushes

When two conditional routes target the same output and their predicates can both match a record, this route-aware dedupe leaves both flb_router_paths in routes_direct, but the later routing mask is only keyed by output id. A route-specific chunk for either route therefore passes both paths in flb_task_create() and creates duplicate flb_task_route entries for the same output; the task status/data helpers then update only the first matching route->out, so delivery, retry, and accounting for that chunk can be duplicated or inconsistent. Please keep the per-route paths for condition evaluation, but collapse matched paths to one task route per output or carry route identity through the mask/task layer.

AGENTS.md reference: AGENTS.md:L224-L228

Useful? React with 👍 / 👎.

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 23, 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.

P1 Badge Preserve storage accounting for shared-output routes

With filesystem storage and total_limit_size, allowing multiple routes_direct entries for the same output breaks the existing route-mask accounting path because masks are keyed only by output id. In route_payload_apply_outputs(), a conditional/per-record payload chunk that was already fs_counted (for example when the output also has Match *) subtracts and clears any “other route” whose output bit is set; after this change, a second route to the same output matches that test, so the current output is subtracted even though the payload still routes there, and it is not added back because chunk->fs_counted remains true. This undercounts the output’s fs_chunks_size and can bypass storage pressure for shared-output conditional routes.

AGENTS.md reference: AGENTS.md:L239-L241

Useful? React with 👍 / 👎.

return FLB_TRUE;
}
}
Expand Down Expand Up @@ -1526,7 +1534,7 @@ int flb_router_apply_config(struct flb_config *config)

route_output->ins = output_ins;

if (input_has_direct_route(input_ins, output_ins)) {
if (input_has_direct_route(input_ins, output_ins, route)) {
continue;
}

Expand Down
35 changes: 35 additions & 0 deletions src/flb_task.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -543,6 +543,28 @@ int flb_task_map_get_task_id(struct flb_config *config) {
return map_get_task_id(config);
}

/*
* Several routes may resolve to the same output instance (e.g. distinct
* conditional routes merging into one output). Each route is evaluated
* independently, but a task must carry at most one route per output so the
* chunk is flushed and accounted for exactly once.
*/
static int task_route_exists(struct flb_task *task,
struct flb_output_instance *out)
{
struct mk_list *head;
struct flb_task_route *route;

mk_list_foreach(head, &task->routes) {
route = mk_list_entry(head, struct flb_task_route, _head);
if (route->out == out) {
return FLB_TRUE;
}
}

return FLB_FALSE;
}

/* Create an engine task to handle the output plugin flushing work */
struct flb_task *flb_task_create(uint64_t ref_id,
const char *buf,
Expand Down Expand Up @@ -700,6 +722,11 @@ struct flb_task *flb_task_create(uint64_t ref_id,
for (stored_match_index = 0;
stored_match_index < stored_match_count;
stored_match_index++) {
if (task_route_exists(task,
stored_matches[stored_match_index])) {
continue;
}

route = flb_calloc(1, sizeof(struct flb_task_route));
if (!route) {
flb_errno();
Expand Down Expand Up @@ -799,6 +826,14 @@ struct flb_task *flb_task_create(uint64_t ref_id,
}
}

/*
* Multiple matching routes can point at the same output; only add
* one task route per output so the chunk is not flushed twice.
*/
if (task_route_exists(task, o_ins)) {
continue;
}

route = flb_calloc(1, sizeof(struct flb_task_route));
if (!route) {
flb_errno();
Expand Down
100 changes: 100 additions & 0 deletions tests/internal/router_config.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1300,6 +1300,105 @@ void test_router_apply_config_success()
flb_sds_destroy(route_output.name);
}

/* Two distinct routes that merge into the same output instance must each be
* wired as an independent direct path. Deduplicating on the output instance
* alone would keep only the first route and silently drop the rest, which
* breaks per-route conditions targeting a shared output. A single route that
* lists the same output twice must still collapse to one path, exercising the
* identical (route, output) dedup branch. */
void test_router_apply_config_merged_output_keeps_all_routes()
{
struct flb_config config;
struct flb_input_instance input;
struct flb_output_instance output;
struct flb_input_routes input_routes;
struct flb_route route_one;
struct flb_route route_two;
struct flb_route_output route_output_one;
struct flb_route_output route_output_one_dup;
struct flb_route_output route_output_two;
struct flb_input_plugin input_plugin;
struct flb_output_plugin output_plugin;
struct cfl_list *head;
struct flb_router_path *path;
int seen_one = FLB_FALSE;
int seen_two = FLB_FALSE;

setup_test_instances(&config, &input, &input_plugin, "dummy", "dummy",
&output, &output_plugin, "printme", "stdout");

memset(&input_routes, 0, sizeof(input_routes));
cfl_list_init(&input_routes._head);
cfl_list_init(&input_routes.routes);
input_routes.input_name = flb_sds_create("dummy");
input_routes.plugin_name = flb_sds_create("dummy");
input_routes.has_alias = FLB_FALSE;
cfl_list_add(&input_routes._head, &config.input_routes);

memset(&route_one, 0, sizeof(route_one));
cfl_list_init(&route_one._head);
cfl_list_init(&route_one.outputs);
route_one.name = flb_sds_create("error_logs");
route_one.signals = FLB_ROUTER_SIGNAL_LOGS;
cfl_list_add(&route_one._head, &input_routes.routes);

memset(&route_output_one, 0, sizeof(route_output_one));
cfl_list_init(&route_output_one._head);
route_output_one.name = flb_sds_create("printme");
cfl_list_add(&route_output_one._head, &route_one.outputs);

/* Same route, same output listed twice: must collapse to one path */
memset(&route_output_one_dup, 0, sizeof(route_output_one_dup));
cfl_list_init(&route_output_one_dup._head);
route_output_one_dup.name = flb_sds_create("printme");
cfl_list_add(&route_output_one_dup._head, &route_one.outputs);

memset(&route_two, 0, sizeof(route_two));
cfl_list_init(&route_two._head);
cfl_list_init(&route_two.outputs);
route_two.name = flb_sds_create("checkout_logs");
route_two.signals = FLB_ROUTER_SIGNAL_LOGS;
cfl_list_add(&route_two._head, &input_routes.routes);

memset(&route_output_two, 0, sizeof(route_output_two));
cfl_list_init(&route_output_two._head);
route_output_two.name = flb_sds_create("printme");
cfl_list_add(&route_output_two._head, &route_two.outputs);

TEST_CHECK(flb_router_apply_config(&config) == 0);

/*
* route_one lists printme twice (collapses to one path) and route_two adds
* a second distinct path to the same output: two paths total.
*/
TEST_CHECK(cfl_list_size(&input.routes_direct) == 2);

cfl_list_foreach(head, &input.routes_direct) {
path = cfl_list_entry(head, struct flb_router_path, _head);
TEST_CHECK(path->ins == &output);
if (path->route == &route_one) {
seen_one = FLB_TRUE;
}
else if (path->route == &route_two) {
seen_two = FLB_TRUE;
}
}
TEST_CHECK(seen_one == FLB_TRUE);
TEST_CHECK(seen_two == FLB_TRUE);

flb_router_exit(&config);

flb_sds_destroy(input.alias);
flb_sds_destroy(output.alias);
flb_sds_destroy(input_routes.input_name);
flb_sds_destroy(input_routes.plugin_name);
flb_sds_destroy(route_one.name);
flb_sds_destroy(route_two.name);
flb_sds_destroy(route_output_one.name);
flb_sds_destroy(route_output_one_dup.name);
flb_sds_destroy(route_output_two.name);
}

void test_router_apply_config_missing_output()
{
struct flb_config config;
Expand Down Expand Up @@ -2637,6 +2736,7 @@ TEST_LIST = {
{ "parse_metrics_file", test_router_config_parse_file_metrics },
{ "parse_contexts_file", test_router_config_parse_file_contexts },
{ "apply_config_success", test_router_apply_config_success },
{ "apply_config_merged_output_keeps_all_routes", test_router_apply_config_merged_output_keeps_all_routes },
{ "apply_config_missing_output", test_router_apply_config_missing_output },
{ "apply_config_rejects_incompatible_output_signal", test_router_apply_config_rejects_incompatible_output_signal },
{ "apply_config_uses_input_alias", test_router_apply_config_uses_input_alias },
Expand Down
Loading