Skip to content

out_s3: Address underflow of buffer size on startup #12266

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

Open
cosmo0920 wants to merge 4 commits into
base: master
Choose a base branch
from
+419 −41
Open

out_s3: Address underflow of buffer size on startup #12266

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a7d3bbd
out_s3: Prevent underflow on accounting
cosmo0920 Aug 12, 2026
bc8eef0
tests: runtime: Add a test case for accounting of buffer size
cosmo0920 Aug 12, 2026
b2b5fa0
out_s3: Address continuous retrying issue on overflow
cosmo0920 Aug 12, 2026
063e7cd
tests: runtime: Adjust test cases for accounting of buffer sizes
cosmo0920 Aug 12, 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
22 changes: 5 additions & 17 deletions plugins/out_s3/s3.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -2278,40 +2278,27 @@ 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) {
return;
}

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");
s3_store_file_delete(ctx, chunk_file);
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;
Expand Down Expand Up @@ -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;
}
Expand Down
4 changes: 2 additions & 2 deletions plugins/out_s3/s3.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
203 changes: 182 additions & 21 deletions plugins/out_s3/s3_store.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,135 @@
#include <fluent-bit/flb_output_plugin.h>
#include <fluent-bit/flb_fstore.h>
#include <fluent-bit/flb_time.h>
#include <cfl/cfl_atomic.h>

#include <inttypes.h>

#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()
{

Expand Down Expand Up @@ -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, &current_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;
}

Expand All @@ -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;
}

Expand All @@ -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);
Expand All @@ -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;
}

Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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 */
Expand Down Expand Up @@ -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;
}

Expand All @@ -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;
}
}
Expand All @@ -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);
Expand Down
Loading
Loading