diff --git a/src/flb_router_config.c b/src/flb_router_config.c index 5a2cb88ea9d..8779bd0e5e8 100644 --- a/src/flb_router_config.c +++ b/src/flb_router_config.c @@ -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; @@ -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) { return FLB_TRUE; } } @@ -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; } diff --git a/src/flb_task.c b/src/flb_task.c index 928f75d5bb0..94112a0b104 100644 --- a/src/flb_task.c +++ b/src/flb_task.c @@ -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, @@ -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(); @@ -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(); diff --git a/tests/internal/router_config.c b/tests/internal/router_config.c index fb1ae28b4a8..90dbb484154 100644 --- a/tests/internal/router_config.c +++ b/tests/internal/router_config.c @@ -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; @@ -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 },