From 5dadedd3a21526d62a1f3043711e99bba0bab0cd Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 10 Aug 2026 17:29:24 +0900 Subject: [PATCH 1/6] engine_dispatch: Implement stale-task cleanup Signed-off-by: Hiroshi Hatake --- src/flb_engine_dispatch.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/flb_engine_dispatch.c b/src/flb_engine_dispatch.c index 701ada3edb7..8ba9d75dcc9 100644 --- a/src/flb_engine_dispatch.c +++ b/src/flb_engine_dispatch.c @@ -69,6 +69,7 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry, /* Could not retrieve chunk content */ flb_error("[engine_dispatch] could not retrieve chunk content, removing retry"); flb_task_retry_destroy(retry); + flb_task_users_release(task); return -1; } From e01062fbe6b9a11024acde7794af46e4bcbf2964 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 10 Aug 2026 17:30:22 +0900 Subject: [PATCH 2/6] tests: internal: Add a regression test case for stale tasks Signed-off-by: Hiroshi Hatake --- tests/internal/CMakeLists.txt | 6 + tests/internal/engine_dispatch.c | 354 +++++++++++++++++++++++++++++++ 2 files changed, 360 insertions(+) create mode 100644 tests/internal/engine_dispatch.c diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt index 746d5039342..b66e561505c 100644 --- a/tests/internal/CMakeLists.txt +++ b/tests/internal/CMakeLists.txt @@ -64,6 +64,7 @@ set(UNIT_TESTS_FILES opentelemetry.c storage_dlq.c engine_adaptive_flush.c + engine_dispatch.c ) if(FLB_OUT_AZURE_BLOB) @@ -279,6 +280,11 @@ endfunction(prepare_unit_tests) prepare_unit_tests(flb-it- "${UNIT_TESTS_FILES}") +if(TARGET flb-it-engine_dispatch) + target_include_directories(flb-it-engine_dispatch PRIVATE + ${PROJECT_SOURCE_DIR}/lib/chunkio/deps) +endif() + option(FLB_TESTS_FIPS_ENABLED "Run tests that require an installed OpenSSL FIPS provider" OFF) if(TARGET flb-it-fips) if(FLB_OUT_S3) diff --git a/tests/internal/engine_dispatch.c b/tests/internal/engine_dispatch.c new file mode 100644 index 00000000000..e78afb52cea --- /dev/null +++ b/tests/internal/engine_dispatch.c @@ -0,0 +1,354 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "flb_tests_internal.h" + +struct test_ctx { + struct flb_config *config; + struct cio_ctx *cio; + struct flb_input_instance *input; +}; + +static void test_ctx_destroy(struct test_ctx *ctx) +{ + if (ctx == NULL) { + return; + } + + if (ctx->config != NULL) { + flb_input_exit_all(ctx->config); + } + + if (ctx->cio != NULL) { + cio_destroy(ctx->cio); + ctx->config->cio = NULL; + } + + if (ctx->config != NULL) { + flb_config_exit(ctx->config); + } + + flb_free(ctx); +} + +static struct test_ctx *test_ctx_create(void) +{ + int ret; + struct test_ctx *ctx; + struct cio_options options; +#ifdef _WIN32 + WSADATA wsa_data; +#endif + + ctx = flb_calloc(1, sizeof(struct test_ctx)); + if (ctx == NULL) { + flb_errno(); + return NULL; + } + + ctx->config = flb_config_init(); + if (ctx->config == NULL) { + test_ctx_destroy(ctx); + return NULL; + } + +#ifdef _WIN32 + WSAStartup(0x0201, &wsa_data); +#endif + + ctx->config->evl = mk_event_loop_create(8); + if (ctx->config->evl == NULL) { + test_ctx_destroy(ctx); + return NULL; + } + + ctx->config->sched = flb_sched_create(ctx->config, ctx->config->evl); + if (ctx->config->sched == NULL) { + test_ctx_destroy(ctx); + return NULL; + } + + cio_options_init(&options); + options.flags = CIO_OPEN; + ctx->cio = cio_create(&options); + if (ctx->cio == NULL) { + test_ctx_destroy(ctx); + return NULL; + } + ctx->config->cio = ctx->cio; + + ctx->input = flb_input_new(ctx->config, "dummy", NULL, FLB_FALSE); + if (ctx->input == NULL) { + test_ctx_destroy(ctx); + return NULL; + } + + ret = flb_storage_input_create(ctx->cio, ctx->input); + if (ret != 0) { + test_ctx_destroy(ctx); + return NULL; + } + + return ctx; +} + +static struct flb_task_retry *create_retry_dispatch_task( + struct test_ctx *ctx, + struct flb_output_instance *output, + int *task_id, + char **chunk_buffer) +{ + struct cio_memfs *memfs; + struct flb_input_chunk *chunk; + struct flb_task *task; + struct flb_task_route *route; + struct flb_task_retry *retry; + + chunk = flb_input_chunk_create(ctx->input, FLB_INPUT_LOGS, "test", 4); + if (chunk == NULL) { + return NULL; + } + + task = task_alloc(ctx->config); + if (task == NULL) { + flb_input_chunk_destroy(chunk, FLB_TRUE); + return NULL; + } + + task->i_ins = ctx->input; + task->ic = chunk; + chunk->task = task; + chunk->busy = FLB_TRUE; + mk_list_add(&task->_head, &ctx->input->tasks); + + route = flb_calloc(1, sizeof(struct flb_task_route)); + if (route == NULL) { + flb_task_destroy(task, FLB_TRUE); + return NULL; + } + route->out = output; + route->status = FLB_TASK_ROUTE_INACTIVE; + mk_list_add(&route->_head, &task->routes); + + retry = flb_calloc(1, sizeof(struct flb_task_retry)); + if (retry == NULL) { + flb_task_destroy(task, FLB_TRUE); + return NULL; + } + retry->attempts = 1; + retry->o_ins = output; + retry->parent = task; + mk_list_add(&retry->_head, &task->retries); + + /* Force flb_input_chunk_flush() to return NULL without altering production code. */ + memfs = ((struct cio_chunk *) chunk->chunk)->backend; + *chunk_buffer = memfs->buf_data; + memfs->buf_data = NULL; + *task_id = task->id; + + return retry; +} + +static void test_retry_flush_failure_releases_last_task_owner(void) +{ + int ret; + int task_id; + char *chunk_buffer; + struct test_ctx *ctx; + struct flb_task *replacement; + struct flb_task_retry *retry; + struct flb_output_instance output; + + ctx = test_ctx_create(); + TEST_CHECK(ctx != NULL); + if (ctx == NULL) { + return; + } + + memset(&output, 0, sizeof(output)); + retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer); + TEST_CHECK(retry != NULL); + if (retry == NULL) { + test_ctx_destroy(ctx); + return; + } + + ret = flb_engine_dispatch_retry(retry, ctx->config); + TEST_CHECK(ret == -1); + TEST_CHECK(ctx->config->task_map[task_id].task == NULL); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + + free(chunk_buffer); + + replacement = task_alloc(ctx->config); + TEST_CHECK(replacement != NULL); + if (replacement != NULL) { + TEST_CHECK(replacement->id == task_id); + flb_task_destroy(replacement, FLB_TRUE); + } + + test_ctx_destroy(ctx); +} + +static void test_retry_flush_failure_preserves_active_task_owner(void) +{ + int ret; + int task_id; + char *chunk_buffer; + struct cio_memfs *memfs; + struct test_ctx *ctx; + struct flb_input_chunk *chunk; + struct flb_task *task; + struct flb_task_retry *retry; + struct flb_output_instance output; + + ctx = test_ctx_create(); + TEST_CHECK(ctx != NULL); + if (ctx == NULL) { + return; + } + + memset(&output, 0, sizeof(output)); + retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer); + TEST_CHECK(retry != NULL); + if (retry == NULL) { + test_ctx_destroy(ctx); + return; + } + + task = retry->parent; + task->users = 1; + + ret = flb_engine_dispatch_retry(retry, ctx->config); + TEST_CHECK(ret == -1); + TEST_CHECK(ctx->config->task_map[task_id].task == task); + TEST_CHECK(task->users == 1); + TEST_CHECK(mk_list_size(&task->retries) == 0); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1); + + if (ctx->config->task_map[task_id].task == task) { + chunk = task->ic; + memfs = ((struct cio_chunk *) chunk->chunk)->backend; + memfs->buf_data = chunk_buffer; + task->users = 0; + flb_task_users_release(task); + } + else { + free(chunk_buffer); + } + + TEST_CHECK(ctx->config->task_map[task_id].task == NULL); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + + test_ctx_destroy(ctx); +} + +static void test_retry_flush_failure_preserves_pending_retry(void) +{ + int ret; + int task_id; + char *chunk_buffer; + struct cio_memfs *memfs; + struct test_ctx *ctx; + struct flb_input_chunk *chunk; + struct flb_task *task; + struct flb_task_route *route; + struct flb_task_retry *remaining_retry; + struct flb_task_retry *retry; + struct flb_output_instance output_a; + struct flb_output_instance output_b; + + ctx = test_ctx_create(); + TEST_CHECK(ctx != NULL); + if (ctx == NULL) { + return; + } + + memset(&output_a, 0, sizeof(output_a)); + memset(&output_b, 0, sizeof(output_b)); + retry = create_retry_dispatch_task(ctx, &output_a, &task_id, &chunk_buffer); + TEST_CHECK(retry != NULL); + if (retry == NULL) { + test_ctx_destroy(ctx); + return; + } + task = retry->parent; + + route = flb_calloc(1, sizeof(struct flb_task_route)); + remaining_retry = flb_calloc(1, sizeof(struct flb_task_retry)); + TEST_CHECK(route != NULL); + TEST_CHECK(remaining_retry != NULL); + if (route == NULL || remaining_retry == NULL) { + flb_free(route); + flb_free(remaining_retry); + chunk = task->ic; + memfs = ((struct cio_chunk *) chunk->chunk)->backend; + memfs->buf_data = chunk_buffer; + flb_task_destroy(task, FLB_TRUE); + test_ctx_destroy(ctx); + return; + } + + route->out = &output_b; + route->status = FLB_TASK_ROUTE_INACTIVE; + mk_list_add(&route->_head, &task->routes); + + remaining_retry->attempts = 1; + remaining_retry->o_ins = &output_b; + remaining_retry->parent = task; + mk_list_add(&remaining_retry->_head, &task->retries); + + ret = flb_engine_dispatch_retry(retry, ctx->config); + TEST_CHECK(ret == -1); + TEST_CHECK(ctx->config->task_map[task_id].task == task); + TEST_CHECK(task->users == 0); + TEST_CHECK(mk_list_size(&task->retries) == 1); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1); + + if (ctx->config->task_map[task_id].task == task) { + chunk = task->ic; + memfs = ((struct cio_chunk *) chunk->chunk)->backend; + memfs->buf_data = chunk_buffer; + flb_task_retry_destroy(remaining_retry); + flb_task_users_release(task); + } + else { + free(chunk_buffer); + } + + TEST_CHECK(ctx->config->task_map[task_id].task == NULL); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + + test_ctx_destroy(ctx); +} + +TEST_LIST = { + { "retry_flush_failure_releases_last_task_owner", + test_retry_flush_failure_releases_last_task_owner }, + { "retry_flush_failure_preserves_active_task_owner", + test_retry_flush_failure_preserves_active_task_owner }, + { "retry_flush_failure_preserves_pending_retry", + test_retry_flush_failure_preserves_pending_retry }, + { 0 } +}; From 106e1e2c9e46dc1df301ebdc4514671a5420d108 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 10 Aug 2026 18:40:52 +0900 Subject: [PATCH 3/6] engine_dispatch: Record retrying failures Signed-off-by: Hiroshi Hatake --- src/flb_engine_dispatch.c | 51 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/src/flb_engine_dispatch.c b/src/flb_engine_dispatch.c index 8ba9d75dcc9..968bc6cf6ad 100644 --- a/src/flb_engine_dispatch.c +++ b/src/flb_engine_dispatch.c @@ -29,9 +29,59 @@ #include #include #include +#include #include +#include +static void record_retry_failure_metrics(struct flb_task *task, + struct flb_output_instance *output, + struct flb_config *config) +{ + int effective_records; + size_t effective_bytes; + uint64_t timestamp; + char *input_name; + char *output_name; + + effective_records = 0; + effective_bytes = 0; + timestamp = cfl_time_now(); + input_name = (char *) flb_input_name(task->i_ins); + output_name = (char *) flb_output_name(output); + + flb_task_acquire_lock(task); + if (flb_task_get_route_data(task, output, + &effective_records, + &effective_bytes) != 0 && + task->event_chunk != NULL) { + effective_records = task->event_chunk->total_events; + effective_bytes = task->event_chunk->size; + } + flb_task_release_lock(task); + + cmt_counter_inc(output->cmt_retries_failed, timestamp, + 1, (char *[]) {output_name}); + cmt_counter_add(output->cmt_dropped_records, timestamp, effective_records, + 1, (char *[]) {output_name}); + + if (config->router && task->event_chunk && + task->event_chunk->type == FLB_EVENT_TYPE_LOGS) { + cmt_counter_add(config->router->logs_drop_records_total, timestamp, + effective_records, + 2, (char *[]) {input_name, output_name}); + cmt_counter_add(config->router->logs_drop_bytes_total, timestamp, + effective_bytes, + 2, (char *[]) {input_name, output_name}); + } + +#ifdef FLB_HAVE_METRICS + flb_metrics_sum(FLB_METRIC_OUT_RETRY_FAILED, 1, output->metrics); + flb_metrics_sum(FLB_METRIC_OUT_DROPPED_RECORDS, + effective_records, output->metrics); +#endif +} + /* It creates a new output thread using a 'Retry' context */ int flb_engine_dispatch_retry(struct flb_task_retry *retry, struct flb_config *config) @@ -68,6 +118,7 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry, if (!buf_data) { /* Could not retrieve chunk content */ flb_error("[engine_dispatch] could not retrieve chunk content, removing retry"); + record_retry_failure_metrics(task, retry->o_ins, config); flb_task_retry_destroy(retry); flb_task_users_release(task); return -1; From 3b29e2a2bbaf4a1e19d1b05e65b7091ff16e69ce Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 10 Aug 2026 18:41:27 +0900 Subject: [PATCH 4/6] tests: internal: Add assertions for the number of failing retries Signed-off-by: Hiroshi Hatake --- tests/internal/engine_dispatch.c | 185 ++++++++++++++++++++++++++++++- 1 file changed, 180 insertions(+), 5 deletions(-) diff --git a/tests/internal/engine_dispatch.c b/tests/internal/engine_dispatch.c index e78afb52cea..f164d0b9803 100644 --- a/tests/internal/engine_dispatch.c +++ b/tests/internal/engine_dispatch.c @@ -5,10 +5,13 @@ #include #include +#include #include #include #include +#include #include +#include #include #include #include @@ -19,10 +22,17 @@ #include "flb_tests_internal.h" +#define TEST_ROUTE_RECORDS 7 +#define TEST_ROUTE_BYTES 29 +#define TEST_CHUNK_RECORDS 13 + struct test_ctx { struct flb_config *config; struct cio_ctx *cio; struct flb_input_instance *input; +#ifdef _WIN32 + int winsock_initialized; +#endif }; static void test_ctx_destroy(struct test_ctx *ctx) @@ -44,6 +54,13 @@ static void test_ctx_destroy(struct test_ctx *ctx) flb_config_exit(ctx->config); } +#ifdef _WIN32 + if (ctx->winsock_initialized == FLB_TRUE) { + WSACleanup(); + ctx->winsock_initialized = FLB_FALSE; + } +#endif + flb_free(ctx); } @@ -69,7 +86,12 @@ static struct test_ctx *test_ctx_create(void) } #ifdef _WIN32 - WSAStartup(0x0201, &wsa_data); + ret = WSAStartup(0x0201, &wsa_data); + if (ret != 0) { + test_ctx_destroy(ctx); + return NULL; + } + ctx->winsock_initialized = FLB_TRUE; #endif ctx->config->evl = mk_event_loop_create(8); @@ -108,6 +130,121 @@ static struct test_ctx *test_ctx_create(void) return ctx; } +static int test_output_init(struct flb_output_instance *output, const char *name) +{ + memset(output, 0, sizeof(struct flb_output_instance)); + strncpy(output->name, name, sizeof(output->name) - 1); + + output->cmt = cmt_create(); + if (output->cmt == NULL) { + return -1; + } + + output->cmt_retries_failed = cmt_counter_create(output->cmt, + "fluentbit", + "output", + "retries_failed_total", + "Failed retries", + 1, + (char *[]) {"name"}); + output->cmt_dropped_records = cmt_counter_create(output->cmt, + "fluentbit", + "output", + "dropped_records_total", + "Dropped records", + 1, + (char *[]) {"name"}); + if (output->cmt_retries_failed == NULL || + output->cmt_dropped_records == NULL) { + cmt_destroy(output->cmt); + output->cmt = NULL; + return -1; + } + +#ifdef FLB_HAVE_METRICS + output->metrics = flb_metrics_create(name); + if (output->metrics == NULL || + flb_metrics_add(FLB_METRIC_OUT_RETRY_FAILED, + "retries_failed", output->metrics) == -1 || + flb_metrics_add(FLB_METRIC_OUT_DROPPED_RECORDS, + "dropped_records", output->metrics) == -1) { + if (output->metrics != NULL) { + flb_metrics_destroy(output->metrics); + output->metrics = NULL; + } + cmt_destroy(output->cmt); + output->cmt = NULL; + return -1; + } +#endif + + return 0; +} + +static void test_output_destroy(struct flb_output_instance *output) +{ +#ifdef FLB_HAVE_METRICS + if (output->metrics != NULL) { + flb_metrics_destroy(output->metrics); + output->metrics = NULL; + } +#endif + + if (output->cmt != NULL) { + cmt_destroy(output->cmt); + output->cmt = NULL; + } +} + +static void check_retry_failure_metrics(struct test_ctx *ctx, + struct flb_output_instance *output) +{ + int ret; + double value; + char *input_name; + char *output_name; +#ifdef FLB_HAVE_METRICS + struct flb_metric *metric; +#endif + + input_name = (char *) flb_input_name(ctx->input); + output_name = (char *) flb_output_name(output); + + ret = cmt_counter_get_val(output->cmt_retries_failed, + 1, (char *[]) {output_name}, &value); + TEST_CHECK(ret == 0); + TEST_CHECK(value == 1); + + ret = cmt_counter_get_val(output->cmt_dropped_records, + 1, (char *[]) {output_name}, &value); + TEST_CHECK(ret == 0); + TEST_CHECK(value == TEST_ROUTE_RECORDS); + + ret = cmt_counter_get_val(ctx->config->router->logs_drop_records_total, + 2, (char *[]) {input_name, output_name}, &value); + TEST_CHECK(ret == 0); + TEST_CHECK(value == TEST_ROUTE_RECORDS); + + ret = cmt_counter_get_val(ctx->config->router->logs_drop_bytes_total, + 2, (char *[]) {input_name, output_name}, &value); + TEST_CHECK(ret == 0); + TEST_CHECK(value == TEST_ROUTE_BYTES); + +#ifdef FLB_HAVE_METRICS + metric = flb_metrics_get_id(FLB_METRIC_OUT_RETRY_FAILED, output->metrics); + TEST_CHECK(metric != NULL); + if (metric != NULL) { + TEST_CHECK(metric->val == 1); + } + + metric = flb_metrics_get_id(FLB_METRIC_OUT_DROPPED_RECORDS, output->metrics); + TEST_CHECK(metric != NULL); + if (metric != NULL) { + TEST_CHECK(metric->val == TEST_ROUTE_RECORDS); + } +#endif +} + static struct flb_task_retry *create_retry_dispatch_task( struct test_ctx *ctx, struct flb_output_instance *output, @@ -137,6 +274,17 @@ static struct flb_task_retry *create_retry_dispatch_task( chunk->busy = FLB_TRUE; mk_list_add(&task->_head, &ctx->input->tasks); + memfs = ((struct cio_chunk *) chunk->chunk)->backend; + task->event_chunk = flb_event_chunk_create(FLB_EVENT_TYPE_LOGS, + TEST_CHUNK_RECORDS, + "test", 4, + memfs->buf_data, + memfs->buf_len); + if (task->event_chunk == NULL) { + flb_task_destroy(task, FLB_TRUE); + return NULL; + } + route = flb_calloc(1, sizeof(struct flb_task_route)); if (route == NULL) { flb_task_destroy(task, FLB_TRUE); @@ -144,6 +292,8 @@ static struct flb_task_retry *create_retry_dispatch_task( } route->out = output; route->status = FLB_TASK_ROUTE_INACTIVE; + route->records = TEST_ROUTE_RECORDS; + route->bytes = TEST_ROUTE_BYTES; mk_list_add(&route->_head, &task->routes); retry = flb_calloc(1, sizeof(struct flb_task_retry)); @@ -157,7 +307,6 @@ static struct flb_task_retry *create_retry_dispatch_task( mk_list_add(&retry->_head, &task->retries); /* Force flb_input_chunk_flush() to return NULL without altering production code. */ - memfs = ((struct cio_chunk *) chunk->chunk)->backend; *chunk_buffer = memfs->buf_data; memfs->buf_data = NULL; *task_id = task->id; @@ -181,10 +330,16 @@ static void test_retry_flush_failure_releases_last_task_owner(void) return; } - memset(&output, 0, sizeof(output)); + ret = test_output_init(&output, "output_a"); + TEST_CHECK(ret == 0); + if (ret != 0) { + test_ctx_destroy(ctx); + return; + } retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer); TEST_CHECK(retry != NULL); if (retry == NULL) { + test_output_destroy(&output); test_ctx_destroy(ctx); return; } @@ -194,6 +349,7 @@ static void test_retry_flush_failure_releases_last_task_owner(void) TEST_CHECK(ctx->config->task_map[task_id].task == NULL); TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + check_retry_failure_metrics(ctx, &output); free(chunk_buffer); @@ -204,6 +360,7 @@ static void test_retry_flush_failure_releases_last_task_owner(void) flb_task_destroy(replacement, FLB_TRUE); } + test_output_destroy(&output); test_ctx_destroy(ctx); } @@ -225,10 +382,16 @@ static void test_retry_flush_failure_preserves_active_task_owner(void) return; } - memset(&output, 0, sizeof(output)); + ret = test_output_init(&output, "output_a"); + TEST_CHECK(ret == 0); + if (ret != 0) { + test_ctx_destroy(ctx); + return; + } retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer); TEST_CHECK(retry != NULL); if (retry == NULL) { + test_output_destroy(&output); test_ctx_destroy(ctx); return; } @@ -243,6 +406,7 @@ static void test_retry_flush_failure_preserves_active_task_owner(void) TEST_CHECK(mk_list_size(&task->retries) == 0); TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1); TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1); + check_retry_failure_metrics(ctx, &output); if (ctx->config->task_map[task_id].task == task) { chunk = task->ic; @@ -259,6 +423,7 @@ static void test_retry_flush_failure_preserves_active_task_owner(void) TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + test_output_destroy(&output); test_ctx_destroy(ctx); } @@ -283,11 +448,18 @@ static void test_retry_flush_failure_preserves_pending_retry(void) return; } - memset(&output_a, 0, sizeof(output_a)); + ret = test_output_init(&output_a, "output_a"); + TEST_CHECK(ret == 0); + if (ret != 0) { + test_ctx_destroy(ctx); + return; + } memset(&output_b, 0, sizeof(output_b)); + strncpy(output_b.name, "output_b", sizeof(output_b.name) - 1); retry = create_retry_dispatch_task(ctx, &output_a, &task_id, &chunk_buffer); TEST_CHECK(retry != NULL); if (retry == NULL) { + test_output_destroy(&output_a); test_ctx_destroy(ctx); return; } @@ -304,6 +476,7 @@ static void test_retry_flush_failure_preserves_pending_retry(void) memfs = ((struct cio_chunk *) chunk->chunk)->backend; memfs->buf_data = chunk_buffer; flb_task_destroy(task, FLB_TRUE); + test_output_destroy(&output_a); test_ctx_destroy(ctx); return; } @@ -324,6 +497,7 @@ static void test_retry_flush_failure_preserves_pending_retry(void) TEST_CHECK(mk_list_size(&task->retries) == 1); TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1); TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1); + check_retry_failure_metrics(ctx, &output_a); if (ctx->config->task_map[task_id].task == task) { chunk = task->ic; @@ -340,6 +514,7 @@ static void test_retry_flush_failure_preserves_pending_retry(void) TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0); TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0); + test_output_destroy(&output_a); test_ctx_destroy(ctx); } From 044ba3599257df9816e6d5bf31acbab4dad8ba02 Mon Sep 17 00:00:00 2001 From: ku524 Date: Tue, 11 Aug 2026 15:06:46 +0900 Subject: [PATCH 5/6] engine_dispatch: re-schedule retry when chunk content cannot be read flb_input_chunk_flush() returning NULL is usually transient, so dropping the retry discards records a later attempt could still deliver. Spend a delivery attempt on it instead and only give up once the configured retry limit is reached, where the existing accounting already applies. Behaviour is unchanged with the default retry_limit of 1: the first read failure still drops immediately. Signed-off-by: ku524 --- src/flb_engine_dispatch.c | 39 +++++++++++++++++++++++++++++++++------ 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/src/flb_engine_dispatch.c b/src/flb_engine_dispatch.c index 968bc6cf6ad..dee466ea012 100644 --- a/src/flb_engine_dispatch.c +++ b/src/flb_engine_dispatch.c @@ -90,6 +90,7 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry, char *buf_data; size_t buf_size; struct flb_task *task; + struct flb_output_instance *ins; task = retry->parent; @@ -116,12 +117,38 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry, /* There is a match, get the buffer */ buf_data = (char *) flb_input_chunk_flush(task->ic, &buf_size); if (!buf_data) { - /* Could not retrieve chunk content */ - flb_error("[engine_dispatch] could not retrieve chunk content, removing retry"); - record_retry_failure_metrics(task, retry->o_ins, config); - flb_task_retry_destroy(retry); - flb_task_users_release(task); - return -1; + /* + * The chunk is up but its content could not be read. That is usually + * transient, so spend a delivery attempt on it instead of discarding + * records a later attempt could still deliver. + * + * Destroying the retry without releasing the task would leave the task + * with no users and no retries, a state nothing reaps. + */ + ins = retry->o_ins; + + if (retry->attempts >= ins->retry_limit && ins->retry_limit >= 0) { + flb_error("[engine_dispatch] could not retrieve chunk content, " + "task_id=%i reached retry-attempts limit %i/%i, dropping", + task->id, retry->attempts, ins->retry_limit); + record_retry_failure_metrics(task, ins, config); + flb_task_retry_destroy(retry); + flb_task_users_release(task); + return -1; + } + + retry->attempts++; + flb_warn("[engine_dispatch] could not retrieve chunk content, " + "re-scheduling task_id=%i attempts=%i", + task->id, retry->attempts); + + ret = flb_task_retry_reschedule(retry, config); + if (ret == -1) { + return -1; + } + + /* Just return because it has been re-scheduled */ + return 0; } /* Update the buffer reference */ From fc3a2fc98be30a36d4e8570e989fdbada7e244c2 Mon Sep 17 00:00:00 2001 From: ku524 Date: Tue, 11 Aug 2026 15:06:46 +0900 Subject: [PATCH 6/6] tests: internal: cover re-scheduling on chunk read failure Asserts that with retry budget left the task keeps its task-map slot, its chunk and a pending retry, and that no drop accounting is recorded. Signed-off-by: ku524 --- tests/internal/engine_dispatch.c | 74 ++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/tests/internal/engine_dispatch.c b/tests/internal/engine_dispatch.c index f164d0b9803..ecb340179f6 100644 --- a/tests/internal/engine_dispatch.c +++ b/tests/internal/engine_dispatch.c @@ -518,9 +518,83 @@ static void test_retry_flush_failure_preserves_pending_retry(void) test_ctx_destroy(ctx); } +/* + * With retry budget left the chunk must be kept and the attempt re-scheduled, + * otherwise a transient read failure would delete records a later attempt + * could still deliver. + */ +static void test_retry_flush_failure_reschedules_within_retry_limit(void) +{ + int ret; + int task_id; + double value; + char *chunk_buffer; + char *output_name; + struct test_ctx *ctx; + struct flb_task *task; + struct flb_task_retry *retry; + struct flb_output_instance output; + + ctx = test_ctx_create(); + TEST_CHECK(ctx != NULL); + if (ctx == NULL) { + return; + } + + ret = test_output_init(&output, "output_a"); + TEST_CHECK(ret == 0); + if (ret != 0) { + test_ctx_destroy(ctx); + return; + } + + /* the retry starts with attempts=1, so this leaves budget available */ + output.retry_limit = 5; + + retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer); + TEST_CHECK(retry != NULL); + if (retry == NULL) { + test_output_destroy(&output); + test_ctx_destroy(ctx); + return; + } + task = retry->parent; + + ret = flb_engine_dispatch_retry(retry, ctx->config); + TEST_CHECK(ret == 0); + + /* the task keeps its slot, its chunk and a pending retry */ + TEST_CHECK(ctx->config->task_map[task_id].task == task); + TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1); + TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1); + TEST_CHECK(mk_list_size(&task->retries) == 1); + TEST_CHECK(retry->attempts == 2); + + /* + * Nothing was dropped, so no drop accounting must be recorded. An untouched + * counter has no series yet, so cmt_counter_get_val() reports a failure. + */ + output_name = (char *) flb_output_name(&output); + ret = cmt_counter_get_val(output.cmt_retries_failed, + 1, (char *[]) {output_name}, &value); + TEST_CHECK(ret != 0 || value == 0); + + ret = cmt_counter_get_val(output.cmt_dropped_records, + 1, (char *[]) {output_name}, &value); + TEST_CHECK(ret != 0 || value == 0); + + flb_task_destroy(task, FLB_TRUE); + free(chunk_buffer); + + test_output_destroy(&output); + test_ctx_destroy(ctx); +} + TEST_LIST = { { "retry_flush_failure_releases_last_task_owner", test_retry_flush_failure_releases_last_task_owner }, + { "retry_flush_failure_reschedules_within_retry_limit", + test_retry_flush_failure_reschedules_within_retry_limit }, { "retry_flush_failure_preserves_active_task_owner", test_retry_flush_failure_preserves_active_task_owner }, { "retry_flush_failure_preserves_pending_retry",