From 875157d6d4a087b7d1a6ef88335bae6138156968 Mon Sep 17 00:00:00 2001 From: amirgo1 Date: Wed, 7 Oct 2026 21:18:31 +0000 Subject: [PATCH 1/2] out_gcs: add upload_on_shutdown option to upload buffered files on exit out_gcs leaves buffered files in store_dir for the next start. When the next start never comes, for example when a Kubernetes node is removed, up to upload_timeout of buffered data per tag is lost. Add upload_on_shutdown (default false) and upload_on_shutdown_timeout (default 20s). On exit, every buffered file is sealed and uploaded from cb_exit, which runs on the pipeline thread after the output workers have been joined; the upstream stays in sync mode, so blocking requests are safe there. Per-request timeouts are clamped to the remaining budget, and files not uploaded in time stay in store_dir as before. Nothing is uploaded on hot reload, where the new instance recovers store_dir. Also create the upload timer in cb_worker_init. A backlog found at startup used to create the timer from cb_init on the engine scheduler, so every later upload ran on the pipeline thread, and the backlog was uploaded synchronously during init. With workers, the backlog is now uploaded from a one-shot timer on the worker shortly after start. Signed-off-by: amirgo1 --- plugins/out_gcs/gcs.c | 210 ++++++++++++++++++++++++++++++++++++------ plugins/out_gcs/gcs.h | 2 + 2 files changed, 184 insertions(+), 28 deletions(-) diff --git a/plugins/out_gcs/gcs.c b/plugins/out_gcs/gcs.c index 9845efa1cc9..c001c9f8b60 100644 --- a/plugins/out_gcs/gcs.c +++ b/plugins/out_gcs/gcs.c @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -1310,6 +1311,9 @@ static int gcs_upload_object(struct flb_gcs *ctx, if (gcs_under_test_mode() == FLB_TRUE) { mock_gcs_call_increment_counter("UploadObject"); + if (getenv("TEST_GCS_UPLOAD_DELAY_MS") != NULL) { + flb_time_msleep(atoi(getenv("TEST_GCS_UPLOAD_DELAY_MS"))); + } gcs_setenv("TEST_GCS_LAST_URI", uri); gcs_setenv("TEST_GCS_LAST_CONTENT_TYPE", content_type_header.val); if (body_size >= 2 && @@ -1573,14 +1577,39 @@ static int upload_data(struct flb_gcs *ctx, return ret; } +/* On failure the file stays queued and is retried after a linear backoff. */ +static int upload_queue_entry(struct flb_gcs *ctx, struct upload_queue *entry, + time_t now) +{ + char *buffer; + size_t buffer_size; + int ret; + + gcs_store_file_lock(entry->upload_file); + + ret = construct_request_buffer(ctx, entry, &buffer, &buffer_size); + if (ret == 0) { + ret = upload_data(ctx, entry, buffer, buffer_size); + flb_free(buffer); + } + + if (ret != 0) { + gcs_store_file_unlock(entry->upload_file); + entry->retry_counter++; + entry->upload_time = now + (2 * entry->retry_counter); + return -1; + } + + gcs_store_file_delete(ctx, entry->upload_file); + remove_from_queue(entry); + return 0; +} + static int process_upload_queue(struct flb_gcs *ctx) { struct mk_list *head; struct mk_list *tmp; struct upload_queue *entry; - char *buffer; - size_t buffer_size; - int ret; time_t now; /* @@ -1601,28 +1630,8 @@ static int process_upload_queue(struct flb_gcs *ctx) } continue; } - gcs_store_file_lock(entry->upload_file); - ret = construct_request_buffer(ctx, entry, &buffer, &buffer_size); - if (ret == -1) { - gcs_store_file_unlock(entry->upload_file); - entry->retry_counter++; - entry->upload_time = now + (2 * entry->retry_counter); - break; - } - - ret = upload_data(ctx, entry, buffer, buffer_size); - - if (ret == 0) { - gcs_store_file_delete(ctx, entry->upload_file); - flb_free(buffer); - remove_from_queue(entry); - } - else { - flb_free(buffer); - gcs_store_file_unlock(entry->upload_file); - entry->retry_counter++; - entry->upload_time = now + (2 * entry->retry_counter); + if (upload_queue_entry(ctx, entry, now) == -1) { /* * Stop after a failed upload so the next pass can retry it * without blocking this thread on more network requests. @@ -1747,6 +1756,35 @@ static int flush_init(struct flb_gcs *ctx) return 0; } +/* + * Create the upload timer on the worker's scheduler so periodic uploads, + * including a backlog recovered at startup, never run on the engine thread. + */ +static int cb_gcs_worker_init(void *data, struct flb_config *config) +{ + int ret = 0; + struct flb_sched *sched; + struct flb_gcs *ctx = data; + + (void) config; + + pthread_mutex_lock(&ctx->upload_lock); + if (ctx->timer_created == FLB_FALSE) { + ret = flush_init(ctx); + if (ret == 0 && mk_list_size(&ctx->upload_queue) > 0) { + sched = flb_sched_ctx_get(); + ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_ONESHOT, + 1000, cb_gcs_upload, ctx, NULL); + } + } + pthread_mutex_unlock(&ctx->upload_lock); + + if (ret == -1) { + flb_plg_error(ctx->ins, "could not create the upload timer"); + } + return ret; +} + static int gcs_init_identity_federation(struct flb_gcs *ctx, struct flb_config *config) { int io_flags = FLB_IO_TLS; @@ -1884,6 +1922,11 @@ static int cb_gcs_init(struct flb_output_instance *ins, struct flb_config *confi goto error; } + if (ctx->upload_on_shutdown_timeout < 0) { + flb_plg_error(ins, "'upload_on_shutdown_timeout' must not be negative"); + goto error; + } + if (ctx->total_file_size != 0 && ctx->total_file_size < FLB_GCS_MIN_TOTAL_FILE_SIZE) { flb_plg_error(ins, "'total_file_size' must be at least 1M " @@ -2090,11 +2133,13 @@ static int cb_gcs_init(struct flb_output_instance *ins, struct flb_config *confi if (gcs_store_has_data(ctx) == FLB_TRUE) { enqueue_backlog_files(ctx); - if (mk_list_size(&ctx->upload_queue) > 0 && flush_init(ctx) == -1) { - goto error; + /* with workers, cb_gcs_worker_init uploads the backlog */ + if (ins->tp_workers == 0) { + if (mk_list_size(&ctx->upload_queue) > 0 && flush_init(ctx) == -1) { + goto error; + } + process_upload_queue(ctx); } - - process_upload_queue(ctx); } flb_output_set_context(ins, ctx); @@ -2282,8 +2327,104 @@ static int gcs_ctx_destroy(void *data, struct flb_config *config) return 0; } +static void clamp_net_timeout(int *timeout, int remaining) +{ + if (*timeout <= 0 || *timeout > remaining) { + *timeout = remaining; + } +} + +/* + * Runs from cb_exit on the pipeline thread after the output workers are + * joined; the upstreams are in sync mode, so blocking requests are safe. + * Files not uploaded within upload_on_shutdown_timeout stay in store_dir + * for the next start. + */ +static void upload_all_on_shutdown(struct flb_gcs *ctx) +{ + int total; + int uploaded = 0; + int failed = 0; + int left; + time_t now; + time_t deadline = 0; + struct flb_net_setup net; + struct mk_list *head; + struct mk_list *tmp; + struct flb_fstore_file *fsf; + struct upload_queue *entry; + + if (!ctx->fs_stream || !ctx->u || gcs_store_has_data(ctx) == FLB_FALSE) { + return; + } + + pthread_mutex_lock(&ctx->upload_lock); + + /* every flush queues its file; this also catches any it failed to queue */ + mk_list_foreach(head, &ctx->fs_stream->files) { + fsf = mk_list_entry(head, struct flb_fstore_file, _head); + if (!fsf->data && attach_recovered_chunk(ctx, fsf) == -1) { + continue; + } + gcs_store_file_seal(fsf->data); + add_to_queue(ctx, fsf->data, (const char *) fsf->meta_buf, fsf->meta_size); + } + + total = mk_list_size(&ctx->upload_queue); + if (ctx->upload_on_shutdown_timeout > 0) { + deadline = time(NULL) + ctx->upload_on_shutdown_timeout; + } + flb_plg_info(ctx->ins, "uploading %d buffered file(s) on shutdown", total); + + net = ctx->u->base.net; + mk_list_foreach_safe(head, tmp, &ctx->upload_queue) { + entry = mk_list_entry(head, struct upload_queue, _head); + now = time(NULL); + if (deadline > 0) { + if (now >= deadline) { + break; + } + clamp_net_timeout(&ctx->u->base.net.connect_timeout, deadline - now); + clamp_net_timeout(&ctx->u->base.net.io_timeout, deadline - now); + } + + if (upload_queue_entry(ctx, entry, now) == 0) { + uploaded++; + } + else { + failed++; + if (ctx->preserve_data_ordering == FLB_TRUE) { + break; + } + } + ctx->u->base.net = net; + } + ctx->u->base.net = net; + + left = mk_list_size(&ctx->upload_queue); + if (left > 0) { + flb_plg_warn(ctx->ins, "%d of %d buffered file(s) were not uploaded on " + "shutdown (%d failed); they stay in %s for the next start", + left, total, failed, ctx->store_dir); + } + else { + flb_plg_info(ctx->ins, "uploaded all %d buffered file(s) on shutdown", + uploaded); + } + + pthread_mutex_unlock(&ctx->upload_lock); +} + static int cb_gcs_exit(void *data, struct flb_config *config) { + struct flb_gcs *ctx = data; + + /* on hot reload the new instance recovers store_dir right away */ + if (ctx && ctx->upload_on_shutdown == FLB_TRUE && + config->shutdown_by_hot_reloading == FLB_FALSE) { + upload_all_on_shutdown(ctx); + } + gcs_ctx_destroy(data, config); return 0; @@ -2333,6 +2474,18 @@ static struct flb_config_map config_map[] = { "Batches are not split, so objects can exceed this size. Minimum 1M, " "0 disables the size trigger. Ordered uploads still wait for older files." }, + { + FLB_CONFIG_MAP_BOOL, "upload_on_shutdown", "false", + 0, FLB_TRUE, offsetof(struct flb_gcs, upload_on_shutdown), + "Upload all locally buffered files when Fluent Bit stops, instead of " + "leaving them in store_dir for the next start." + }, + { + FLB_CONFIG_MAP_TIME, "upload_on_shutdown_timeout", "20s", + 0, FLB_TRUE, offsetof(struct flb_gcs, upload_on_shutdown_timeout), + "Maximum time spent uploading buffered files on shutdown (0 means no " + "limit). Files not uploaded in time stay in store_dir for the next start." + }, { FLB_CONFIG_MAP_BOOL, "send_content_md5", "false", 0, FLB_TRUE, offsetof(struct flb_gcs, send_content_md5), @@ -2437,6 +2590,7 @@ struct flb_output_plugin out_gcs_plugin = { .cb_init = cb_gcs_init, .cb_flush = cb_gcs_flush, .cb_exit = cb_gcs_exit, + .cb_worker_init = cb_gcs_worker_init, .event_type = FLB_OUTPUT_LOGS, .config_map = config_map, .flags = FLB_OUTPUT_NET | FLB_IO_TLS, diff --git a/plugins/out_gcs/gcs.h b/plugins/out_gcs/gcs.h index 609c9b385b8..a930ac55fc9 100644 --- a/plugins/out_gcs/gcs.h +++ b/plugins/out_gcs/gcs.h @@ -130,6 +130,8 @@ struct flb_gcs { int upload_queue_processing; int timer_created; int timer_ms; + int upload_on_shutdown; + int upload_on_shutdown_timeout; struct flb_gcs_oauth_credentials *oauth_credentials; From 85254f4fff3be90f0e291def6272e7fad6977f0e Mon Sep 17 00:00:00 2001 From: amirgo1 Date: Wed, 7 Oct 2026 21:18:31 +0000 Subject: [PATCH 2/2] tests: add out_gcs upload_on_shutdown tests Signed-off-by: amirgo1 --- tests/runtime/out_gcs.c | 175 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/tests/runtime/out_gcs.c b/tests/runtime/out_gcs.c index 2b13636bf15..3c3f2ef9ae8 100644 --- a/tests/runtime/out_gcs.c +++ b/tests/runtime/out_gcs.c @@ -1865,6 +1865,177 @@ void flb_test_gcs_total_file_size_zero_disables_size_trigger(void) flb_free(store_dir); } +static int gcs_upload_call_count(void) +{ + char *count; + + count = getenv("TEST_GCS_UploadObject_CALL_COUNT"); + return count ? atoi(count) : 0; +} + +/* + * Start an instance on store_dir, buffer one record per tag (nothing is due + * yet with upload_timeout 10m), then stop it. + */ +static void gcs_run_and_stop(const char *store_dir, int tags, const char *workers, + const char *upload_on_shutdown, const char *timeout) +{ + int i; + int ret; + int in_ffd[4]; + int out_ffd; + char tag[16]; + flb_ctx_t *ctx; + + ctx = flb_create(); + flb_service_set(ctx, "Grace", "1", NULL); + for (i = 0; i < tags; i++) { + snprintf(tag, sizeof(tag), "test.%d", i); + in_ffd[i] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd[i] >= 0); + flb_input_set(ctx, in_ffd[i], "tag", tag, NULL); + } + + out_ffd = flb_output(ctx, (char *) "gcs", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "google_service_credentials", SERVICE_CREDENTIALS, NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "10m", NULL); + flb_output_set(ctx, out_ffd, "workers", workers, NULL); + flb_output_set(ctx, out_ffd, "upload_on_shutdown", upload_on_shutdown, NULL); + if (timeout) { + flb_output_set(ctx, out_ffd, "upload_on_shutdown_timeout", timeout, NULL); + } + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + if (ret == 0) { + for (i = 0; i < tags; i++) { + flb_lib_push(ctx, in_ffd[i], (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + } + sleep(2); + flb_stop(ctx); + } + flb_destroy(ctx); +} + +static void gcs_shutdown_test_cleanup(char *store_dir) +{ + unsetenv("FLB_GCS_PLUGIN_UNDER_TEST"); + unsetenv("TEST_GCS_UPLOAD_ERROR"); + unsetenv("TEST_GCS_UPLOAD_DELAY_MS"); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + unsetenv("TEST_GCS_LAST_URI"); + unsetenv("TEST_GCS_LAST_BODY_GZIP"); + unsetenv("TEST_GCS_LAST_BODY_PARQUET"); + unsetenv("TEST_GCS_LAST_CONTENT_TYPE"); + flb_free(store_dir); +} + +void flb_test_gcs_shutdown_uploads_pending_when_enabled(void) +{ + int count; + char *store_dir; + + store_dir = create_test_store_directory("/flb-gcs-test-shutdown-upload-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + setenv("FLB_GCS_PLUGIN_UNDER_TEST", "true", 1); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + + gcs_run_and_stop(store_dir, 1, "1", "true", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 1, + "Expected the buffered file to be uploaded on shutdown, got %d", count); + + /* the next start finds nothing left to recover */ + gcs_run_and_stop(store_dir, 0, "1", "false", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 1, "Expected no backlog after the shutdown upload, got %d", count); + + gcs_shutdown_test_cleanup(store_dir); +} + +void flb_test_gcs_shutdown_upload_failure_keeps_file(void) +{ + int count; + char *store_dir; + + store_dir = create_test_store_directory("/flb-gcs-test-shutdown-fail-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + setenv("FLB_GCS_PLUGIN_UNDER_TEST", "true", 1); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + + setenv("TEST_GCS_UPLOAD_ERROR", "true", 1); + gcs_run_and_stop(store_dir, 1, "1", "true", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 1, "Expected one failed upload attempt on shutdown, got %d", count); + + /* the file stayed in store_dir and the next start uploads it */ + unsetenv("TEST_GCS_UPLOAD_ERROR"); + gcs_run_and_stop(store_dir, 0, "1", "false", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 2, "Expected the kept file to be recovered and uploaded, got %d", count); + + gcs_shutdown_test_cleanup(store_dir); +} + +void flb_test_gcs_shutdown_upload_respects_timeout(void) +{ + int first; + int count; + char *store_dir; + + store_dir = create_test_store_directory("/flb-gcs-test-shutdown-timeout-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + setenv("FLB_GCS_PLUGIN_UNDER_TEST", "true", 1); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + + /* four files at 1.5s each cannot all fit in a 2s budget */ + setenv("TEST_GCS_UPLOAD_DELAY_MS", "1500", 1); + gcs_run_and_stop(store_dir, 4, "1", "true", "2s"); + first = gcs_upload_call_count(); + TEST_CHECK_(first >= 1 && first < 4, + "Expected the budget to stop the shutdown uploads early, got %d", first); + + unsetenv("TEST_GCS_UPLOAD_DELAY_MS"); + gcs_run_and_stop(store_dir, 0, "1", "false", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 4, "Expected the remaining files to be recovered, got %d in total", count); + + gcs_shutdown_test_cleanup(store_dir); +} + +void flb_test_gcs_shutdown_upload_with_workers(void) +{ + int count; + char *store_dir; + + store_dir = create_test_store_directory("/flb-gcs-test-shutdown-workers-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + setenv("FLB_GCS_PLUGIN_UNDER_TEST", "true", 1); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + + gcs_run_and_stop(store_dir, 2, "2", "true", NULL); + count = gcs_upload_call_count(); + TEST_CHECK_(count == 2, "Expected one shutdown upload per tag, got %d", count); + + gcs_shutdown_test_cleanup(store_dir); +} + TEST_LIST = { {"jwt_signing", flb_test_gcs_jwt_signing}, {"net_settings_applied_to_upstream", flb_test_gcs_net_settings_applied_to_upstream}, @@ -1906,6 +2077,10 @@ TEST_LIST = { {"rejects_invalid_metadata_response", flb_test_gcs_rejects_invalid_metadata_response}, {"upload_error", flb_test_gcs_upload_error}, {"shutdown_preserves_pending_upload", flb_test_gcs_shutdown_preserves_pending_upload}, + {"shutdown_uploads_pending_when_enabled", flb_test_gcs_shutdown_uploads_pending_when_enabled}, + {"shutdown_upload_failure_keeps_file", flb_test_gcs_shutdown_upload_failure_keeps_file}, + {"shutdown_upload_respects_timeout", flb_test_gcs_shutdown_upload_respects_timeout}, + {"shutdown_upload_with_workers", flb_test_gcs_shutdown_upload_with_workers}, {"unify_tag_buffers_multiple_tags", flb_test_gcs_unify_tag_buffers_multiple_tags}, {"unify_tag_disabled_by_default", flb_test_gcs_unify_tag_disabled_by_default}, {NULL, NULL}