diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index 9bc8bee23df..fad4b931568 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -2212,7 +2212,7 @@ static int upload_queue_valid(struct upload_queue *upload_contents, time_t now, "Exiting"); return -1; } - if (upload_contents->upload_file->size <= 0) { + if (s3_store_file_size_get(upload_contents->upload_file) == 0) { flb_plg_debug(ctx->ins, "Encountered empty chunk file in upload_queue. " "Deleting empty chunk file"); remove_from_queue(upload_contents); @@ -2278,7 +2278,6 @@ static int buffer_chunk(void *out_context, struct s3_file *upload_file, static void s3_chunk_retry_exhausted_cleanup(struct flb_s3 *ctx, struct s3_file *chunk_file) { - size_t reclaim_size; int ret; if (chunk_file == NULL) { @@ -2286,17 +2285,13 @@ static void s3_chunk_retry_exhausted_cleanup(struct flb_s3 *ctx, } if (ctx->retry_exhausted_action == S3_RETRY_EXHAUSTED_QUARANTINE) { - reclaim_size = chunk_file->size; - - if (ctx->quarantine_dir_limit_size > 0 && - ctx->quarantine_buffer_size + reclaim_size > ctx->quarantine_dir_limit_size) { + ret = s3_store_file_quarantine(ctx, chunk_file); + if (ret == S3_STORE_QUARANTINE_FULL) { flb_plg_warn(ctx->ins, "quarantine limit reached, deleting retry-exhausted chunk"); s3_store_file_delete(ctx, chunk_file); return; } - - ret = s3_store_file_quarantine(ctx, chunk_file); if (ret < 0) { flb_plg_error(ctx->ins, "could not quarantine, deleting retry-exhausted chunk"); @@ -2304,14 +2299,6 @@ static void s3_chunk_retry_exhausted_cleanup(struct flb_s3 *ctx, return; } - if (ctx->current_buffer_size >= reclaim_size) { - ctx->current_buffer_size -= reclaim_size; - } - else { - ctx->current_buffer_size = 0; - } - ctx->quarantine_buffer_size += reclaim_size; - flb_plg_warn(ctx->ins, "retry-exhausted chunk moved to quarantine"); return; @@ -4349,7 +4336,8 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, } /* If total_file_size has been reached, upload file */ - if ((upload_file && upload_file->size + chunk_size > ctx->upload_chunk_size) || + if ((upload_file && + s3_store_file_size_get(upload_file) + chunk_size > ctx->upload_chunk_size) || (m_upload_file && m_upload_file->bytes + chunk_size > ctx->file_size)) { total_file_size_check = FLB_TRUE; } diff --git a/plugins/out_s3/s3.h b/plugins/out_s3/s3.h index 08fc846d0ea..744cbbf7fce 100644 --- a/plugins/out_s3/s3.h +++ b/plugins/out_s3/s3.h @@ -153,8 +153,8 @@ struct flb_s3 { struct flb_tls *authorization_endpoint_tls_context; /* track the total amount of buffered data */ - size_t current_buffer_size; - size_t quarantine_buffer_size; + uint64_t current_buffer_size; + uint64_t quarantine_buffer_size; struct flb_aws_provider *provider; struct flb_aws_provider *base_provider; diff --git a/plugins/out_s3/s3_store.c b/plugins/out_s3/s3_store.c index fbaf1753c1d..fe689d83b0b 100644 --- a/plugins/out_s3/s3_store.c +++ b/plugins/out_s3/s3_store.c @@ -20,10 +20,135 @@ #include #include #include +#include + +#include #include "s3.h" #include "s3_store.h" +static int counter_add(uint64_t *counter, uint64_t increment, + uint64_t *new_value) +{ + uint64_t current_value; + uint64_t updated_value; + + while (FLB_TRUE) { + current_value = cfl_atomic_load(counter); + if (increment > UINT64_MAX - current_value) { + return -1; + } + + updated_value = current_value + increment; + if (cfl_atomic_compare_exchange(counter, current_value, updated_value)) { + if (new_value != NULL) { + *new_value = updated_value; + } + return 0; + } + } +} + +static int counter_subtract(uint64_t *counter, uint64_t decrement, + uint64_t *previous_value) +{ + int underflow; + uint64_t current_value; + uint64_t updated_value; + + while (FLB_TRUE) { + current_value = cfl_atomic_load(counter); + underflow = current_value < decrement; + if (underflow == FLB_TRUE) { + updated_value = 0; + } + else { + updated_value = current_value - decrement; + } + + if (cfl_atomic_compare_exchange(counter, current_value, updated_value)) { + if (previous_value != NULL) { + *previous_value = current_value; + } + return underflow == FLB_TRUE ? -1 : 0; + } + } +} + +static int buffer_size_reserve(struct flb_s3 *ctx, size_t bytes, + uint64_t *current_value, + uint64_t *new_value) +{ + uint64_t buffer_size; + uint64_t updated_size; + uint64_t increment; + + increment = (uint64_t) bytes; + + while (FLB_TRUE) { + buffer_size = cfl_atomic_load(&ctx->current_buffer_size); + if (current_value != NULL) { + *current_value = buffer_size; + } + + if (increment > UINT64_MAX - buffer_size) { + return -1; + } + + updated_size = buffer_size + increment; + if (ctx->store_dir_limit_size > 0 && + updated_size >= (uint64_t) ctx->store_dir_limit_size) { + return -1; + } + + if (cfl_atomic_compare_exchange(&ctx->current_buffer_size, + buffer_size, updated_size)) { + if (new_value != NULL) { + *new_value = updated_size; + } + return 0; + } + } +} + +static int quarantine_size_reserve(struct flb_s3 *ctx, uint64_t bytes) +{ + uint64_t buffer_size; + uint64_t updated_size; + + while (FLB_TRUE) { + buffer_size = cfl_atomic_load(&ctx->quarantine_buffer_size); + if (bytes > UINT64_MAX - buffer_size) { + return -1; + } + + updated_size = buffer_size + bytes; + if (ctx->quarantine_dir_limit_size > 0 && + updated_size > (uint64_t) ctx->quarantine_dir_limit_size) { + return -1; + } + + if (cfl_atomic_compare_exchange(&ctx->quarantine_buffer_size, + buffer_size, updated_size)) { + return 0; + } + } +} + +static void buffer_size_release(struct flb_s3 *ctx, uint64_t bytes) +{ + int ret; + uint64_t previous_value; + + ret = counter_subtract(&ctx->current_buffer_size, bytes, &previous_value); + if (ret < 0) { + flb_plg_warn(ctx->ins, + "buffer accounting mismatch: current_buffer_size=%" PRIu64 + ", reclaim_size=%" PRIu64 "; clamping to zero", + previous_value, bytes); + } +} + static int s3_store_under_travis_ci() { @@ -131,11 +256,16 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, int ret; flb_sds_t name; struct flb_fstore_file *fsf; - size_t space_remaining; + uint64_t current_buffer_size; + uint64_t new_buffer_size; - if (ctx->store_dir_limit_size > 0 && ctx->current_buffer_size + bytes >= ctx->store_dir_limit_size) { - flb_plg_error(ctx->ins, "Buffer is full: current_buffer_size=%zu, new_data=%zu, store_dir_limit_size=%zu bytes", - ctx->current_buffer_size, bytes, ctx->store_dir_limit_size); + ret = buffer_size_reserve(ctx, bytes, ¤t_buffer_size, + &new_buffer_size); + if (ret < 0) { + flb_plg_error(ctx->ins, + "Buffer is full: current_buffer_size=%" PRIu64 + ", new_data=%zu, store_dir_limit_size=%zu bytes", + current_buffer_size, bytes, ctx->store_dir_limit_size); return -1; } @@ -144,6 +274,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, name = gen_store_filename(tag); if (!name) { flb_plg_error(ctx->ins, "could not generate chunk file name"); + buffer_size_release(ctx, bytes); return -1; } @@ -153,6 +284,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, flb_plg_error(ctx->ins, "could not create the file '%s' in the store", name); flb_sds_destroy(name); + buffer_size_release(ctx, bytes); return -1; } flb_sds_destroy(name); @@ -163,6 +295,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, flb_plg_error(ctx->ins, "error writing tag metadata"); flb_plg_warn(ctx->ins, "Deleting buffer file because metadata could not be written"); flb_fstore_file_delete(ctx->fs, fsf); + buffer_size_release(ctx, bytes); return -1; } @@ -173,6 +306,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, flb_plg_error(ctx->ins, "cannot allocate s3 file context"); flb_plg_warn(ctx->ins, "Deleting buffer file because S3 context creation failed"); flb_fstore_file_delete(ctx->fs, fsf); + buffer_size_release(ctx, bytes); return -1; } s3_file->fsf = fsf; @@ -190,19 +324,22 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, ret = flb_fstore_file_append(fsf, data, bytes); if (ret != 0) { flb_plg_error(ctx->ins, "error writing data to local s3 file"); + buffer_size_release(ctx, bytes); return -1; } - s3_file->size += bytes; - ctx->current_buffer_size += bytes; + ret = counter_add(&s3_file->size, (uint64_t) bytes, NULL); + if (ret < 0) { + flb_plg_error(ctx->ins, "local s3 file size accounting overflow"); + } /* if buffer is 95% full, warn user */ - if (ctx->store_dir_limit_size > 0) { - space_remaining = ctx->store_dir_limit_size - ctx->current_buffer_size; - if ((space_remaining * 20) < ctx->store_dir_limit_size) { - flb_plg_warn(ctx->ins, "Buffer is almost full: current_buffer_size=%zu, store_dir_limit_size=%zu bytes", - ctx->current_buffer_size, ctx->store_dir_limit_size); - return -1; - } + if (ctx->store_dir_limit_size > 0 && + new_buffer_size > (uint64_t) ctx->store_dir_limit_size - + (ctx->store_dir_limit_size / 20)) { + flb_plg_warn(ctx->ins, + "Buffer is almost full: current_buffer_size=%" PRIu64 + ", store_dir_limit_size=%zu bytes", + new_buffer_size, ctx->store_dir_limit_size); } return 0; @@ -220,11 +357,6 @@ static int set_files_context(struct flb_s3 *ctx) mk_list_foreach(head, &ctx->fs->streams) { fs_stream = mk_list_entry(head, struct flb_fstore_stream, _head); - /* skip current stream since it's new */ - if (fs_stream == ctx->stream_active) { - continue; - } - /* skip multi-upload */ if (fs_stream == ctx->stream_upload) { continue; @@ -247,9 +379,21 @@ static int set_files_context(struct flb_s3 *ctx) s3_file->first_log_time = time(NULL); s3_file->create_time = time(NULL); - file_size = cio_chunk_get_real_size(fsf->chunk); + file_size = cio_chunk_get_content_size(fsf->chunk); if (file_size > 0) { - s3_file->size = (size_t) file_size; + cfl_atomic_store(&s3_file->size, (uint64_t) file_size); + + if (fs_stream == ctx->stream_quarantine) { + if (counter_add(&ctx->quarantine_buffer_size, + (uint64_t) file_size, NULL) < 0) { + cfl_atomic_store(&ctx->quarantine_buffer_size, + UINT64_MAX); + } + } + else if (counter_add(&ctx->current_buffer_size, + (uint64_t) file_size, NULL) < 0) { + cfl_atomic_store(&ctx->current_buffer_size, UINT64_MAX); + } } /* Use fstore opaque 'data' reference to keep our context */ @@ -444,9 +588,16 @@ int s3_store_file_quarantine(struct flb_s3 *ctx, struct s3_file *s3_file) return -1; } + ret = quarantine_size_reserve(ctx, (uint64_t) size); + if (ret < 0) { + flb_free(buf); + return S3_STORE_QUARANTINE_FULL; + } + qfsf = flb_fstore_file_create(ctx->fs, ctx->stream_quarantine, fsf->name, size); if (qfsf == NULL) { flb_free(buf); + counter_subtract(&ctx->quarantine_buffer_size, (uint64_t) size, NULL); return -1; } @@ -455,6 +606,7 @@ int s3_store_file_quarantine(struct flb_s3 *ctx, struct s3_file *s3_file) if (ret < 0) { flb_free(buf); flb_fstore_file_delete(ctx->fs, qfsf); + counter_subtract(&ctx->quarantine_buffer_size, (uint64_t) size, NULL); return -1; } } @@ -463,20 +615,29 @@ int s3_store_file_quarantine(struct flb_s3 *ctx, struct s3_file *s3_file) flb_free(buf); if (ret < 0) { flb_fstore_file_delete(ctx->fs, qfsf); + counter_subtract(&ctx->quarantine_buffer_size, (uint64_t) size, NULL); return -1; } flb_fstore_file_delete(ctx->fs, fsf); + buffer_size_release(ctx, s3_store_file_size_get(s3_file)); flb_free(s3_file); return 0; } +uint64_t s3_store_file_size_get(struct s3_file *s3_file) +{ + return cfl_atomic_load(&s3_file->size); +} + int s3_store_file_delete(struct flb_s3 *ctx, struct s3_file *s3_file) { struct flb_fstore_file *fsf; + uint64_t file_size; fsf = s3_file->fsf; - ctx->current_buffer_size -= s3_file->size; + file_size = s3_store_file_size_get(s3_file); + buffer_size_release(ctx, file_size); /* permanent deletion */ flb_fstore_file_delete(ctx->fs, fsf); diff --git a/plugins/out_s3/s3_store.h b/plugins/out_s3/s3_store.h index 3c3b5fb27b1..7d787c0b337 100644 --- a/plugins/out_s3/s3_store.h +++ b/plugins/out_s3/s3_store.h @@ -26,13 +26,15 @@ struct s3_file { int locked; /* locked chunk is busy, cannot write to it */ int failures; /* delivery failures */ - size_t size; /* file size */ + uint64_t size; /* file size */ time_t create_time; /* creation time */ time_t first_log_time; /* first log time */ flb_sds_t file_path; /* file path */ struct flb_fstore_file *fsf; /* reference to parent flb_fstore_file */ }; +#define S3_STORE_QUARANTINE_FULL -2 + int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, const char *tag, int tag_len, char *data, size_t bytes, @@ -46,6 +48,7 @@ int s3_store_has_uploads(struct flb_s3 *ctx); int s3_store_file_inactive(struct flb_s3 *ctx, struct s3_file *s3_file); int s3_store_file_quarantine(struct flb_s3 *ctx, struct s3_file *s3_file); +uint64_t s3_store_file_size_get(struct s3_file *s3_file); struct s3_file *s3_store_file_get(struct flb_s3 *ctx, const char *tag, int tag_len); int s3_store_file_delete(struct flb_s3 *ctx, struct s3_file *s3_file); diff --git a/tests/runtime/out_s3.c b/tests/runtime/out_s3.c index 0b7f221e53d..3aa8c114e9c 100644 --- a/tests/runtime/out_s3.c +++ b/tests/runtime/out_s3.c @@ -1,9 +1,13 @@ /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ #include #include +#include +#include "../../plugins/out_s3/s3.h" +#include "../../plugins/out_s3/s3_store.h" #include "flb_tests_runtime.h" #include "../include/flb_tests_tmpdir.h" #include +#include #ifdef FLB_SYSTEM_WINDOWS #include @@ -28,6 +32,19 @@ Uuag1LuByRx9e6j5Onimru9pO4ZVKnJ2Qz7/C1NPcfTWAtRPfTaOFg==\ " +static struct flb_s3 *get_s3_context(flb_ctx_t *ctx) +{ + struct flb_output_instance *ins; + + if (ctx == NULL || mk_list_is_empty(&ctx->config->outputs) == 0) { + return NULL; + } + + ins = mk_list_entry_first(&ctx->config->outputs, + struct flb_output_instance, _head); + return ins->context; +} + static int count_files_recursive(const char *path) { #ifdef FLB_SYSTEM_WINDOWS @@ -1072,6 +1089,213 @@ void flb_test_s3_default_retry_exhausted_action_quarantine(void) flb_free(store_dir); } +void flb_test_s3_near_full_buffer_append_succeeds(void) +{ + int ret; + int in_ffd; + int out_ffd; + char payload[96]; + char excess_payload[5]; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + struct s3_file *s3_file; + + store_dir = create_test_store_directory("/flb-s3-test-near-full-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + memset(payload, 'a', sizeof(payload)); + memset(excess_payload, 'b', sizeof(excess_payload)); + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + + ctx = flb_create(); + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "s3", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "use_put_object", "true", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "5M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "1h", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "store_dir_limit_size", "100", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + ret = s3_store_buffer_put(s3_ctx, NULL, "test", 4, payload, + sizeof(payload), time(NULL)); + TEST_CHECK_(ret == 0, + "Expected a committed near-full append to return success, got %d", + ret); + TEST_CHECK(cfl_atomic_load(&s3_ctx->current_buffer_size) == sizeof(payload)); + + s3_file = s3_store_file_get(s3_ctx, "test", 4); + TEST_CHECK(s3_file != NULL); + TEST_CHECK(s3_store_file_size_get(s3_file) == sizeof(payload)); + + ret = s3_store_buffer_put(s3_ctx, s3_file, "test", 4, excess_payload, + sizeof(excess_payload), time(NULL)); + TEST_CHECK_(ret == -1, + "Expected an over-limit append to fail, got %d", + ret); + TEST_CHECK(cfl_atomic_load(&s3_ctx->current_buffer_size) == sizeof(payload)); + TEST_CHECK(s3_store_file_size_get(s3_file) == sizeof(payload)); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_PutObject_CALL_COUNT"); + flb_free(store_dir); +} + +void flb_test_s3_startup_buffer_size_accounting(void) +{ + int ret; + int in_ffd; + int out_ffd; + int call_count; + int file_count; + uint64_t live_buffer_size; + uint64_t restored_buffer_size; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + + store_dir = create_test_store_directory("/flb-s3-test-startup-size-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + + ctx = flb_create(); + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "s3", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "use_put_object", "true", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "5M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", S3_TEST_UPLOAD_TIMEOUT, NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "store_dir_limit_size", "1M", NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "10", NULL); + flb_output_set(ctx, out_ffd, "retry_exhausted_action", "delete", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_s3_call_count("PutObject", 1); + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + live_buffer_size = cfl_atomic_load(&s3_ctx->current_buffer_size); + TEST_CHECK_(live_buffer_size > 0, + "Expected live buffer accounting to contain payload bytes"); + + flb_stop(ctx); + flb_destroy(ctx); + + file_count = count_files_recursive(store_dir); + TEST_CHECK_(file_count > 0, + "Expected a buffered file to survive the first run, got %d", + file_count); + + unsetenv("TEST_PutObject_CALL_COUNT"); + + ctx = flb_create(); + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "s3", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "use_put_object", "true", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "5M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", S3_TEST_UPLOAD_TIMEOUT, NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "store_dir_limit_size", "1M", NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "10", NULL); + flb_output_set(ctx, out_ffd, "retry_exhausted_action", "delete", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + restored_buffer_size = cfl_atomic_load(&s3_ctx->current_buffer_size); + TEST_CHECK_(restored_buffer_size == live_buffer_size, + "Expected restored payload bytes=%" PRIu64 ", got %" PRIu64, + live_buffer_size, restored_buffer_size); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("TEST_PUT_OBJECT_ERROR"); + unsetenv("TEST_PutObject_CALL_COUNT"); + + ctx = flb_create(); + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "s3", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "use_put_object", "true", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "5M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", S3_TEST_UPLOAD_TIMEOUT, NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "store_dir_limit_size", "1M", NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "10", NULL); + flb_output_set(ctx, out_ffd, "retry_exhausted_action", "delete", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + wait_for_s3_call_count("PutObject", 1); + + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_s3_call_count("PutObject", 2); + + call_count = get_s3_call_count("PutObject"); + TEST_CHECK_(call_count == 2, + "Expected startup resend and new upload, got %d PutObject calls", + call_count); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_PUT_OBJECT_ERROR"); + unsetenv("TEST_PutObject_CALL_COUNT"); + flb_free(store_dir); +} + /* Test list */ TEST_LIST = { {"multipart_success", flb_test_s3_multipart_success }, @@ -1080,6 +1304,8 @@ TEST_LIST = { {"putobject_retry_limit_semantics", flb_test_s3_putobject_retry_limit_semantics }, {"default_retry_limit", flb_test_s3_default_retry_limit }, {"default_retry_exhausted_action_quarantine", flb_test_s3_default_retry_exhausted_action_quarantine }, + {"near_full_buffer_append_succeeds", flb_test_s3_near_full_buffer_append_succeeds }, + {"startup_buffer_size_accounting", flb_test_s3_startup_buffer_size_accounting }, {"create_upload_error", flb_test_s3_create_upload_error }, {"upload_part_error", flb_test_s3_upload_part_error }, {"complete_upload_error", flb_test_s3_complete_upload_error },