diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index 3a0f1273b91..fbcaf8d2980 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -35,6 +35,7 @@ #include #include #include +#include #include #include @@ -75,7 +76,7 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t file_first_log_time, char *body, size_t body_size); -static int put_all_chunks(struct flb_s3 *ctx); +static int put_all_chunks(struct flb_s3 *ctx, int cleanup_empty_streams); static void cb_s3_upload(struct flb_config *ctx, void *data); @@ -90,6 +91,13 @@ static void remove_from_queue(struct upload_queue *entry); static void s3_chunk_retry_exhausted_cleanup(struct flb_s3 *ctx, struct s3_file *chunk_file); static int s3_get_retry_exhausted_action(const char *value); +static void s3_upload_queue(struct flb_config *config, void *out_context); +static void s3_upload_queue_retry_cancel(struct flb_s3 *ctx); +static void s3_upload_queue_release(struct flb_s3 *ctx); +static int enqueue_oldest_timed_out_chunk(struct flb_s3 *ctx, time_t now, + int skip_held_tags); +static void complete_pending_uploads(struct flb_s3 *ctx); +static void complete_pending_uploads_once(struct flb_s3 *ctx, int *checked); static int blob_initialize_authorization_endpoint_upstream(struct flb_s3 *context); @@ -421,6 +429,26 @@ static void mock_s3_call_increment_counter(char *api) setenv(env_var, buf, 1); } +static void mock_s3_call_record_uri(char *api, char *uri) +{ + char count_env_var[64]; + char uri_env_var[64]; + char *count; + + if (getenv("TEST_RECORD_S3_URIS") == NULL) { + return; + } + + snprintf(count_env_var, sizeof(count_env_var), "TEST_%s_CALL_COUNT", api); + count = getenv(count_env_var); + if (count == NULL) { + return; + } + + snprintf(uri_env_var, sizeof(uri_env_var), "TEST_%s_URI_%s", api, count); + setenv(uri_env_var, uri, 1); +} + struct flb_http_client *mock_s3_call(char *error_env_var, char *api) { /* create an http client so that we can set the response */ @@ -508,6 +536,23 @@ struct flb_http_client *mock_s3_call(char *error_env_var, char *api) return c; } +static struct flb_http_client *mock_s3_call_for_tag(char *error_env_var, + char *error_tag_env_var, + char *api, + const char *tag, + int tag_len) +{ + char *error_tag; + + error_tag = getenv(error_tag_env_var); + if (error_tag != NULL && + (strlen(error_tag) != tag_len || memcmp(error_tag, tag, tag_len) != 0)) { + return mock_s3_call("TEST_S3_OTHER_TAG_ERROR", api); + } + + return mock_s3_call(error_env_var, api); +} + static flb_sds_t concat_path(char *p1, char *p2) { flb_sds_t dir; @@ -786,10 +831,13 @@ static void s3_context_destroy(struct flb_s3 *ctx) mk_list_foreach_safe(head, tmp, &ctx->upload_queue) { upload_contents = mk_list_entry(head, struct upload_queue, _head); - s3_store_file_delete(ctx, upload_contents->upload_file); remove_from_queue(upload_contents); } + if (ctx->files_mutex_initialized == FLB_TRUE) { + pthread_mutex_destroy(&ctx->files_mutex); + } + flb_free(ctx); } @@ -818,6 +866,14 @@ static int cb_s3_init(struct flb_output_instance *ins, return -1; } ctx->ins = ins; + + ret = pthread_mutex_init(&ctx->files_mutex, NULL); + if (ret != 0) { + flb_free(ctx); + return -1; + } + ctx->files_mutex_initialized = FLB_TRUE; + mk_list_init(&ctx->uploads); mk_list_init(&ctx->upload_queue); @@ -1400,7 +1456,7 @@ static int cb_s3_init(struct flb_output_instance *ins, "executions to S3; buffer=%s", ctx->fs->root_path); ctx->has_old_buffers = FLB_FALSE; - ret = put_all_chunks(ctx); + ret = put_all_chunks(ctx, FLB_TRUE); if (ret < 0) { ctx->has_old_buffers = FLB_TRUE; flb_plg_error(ctx->ins, @@ -1644,6 +1700,7 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, flb_plg_error(ctx->ins, "Could not find or create upload for tag %s", tag); if (chunk) { s3_store_file_unlock(chunk); + chunk->failures += 1; } if (payload_needs_free) { flb_free(payload_buf); @@ -1658,6 +1715,7 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, flb_plg_error(ctx->ins, "Could not initiate multipart upload"); if (chunk) { s3_store_file_unlock(chunk); + chunk->failures += 1; } if (payload_needs_free) { flb_free(payload_buf); @@ -1722,9 +1780,11 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, * Used on shut down to try to send all buffered data * Used on start up to try to send any leftover buffers from previous executions */ -static int put_all_chunks(struct flb_s3 *ctx) +static int put_all_chunks(struct flb_s3 *ctx, int cleanup_empty_streams) { + int result = 0; struct s3_file *chunk; + struct mk_list *stream_tmp; struct mk_list *tmp; struct mk_list *head; struct mk_list *f_head; @@ -1736,7 +1796,7 @@ static int put_all_chunks(struct flb_s3 *ctx) size_t buffer_size; int ret; - mk_list_foreach(head, &ctx->fs->streams) { + mk_list_foreach_safe(head, stream_tmp, &ctx->fs->streams) { /* skip multi upload stream */ fs_stream = mk_list_entry(head, struct flb_fstore_stream, _head); if (fs_stream == ctx->stream_upload) { @@ -1773,7 +1833,12 @@ static int put_all_chunks(struct flb_s3 *ctx) flb_plg_error(ctx->ins, "Could not construct request buffer for %s", chunk->file_path); - return -1; + result = -1; + if (ctx->preserve_data_ordering == FLB_TRUE || + ctx->key_fmt_has_seq_index == FLB_TRUE) { + return result; + } + continue; } #ifdef FLB_HAVE_ARROW @@ -1824,15 +1889,26 @@ static int put_all_chunks(struct flb_s3 *ctx) if (ret < 0) { s3_store_file_unlock(chunk); chunk->failures += 1; - return -1; + result = -1; + if (ctx->preserve_data_ordering == FLB_TRUE || + ctx->key_fmt_has_seq_index == FLB_TRUE) { + return result; + } + continue; } /* data was sent successfully- delete the local buffer */ s3_store_file_delete(ctx, chunk); } + + if (cleanup_empty_streams == FLB_TRUE && + fs_stream != ctx->stream_active && + mk_list_is_empty(&fs_stream->files) == 0) { + flb_fstore_stream_destroy(fs_stream, FLB_TRUE); + } } - return 0; + return result; } /* @@ -1849,6 +1925,11 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, size_t buffer_size = 0; int ret; + if (s3_plugin_under_test() == FLB_TRUE && + getenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR") != NULL) { + return -1; + } + if (new_data == NULL && chunk == NULL) { flb_plg_error(ctx->ins, "[construct_request_buffer] Something went wrong" " both chunk and new_data are NULL"); @@ -1984,7 +2065,10 @@ static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t file_first_ s3_client = ctx->s3_client; if (s3_plugin_under_test() == FLB_TRUE) { - c = mock_s3_call("TEST_PUT_OBJECT_ERROR", "PutObject"); + c = mock_s3_call_for_tag("TEST_PUT_OBJECT_ERROR", + "TEST_PUT_OBJECT_ERROR_TAG", + "PutObject", tag, (int) strlen(tag)); + mock_s3_call_record_uri("PutObject", uri); } else { ret = create_headers(ctx, final_body_md5, &headers, &num_headers, FLB_FALSE); @@ -2193,11 +2277,13 @@ void remove_from_queue(struct upload_queue *entry) } /* Validity check for upload queue object */ -static int upload_queue_valid(struct upload_queue *upload_contents, time_t now, - void *out_context) +static int upload_queue_valid(struct upload_queue *upload_contents, + int *entry_removed, void *out_context) { struct flb_s3 *ctx = out_context; + *entry_removed = FLB_FALSE; + if (upload_contents == NULL) { flb_plg_error(ctx->ins, "Error getting entry from upload_queue"); return -1; @@ -2214,13 +2300,14 @@ static int upload_queue_valid(struct upload_queue *upload_contents, time_t now, return -1; } 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"); + flb_plg_debug(ctx->ins, + "Encountered empty chunk file '%s' for tag '%s' in " + "upload_queue. Deleting empty chunk file", + upload_contents->upload_file->fsf->name, + upload_contents->tag); + s3_store_file_delete(ctx, upload_contents->upload_file); remove_from_queue(upload_contents); - return -1; - } - if (now < upload_contents->upload_time) { - flb_plg_debug(ctx->ins, "Found valid chunk file but not ready to upload"); + *entry_removed = FLB_TRUE; return -1; } return 0; @@ -2242,7 +2329,7 @@ static int send_upload_request(void *out_context, flb_sds_t chunk, if (ret < 0) { flb_plg_error(ctx->ins, "Could not construct request buffer for %s", upload_file->file_path); - return -1; + return FLB_RETRY; } /* Upload to S3 */ @@ -2324,48 +2411,202 @@ static int s3_get_retry_exhausted_action(const char *value) return -1; } +static void s3_upload_queue_retry_cancel(struct flb_s3 *ctx) +{ + if (ctx->upload_queue_retry_timer == NULL) { + return; + } + + flb_sched_timer_invalidate(ctx->upload_queue_retry_timer); + ctx->upload_queue_retry_timer = NULL; +} + +/* + * The scheduler is destroyed before output exit callbacks run, so a retry + * timer cannot be invalidated during shutdown. Release queued file locks so + * the regular shutdown drain can make one final upload attempt instead. + */ +static void s3_upload_queue_release(struct flb_s3 *ctx) +{ + struct mk_list *head; + struct mk_list *tmp; + struct upload_queue *upload_contents; + + ctx->upload_queue_retry_timer = NULL; + + mk_list_foreach_safe(head, tmp, &ctx->upload_queue) { + upload_contents = mk_list_entry(head, struct upload_queue, _head); + s3_store_file_unlock(upload_contents->upload_file); + remove_from_queue(upload_contents); + } +} + +static void s3_upload_queue_retry(struct flb_config *config, void *out_context) +{ + struct flb_s3 *ctx = out_context; + + ctx->upload_queue_retry_timer = NULL; + s3_upload_queue(config, out_context); +} + +static int s3_upload_queue_retry_schedule(struct flb_s3 *ctx, time_t retry_time) +{ + int ret; + int delay_ms; + int64_t delay_seconds; + struct flb_sched *sched; + + if (ctx->upload_queue_retry_timer != NULL) { + return 0; + } + + delay_seconds = retry_time - time(NULL); + if (delay_seconds <= 0) { + delay_ms = 1; + } + else if (delay_seconds > INT_MAX / 1000) { + delay_ms = INT_MAX; + } + else { + delay_ms = (int) delay_seconds * 1000; + } + + sched = flb_sched_ctx_get(); + ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_ONESHOT, + delay_ms, s3_upload_queue_retry, ctx, + &ctx->upload_queue_retry_timer); + if (ret < 0) { + ctx->upload_queue_retry_timer = NULL; + flb_plg_warn(ctx->ins, + "Could not schedule upload queue retry; periodic timer will retry"); + return -1; + } + + return 0; +} + +static int upload_queue_tag_is_blocked(struct flb_s3 *ctx, + struct upload_queue *entry) +{ + struct mk_list *head; + struct upload_queue *queued_entry; + + mk_list_foreach(head, &ctx->upload_queue) { + queued_entry = mk_list_entry(head, struct upload_queue, _head); + if (queued_entry == entry) { + return FLB_FALSE; + } + if (queued_entry->tag_len == entry->tag_len && + memcmp(queued_entry->tag, entry->tag, entry->tag_len) == 0) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + +static int upload_queue_tag_is_held(struct flb_s3 *ctx, const char *tag, + int tag_len, time_t now) +{ + struct mk_list *head; + struct upload_queue *queued_entry; + + mk_list_foreach(head, &ctx->upload_queue) { + queued_entry = mk_list_entry(head, struct upload_queue, _head); + if (queued_entry->tag_len == tag_len && + memcmp(queued_entry->tag, tag, tag_len) == 0) { + return now < queued_entry->upload_time; + } + } + + return FLB_FALSE; +} + /* Uploads all chunk files in queue synchronously */ static void s3_upload_queue(struct flb_config *config, void *out_context) { int ret; + int entry_removed; + int completions_checked; time_t now; + time_t earliest_retry_time; struct upload_queue *upload_contents; + struct s3_file *upload_file; struct flb_s3 *ctx = out_context; struct mk_list *tmp; struct mk_list *head; + (void) config; + completions_checked = FLB_FALSE; + earliest_retry_time = 0; + flb_plg_debug(ctx->ins, "Running upload timer callback (upload_queue).."); +scan: + /* No chunks in upload queue. Scan for timed out chunks. */ if (mk_list_size(&ctx->upload_queue) == 0) { flb_plg_debug(ctx->ins, "No files found in upload_queue. Scanning for timed " "out chunks"); - cb_s3_upload(config, out_context); + ret = enqueue_oldest_timed_out_chunk(ctx, time(NULL), FLB_FALSE); + if (ret < 0) { + flb_plg_error(ctx->ins, + "Could not add timed out chunk to upload queue"); + } + if (ret <= 0) { + complete_pending_uploads_once(ctx, &completions_checked); + goto exit; + } } /* Iterate through each file in upload queue */ mk_list_foreach_safe(head, tmp, &ctx->upload_queue) { upload_contents = mk_list_entry(head, struct upload_queue, _head); + if (upload_queue_tag_is_blocked(ctx, upload_contents) == FLB_TRUE) { + continue; + } + now = time(NULL); + if (now < upload_contents->upload_time) { + flb_plg_debug(ctx->ins, + "Found tag queue head but it is not ready to upload"); + if (ctx->key_fmt_has_seq_index == FLB_TRUE) { + s3_upload_queue_retry_schedule(ctx, upload_contents->upload_time); + complete_pending_uploads_once(ctx, &completions_checked); + goto exit; + } + if (earliest_retry_time == 0 || + upload_contents->upload_time < earliest_retry_time) { + earliest_retry_time = upload_contents->upload_time; + } + continue; + } + /* Checks if upload_contents is valid */ - ret = upload_queue_valid(upload_contents, now, ctx); + ret = upload_queue_valid(upload_contents, &entry_removed, ctx); if (ret < 0) { + if (entry_removed == FLB_TRUE) { + continue; + } + complete_pending_uploads_once(ctx, &completions_checked); goto exit; } + upload_contents->m_upload_file = get_upload( + ctx, upload_contents->tag, upload_contents->tag_len); + /* Try to upload file. Return value can be -1, FLB_OK, FLB_ERROR, FLB_RETRY. */ ret = send_upload_request(ctx, NULL, upload_contents->upload_file, upload_contents->m_upload_file, upload_contents->tag, upload_contents->tag_len); - if (ret < 0) { - goto exit; - } - else if (ret == FLB_OK) { + if (ret == FLB_OK) { + s3_upload_queue_retry_cancel(ctx); remove_from_queue(upload_contents); ctx->retry_time = 0; ctx->upload_queue_success = FLB_TRUE; + complete_pending_uploads_once(ctx, &completions_checked); } else { s3_store_file_lock(upload_contents->upload_file); @@ -2376,24 +2617,64 @@ static void s3_upload_queue(struct flb_config *config, void *out_context) if (upload_contents->retry_counter > ctx->ins->retry_limit) { flb_plg_warn(ctx->ins, "Chunk file failed to send %d times, will not " "retry", upload_contents->retry_counter); - s3_chunk_retry_exhausted_cleanup(ctx, upload_contents->upload_file); - if (upload_contents->m_upload_file) { - mk_list_del(&upload_contents->m_upload_file->_head); - } - multipart_upload_destroy(upload_contents->m_upload_file); + upload_file = upload_contents->upload_file; + /* + * Multipart uploads are shared by tag and may contain parts + * already accepted by S3. Leave their persisted state intact. + */ remove_from_queue(upload_contents); + s3_chunk_retry_exhausted_cleanup(ctx, upload_file); + s3_upload_queue_retry_cancel(ctx); + ctx->retry_time = 0; + complete_pending_uploads_once(ctx, &completions_checked); continue; } /* Retry in N seconds */ upload_contents->upload_time = now + 2 * upload_contents->retry_counter; ctx->retry_time += 2 * upload_contents->retry_counter; + if (ctx->key_fmt_has_seq_index == FLB_TRUE) { + s3_upload_queue_retry_schedule(ctx, upload_contents->upload_time); + flb_plg_debug(ctx->ins, + "Failed to upload file in upload_queue. Will not " + "retry for %d seconds", + 2 * upload_contents->retry_counter); + complete_pending_uploads_once(ctx, &completions_checked); + break; + } + if (earliest_retry_time == 0 || + upload_contents->upload_time < earliest_retry_time) { + earliest_retry_time = upload_contents->upload_time; + } flb_plg_debug(ctx->ins, "Failed to upload file in upload_queue. Will not " - "retry for %d seconds", 2 * upload_contents->retry_counter); - break; + "retry for %d seconds", + 2 * upload_contents->retry_counter); + complete_pending_uploads_once(ctx, &completions_checked); } } + if (ctx->key_fmt_has_seq_index == FLB_TRUE) { + if (mk_list_size(&ctx->upload_queue) == 0) { + goto scan; + } + goto exit; + } + + ret = enqueue_oldest_timed_out_chunk(ctx, time(NULL), FLB_TRUE); + if (ret > 0) { + goto scan; + } + if (ret < 0) { + flb_plg_error(ctx->ins, "Could not add timed out chunk to upload queue"); + } + + s3_upload_queue_retry_cancel(ctx); + if (earliest_retry_time != 0) { + s3_upload_queue_retry_schedule(ctx, earliest_retry_time); + } + + complete_pending_uploads_once(ctx, &completions_checked); + exit: return; } @@ -3729,64 +4010,81 @@ static int cb_s3_upload_blob(struct flb_config *config, void *data) -static void cb_s3_upload(struct flb_config *config, void *data) +static int enqueue_oldest_timed_out_chunk(struct flb_s3 *ctx, time_t now, + int skip_held_tags) { - struct flb_s3 *ctx = data; - struct s3_file *chunk = NULL; - struct multipart_upload *m_upload = NULL; + int ret; + struct s3_file *chunk; + struct s3_file *oldest_chunk; + struct multipart_upload *m_upload; struct flb_fstore_file *fsf; - char *buffer = NULL; - size_t buffer_size = 0; + struct flb_fstore_file *oldest_fsf; struct mk_list *tmp; struct mk_list *head; - int complete; - int ret; - time_t now; - flb_plg_info(ctx->ins, "Running upload timer callback (cb_s3_upload).."); - - now = time(NULL); + oldest_chunk = NULL; + oldest_fsf = NULL; - /* Check all chunks and see if any have timed out */ mk_list_foreach_safe(head, tmp, &ctx->stream_active->files) { fsf = mk_list_entry(head, struct flb_fstore_file, _head); chunk = fsf->data; - if (now < (chunk->create_time + ctx->upload_timeout + ctx->retry_time)) { - continue; /* Only send chunks which have timed out */ + if (chunk->locked == FLB_TRUE) { + continue; } - /* Locked chunks are being processed, skip */ - if (chunk->locked == FLB_TRUE) { + if (chunk->failures > ctx->ins->retry_limit) { + flb_plg_warn(ctx->ins, + "Chunk for tag %s failed to send %d/%d times, will not retry", + (char *) fsf->meta_buf, chunk->failures, + ctx->ins->retry_limit); + s3_chunk_retry_exhausted_cleanup(ctx, chunk); continue; } - m_upload = get_upload(ctx, (const char *) fsf->meta_buf, fsf->meta_size); + if (now < (chunk->create_time + ctx->upload_timeout + ctx->retry_time)) { + continue; + } - ret = construct_request_buffer(ctx, NULL, chunk, &buffer, &buffer_size); - if (ret < 0) { - flb_plg_error(ctx->ins, "Could not construct request buffer for %s", - chunk->file_path); + if (skip_held_tags == FLB_TRUE && + upload_queue_tag_is_held(ctx, (const char *) fsf->meta_buf, + fsf->meta_size, now) == FLB_TRUE) { continue; } - /* FYI: if construct_request_buffer() succeedeed, the s3_file is locked */ - ret = upload_data(ctx, chunk, m_upload, buffer, buffer_size, - (const char *) fsf->meta_buf, fsf->meta_size); - flb_free(buffer); - if (ret != FLB_OK) { - flb_plg_error(ctx->ins, "Could not send chunk with tag %s", - (char *) fsf->meta_buf); - if(chunk->failures > ctx->ins->retry_limit){ - flb_plg_warn(ctx->ins, - "Chunk for tag %s failed to send %d/%d times, will not retry", - (char *) fsf->meta_buf, chunk->failures, ctx->ins->retry_limit); - s3_chunk_retry_exhausted_cleanup(ctx, chunk); - continue; - } + if (oldest_chunk == NULL || + chunk->create_time < oldest_chunk->create_time) { + oldest_chunk = chunk; + oldest_fsf = fsf; } } + if (oldest_chunk == NULL) { + return 0; + } + + m_upload = get_upload(ctx, (const char *) oldest_fsf->meta_buf, + oldest_fsf->meta_size); + s3_store_file_lock(oldest_chunk); + ret = add_to_queue(ctx, oldest_chunk, m_upload, + (const char *) oldest_fsf->meta_buf, + oldest_fsf->meta_size); + if (ret < 0) { + s3_store_file_unlock(oldest_chunk); + return -1; + } + + return 1; +} + +static void complete_pending_uploads(struct flb_s3 *ctx) +{ + struct multipart_upload *m_upload = NULL; + struct mk_list *tmp; + struct mk_list *head; + int complete; + int ret; + /* Check all uploads and see if any need completion */ mk_list_foreach_safe(head, tmp, &ctx->uploads) { m_upload = mk_list_entry(head, struct multipart_upload, _head); @@ -3796,6 +4094,7 @@ static void cb_s3_upload(struct flb_config *config, void *data) flb_plg_error(ctx->ins, "Upload for %s has reached max completion errors, " "plugin will give up", m_upload->s3_key); + /* Keep the persisted upload available for restart recovery. */ mk_list_del(&m_upload->_head); multipart_upload_destroy(m_upload); continue; @@ -3808,7 +4107,8 @@ static void cb_s3_upload(struct flb_config *config, void *data) if (m_upload->upload_state == MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS) { complete = FLB_TRUE; } - if (time(NULL) > (m_upload->init_time + ctx->upload_timeout + ctx->retry_time)) { + if (time(NULL) > + (m_upload->init_time + ctx->upload_timeout + ctx->retry_time)) { flb_plg_info(ctx->ins, "Completing upload for %s because upload_timeout" " has passed", m_upload->s3_key); complete = FLB_TRUE; @@ -3829,6 +4129,91 @@ static void cb_s3_upload(struct flb_config *config, void *data) } } } +} + +static void complete_pending_uploads_once(struct flb_s3 *ctx, int *checked) +{ + if (*checked == FLB_TRUE) { + return; + } + + complete_pending_uploads(ctx); + *checked = FLB_TRUE; +} + +static void cb_s3_upload(struct flb_config *config, void *data) +{ + struct flb_s3 *ctx = data; + struct s3_file *chunk = NULL; + struct multipart_upload *m_upload = NULL; + struct flb_fstore_file *fsf; + char *buffer = NULL; + size_t buffer_size = 0; + struct mk_list *tmp; + struct mk_list *head; + int ret; + time_t now; + + (void) config; + + flb_plg_info(ctx->ins, "Running upload timer callback (cb_s3_upload).."); + + now = time(NULL); + + if (ctx->preserve_data_ordering == FLB_TRUE) { + ret = enqueue_oldest_timed_out_chunk(ctx, now, FLB_FALSE); + if (ret < 0) { + flb_plg_error(ctx->ins, + "Could not add timed out chunk to upload queue"); + } + complete_pending_uploads(ctx); + return; + } + + /* Check all chunks and see if any have timed out */ + mk_list_foreach_safe(head, tmp, &ctx->stream_active->files) { + fsf = mk_list_entry(head, struct flb_fstore_file, _head); + chunk = fsf->data; + + if (now < (chunk->create_time + ctx->upload_timeout + ctx->retry_time)) { + continue; /* Only send chunks which have timed out */ + } + + /* Locked chunks are being processed, skip */ + if (chunk->locked == FLB_TRUE) { + continue; + } + + m_upload = get_upload(ctx, (const char *) fsf->meta_buf, fsf->meta_size); + + ret = construct_request_buffer(ctx, NULL, chunk, &buffer, &buffer_size); + if (ret < 0) { + flb_plg_error(ctx->ins, "Could not construct request buffer for %s", + chunk->file_path); + chunk->failures += 1; + } + else { + /* FYI: if construct_request_buffer() succeeded, the s3_file is locked */ + ret = upload_data(ctx, chunk, m_upload, buffer, buffer_size, + (const char *) fsf->meta_buf, fsf->meta_size); + flb_free(buffer); + } + + if (ret != FLB_OK) { + flb_plg_error(ctx->ins, "Could not send chunk with tag %s", + (char *) fsf->meta_buf); + if (chunk->failures > ctx->ins->retry_limit) { + flb_plg_warn(ctx->ins, + "Chunk for tag %s failed to send %d/%d times, will not retry", + (char *) fsf->meta_buf, chunk->failures, + ctx->ins->retry_limit); + s3_chunk_retry_exhausted_cleanup(ctx, chunk); + continue; + } + } + } + + complete_pending_uploads(ctx); } @@ -4001,7 +4386,7 @@ static void flush_init(void *out_context) "executions to S3; buffer=%s", ctx->fs->root_path); ctx->has_old_buffers = FLB_FALSE; - ret = put_all_chunks(ctx); + ret = put_all_chunks(ctx, FLB_TRUE); if (ret < 0) { ctx->has_old_buffers = FLB_TRUE; flb_plg_error(ctx->ins, @@ -4406,9 +4791,11 @@ static int cb_s3_exit(void *data, struct flb_config *config) return 0; } + s3_upload_queue_release(ctx); + if (s3_store_has_data(ctx) == FLB_TRUE) { flb_plg_info(ctx->ins, "Sending all locally buffered data to S3"); - ret = put_all_chunks(ctx); + ret = put_all_chunks(ctx, FLB_FALSE); if (ret < 0) { flb_plg_error(ctx->ins, "Could not send all chunks on exit"); } diff --git a/plugins/out_s3/s3.h b/plugins/out_s3/s3.h index 744cbbf7fce..747f8371e24 100644 --- a/plugins/out_s3/s3.h +++ b/plugins/out_s3/s3.h @@ -26,6 +26,7 @@ #include #include #include +#include /* S3 output format types */ #define FLB_S3_FORMAT_JSON_LINES 0 @@ -71,6 +72,7 @@ struct upload_queue { struct s3_file *upload_file; + /* Non-owning reference; refresh it before every upload attempt. */ struct multipart_upload *m_upload_file; flb_sds_t tag; int tag_len; @@ -178,6 +180,8 @@ struct flb_s3 { struct flb_fstore_stream *stream_upload; /* multipart upload stream */ struct flb_fstore_stream *stream_quarantine; /* retry-exhausted stream */ struct flb_fstore_stream *stream_metadata; /* s3 metadata stream */ + pthread_mutex_t files_mutex; + int files_mutex_initialized; /* * used to track that unset buffers were found on startup that have not @@ -192,6 +196,7 @@ struct flb_s3 { int preserve_data_ordering; int upload_queue_success; struct mk_list upload_queue; + struct flb_sched_timer *upload_queue_retry_timer; size_t file_size; size_t upload_chunk_size; diff --git a/plugins/out_s3/s3_multipart.c b/plugins/out_s3/s3_multipart.c index 0a82fdd8a20..ef51fbcbb06 100644 --- a/plugins/out_s3/s3_multipart.c +++ b/plugins/out_s3/s3_multipart.c @@ -303,7 +303,8 @@ static int save_upload(struct flb_s3 *ctx, struct multipart_upload *m_upload, return ret; } -static int remove_upload_from_fs(struct flb_s3 *ctx, struct multipart_upload *m_upload) +static int remove_upload_from_fs(struct flb_s3 *ctx, + struct multipart_upload *m_upload) { flb_sds_t key; struct flb_fstore_file *fsf; diff --git a/plugins/out_s3/s3_store.c b/plugins/out_s3/s3_store.c index fe689d83b0b..731ef0645ec 100644 --- a/plugins/out_s3/s3_store.c +++ b/plugins/out_s3/s3_store.c @@ -206,6 +206,8 @@ struct s3_file *s3_store_file_get(struct flb_s3 *ctx, const char *tag, struct flb_fstore_file *fsf = NULL; struct s3_file *s3_file; + pthread_mutex_lock(&ctx->files_mutex); + /* * Based in the current ctx->stream_name, locate a candidate file to * store the incoming data using as a lookup pattern the content Tag. @@ -217,6 +219,8 @@ struct s3_file *s3_store_file_get(struct flb_s3 *ctx, const char *tag, if (fsf->data == NULL) { flb_plg_warn(ctx->ins, "BAD: found flb_fstore_file with NULL data reference, tag=%s, file=%s, will try to delete", tag, fsf->name); flb_fstore_file_delete(ctx->fs, fsf); + fsf = NULL; + continue; } if (fsf->meta_size != tag_len) { @@ -241,10 +245,14 @@ struct s3_file *s3_store_file_get(struct flb_s3 *ctx, const char *tag, } if (!fsf) { + pthread_mutex_unlock(&ctx->files_mutex); return NULL; } - return fsf->data; + s3_file = fsf->data; + pthread_mutex_unlock(&ctx->files_mutex); + + return s3_file; } /* Append data to a new or existing fstore file */ @@ -254,11 +262,15 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, time_t file_first_log_time) { int ret; + int result; flb_sds_t name; struct flb_fstore_file *fsf; uint64_t current_buffer_size; uint64_t new_buffer_size; + result = -1; + pthread_mutex_lock(&ctx->files_mutex); + ret = buffer_size_reserve(ctx, bytes, ¤t_buffer_size, &new_buffer_size); if (ret < 0) { @@ -266,7 +278,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, "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; + goto done; } /* If no target file was found, create a new one */ @@ -275,7 +287,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, if (!name) { flb_plg_error(ctx->ins, "could not generate chunk file name"); buffer_size_release(ctx, bytes); - return -1; + goto done; } /* Create the file */ @@ -285,7 +297,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, name); flb_sds_destroy(name); buffer_size_release(ctx, bytes); - return -1; + goto done; } flb_sds_destroy(name); @@ -296,7 +308,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, 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; + goto done; } /* Allocate local context */ @@ -307,7 +319,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, 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; + goto done; } s3_file->fsf = fsf; s3_file->first_log_time = file_first_log_time; @@ -325,7 +337,7 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, if (ret != 0) { flb_plg_error(ctx->ins, "error writing data to local s3 file"); buffer_size_release(ctx, bytes); - return -1; + goto done; } ret = counter_add(&s3_file->size, (uint64_t) bytes, NULL); if (ret < 0) { @@ -342,7 +354,45 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, new_buffer_size, ctx->store_dir_limit_size); } - return 0; + result = 0; + +done: + pthread_mutex_unlock(&ctx->files_mutex); + return result; +} + +static ssize_t restored_file_size_get(struct flb_s3 *ctx, + struct flb_fstore_file *fsf) +{ + int ret; + int restore_down; + ssize_t file_size; + + restore_down = FLB_FALSE; + + if (cio_chunk_is_up(fsf->chunk) == CIO_FALSE) { + ret = cio_chunk_up_force(fsf->chunk); + if (ret != CIO_OK) { + flb_plg_error(ctx->ins, + "cannot load restored S3 chunk '%s' to determine its size", + fsf->name); + return -1; + } + restore_down = FLB_TRUE; + } + + file_size = cio_chunk_get_content_size(fsf->chunk); + + if (restore_down == FLB_TRUE) { + ret = cio_chunk_down(fsf->chunk); + if (ret != CIO_OK) { + flb_plg_warn(ctx->ins, + "cannot return restored S3 chunk '%s' to the down state", + fsf->name); + } + } + + return file_size; } static int set_files_context(struct flb_s3 *ctx) @@ -379,7 +429,7 @@ 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_content_size(fsf->chunk); + file_size = restored_file_size_get(ctx, fsf); if (file_size > 0) { cfl_atomic_store(&s3_file->size, (uint64_t) file_size); diff --git a/tests/runtime/out_s3.c b/tests/runtime/out_s3.c index 3aa8c114e9c..6134a6d49cd 100644 --- a/tests/runtime/out_s3.c +++ b/tests/runtime/out_s3.c @@ -22,6 +22,7 @@ #define S3_TEST_UPLOAD_TIMEOUT "1s" #define S3_TEST_WAIT_STEP_MS 10 #define S3_TEST_WAIT_TIMEOUT_MS 5000 +#define S3_TEST_STARTUP_FILE_COUNT (CIO_MAX_CHUNKS_UP + 2) /* not a real error code, but tests that the code can respond to any error */ #define ERROR_ACCESS_DENIED "\ @@ -111,6 +112,40 @@ static int count_files_recursive(const char *path) #endif } +static int count_sized_down_s3_files(struct flb_s3 *ctx) +{ + int count; + struct s3_file *s3_file; + struct flb_fstore_file *fsf; + struct flb_fstore_stream *fs_stream; + struct mk_list *file_head; + struct mk_list *stream_head; + + count = 0; + + mk_list_foreach(stream_head, &ctx->fs->streams) { + fs_stream = mk_list_entry(stream_head, struct flb_fstore_stream, _head); + if (fs_stream == ctx->stream_upload) { + continue; + } + + mk_list_foreach(file_head, &fs_stream->files) { + fsf = mk_list_entry(file_head, struct flb_fstore_file, _head); + if (fsf->chunk == NULL || fsf->data == NULL || + cio_chunk_is_up(fsf->chunk) == CIO_TRUE) { + continue; + } + + s3_file = fsf->data; + if (cfl_atomic_load(&s3_file->size) > 0) { + count++; + } + } + } + + return count; +} + static int get_s3_call_count(const char *api) { char name[64]; @@ -122,7 +157,8 @@ static int get_s3_call_count(const char *api) return value ? atoi(value) : 0; } -static void wait_for_s3_call_count(const char *api, int expected) +static void wait_for_s3_call_count_with_timeout(const char *api, int expected, + uint64_t timeout_ms) { uint64_t elapsed_ms; struct flb_time start_time; @@ -133,7 +169,7 @@ static void wait_for_s3_call_count(const char *api, int expected) flb_time_get(&start_time); while (get_s3_call_count(api) < expected && - elapsed_ms < S3_TEST_WAIT_TIMEOUT_MS) { + elapsed_ms < timeout_ms) { flb_time_msleep(S3_TEST_WAIT_STEP_MS); flb_time_get(&end_time); flb_time_diff(&end_time, &start_time, &diff_time); @@ -141,6 +177,12 @@ static void wait_for_s3_call_count(const char *api, int expected) } } +static void wait_for_s3_call_count(const char *api, int expected) +{ + wait_for_s3_call_count_with_timeout(api, expected, + S3_TEST_WAIT_TIMEOUT_MS); +} + static void wait_for_file_count(const char *path, int expected) { uint64_t elapsed_ms; @@ -160,6 +202,25 @@ static void wait_for_file_count(const char *path, int expected) } } +static void wait_for_file_count_at_most(const char *path, int expected) +{ + uint64_t elapsed_ms; + struct flb_time start_time; + struct flb_time end_time; + struct flb_time diff_time; + + elapsed_ms = 0; + flb_time_get(&start_time); + + while (count_files_recursive(path) > expected && + elapsed_ms < S3_TEST_WAIT_TIMEOUT_MS) { + flb_time_msleep(S3_TEST_WAIT_STEP_MS); + flb_time_get(&end_time); + flb_time_diff(&end_time, &start_time, &diff_time); + elapsed_ms = flb_time_to_nanosec(&diff_time) / 1000000; + } +} + static int ensure_test_directory(const char *path) { #ifdef FLB_SYSTEM_WINDOWS @@ -190,6 +251,21 @@ static int ensure_test_directory(const char *path) #endif } +static int test_directory_exists(const char *path) +{ +#ifdef FLB_SYSTEM_WINDOWS + DWORD attributes; + + attributes = GetFileAttributesA(path); + return attributes != INVALID_FILE_ATTRIBUTES && + (attributes & FILE_ATTRIBUTE_DIRECTORY) != 0; +#else + struct stat st; + + return stat(path, &st) == 0 && S_ISDIR(st.st_mode); +#endif +} + static char *create_test_store_directory(const char *postfix) { char *store_dir; @@ -378,6 +454,7 @@ void flb_test_s3_create_upload_error(void) char *call_count_str; int call_count; char *store_dir; + struct flb_s3 *s3_ctx; store_dir = create_test_store_directory("/flb-s3-test-XXXXXX"); TEST_CHECK(store_dir != NULL); @@ -403,18 +480,27 @@ void flb_test_s3_create_upload_error(void) 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,"Retry_Limit", "1", NULL); + flb_output_set(ctx, out_ffd,"retry_exhausted_action", "delete", NULL); + flb_output_set(ctx, out_ffd,"preserve_data_ordering", "false", NULL); ret = flb_start(ctx); TEST_CHECK(ret == 0); flb_lib_push(ctx, in_ffd, (char *) JSON_TD , (int) sizeof(JSON_TD) - 1); - wait_for_s3_call_count("CreateMultipartUpload", 1); + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + + wait_for_s3_call_count("CreateMultipartUpload", 2); + wait_for_file_count_at_most(s3_ctx->stream_active->path, 0); call_count_str = getenv("TEST_CreateMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; - TEST_CHECK_(call_count >= 1, - "Expected >= 1 CreateMultipartUpload calls, got %d", call_count); + TEST_CHECK_(call_count == 2, + "Expected CreateMultipartUpload to stop after two attempts, got %d", + call_count); + TEST_CHECK_(count_files_recursive(s3_ctx->stream_active->path) == 0, + "Expected retry-exhausted chunk to be deleted"); call_count_str = getenv("TEST_UploadPart_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; @@ -487,7 +573,415 @@ void flb_test_s3_upload_part_error(void) flb_stop(ctx); flb_destroy(ctx); unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_UPLOAD_PART_ERROR"); + unsetenv("TEST_UPLOAD_PART_ERROR"); + unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); + unsetenv("TEST_UploadPart_CALL_COUNT"); + unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + unsetenv("TEST_PutObject_CALL_COUNT"); + flb_free(store_dir); +} + +void flb_test_s3_complete_upload_error(void) +{ + int ret; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + char *call_count_str; + int call_count; + char *store_dir; + struct flb_s3 *s3_ctx; + + store_dir = create_test_store_directory("/flb-s3-test-upload-err-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + /* mocks calls- signals that we are in test mode */ + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_COMPLETE_MULTIPART_UPLOAD_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,"upload_timeout", S3_TEST_UPLOAD_TIMEOUT, NULL); + flb_output_set(ctx, out_ffd,"store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd,"Retry_Limit", "1", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + + flb_lib_push(ctx, in_ffd, (char *) JSON_TD , (int) sizeof(JSON_TD) - 1); + + wait_for_s3_call_count("CompleteMultipartUpload", 2); + + call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count = call_count_str ? atoi(call_count_str) : 0; + TEST_CHECK_(call_count >= 2, + "Expected >= 2 CompleteMultipartUpload calls (retried), got %d", + call_count); + + wait_for_file_count(s3_ctx->stream_upload->path, 1); + TEST_CHECK_(count_files_recursive(s3_ctx->stream_upload->path) > 0, + "Expected multipart metadata to remain recoverable after completion failure"); + + flb_stop(ctx); + flb_destroy(ctx); + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); + unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); + unsetenv("TEST_UploadPart_CALL_COUNT"); + unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + unsetenv("TEST_PutObject_CALL_COUNT"); + flb_free(store_dir); +} + +void flb_test_s3_ordered_retry_uses_backoff_deadline(void) +{ + int ret; + int in_ffd; + int out_ffd; + 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-retry-deadline-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_UPLOAD_PART_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", "retry-deadline", 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", "false", NULL); + flb_output_set(ctx, out_ffd, "compression", "gzip", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "100M", NULL); + flb_output_set(ctx, out_ffd, "upload_chunk_size", "50M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "60s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "1", 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); + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + wait_for_file_count(s3_ctx->stream_active->path, 1); + + s3_file = s3_store_file_get(s3_ctx, "retry-deadline", 14); + TEST_CHECK(s3_file != NULL); + s3_file->create_time = time(NULL) - 61; + + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_s3_call_count("UploadPart", 1); + flb_time_msleep(100); + TEST_CHECK_(s3_ctx->retry_time == 2, + "Expected first retry to add a 2 second timeout offset, got %lld", + (long long) s3_ctx->retry_time); + wait_for_file_count_at_most(s3_ctx->stream_active->path, 0); + + TEST_CHECK_(get_s3_call_count("UploadPart") == 2, + "Expected initial attempt and one retry at its deadline, got %d", + get_s3_call_count("UploadPart")); + TEST_CHECK_(count_files_recursive(s3_ctx->stream_active->path) == 0, + "Expected retry-exhausted chunk to be deleted before periodic timer"); + TEST_CHECK_(s3_ctx->retry_time == 0, + "Expected terminal cleanup to reset the timeout offset"); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_UPLOAD_PART_ERROR"); + unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); + unsetenv("TEST_UploadPart_CALL_COUNT"); + flb_free(store_dir); +} + +void flb_test_s3_ordered_timer_isolates_tag_backoff(void) +{ + int ret; + int index; + int out_ffd; + int input_fds[2]; + char *uri; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + struct s3_file *oldest_file; + struct s3_file *later_file; + + store_dir = create_test_store_directory("/flb-s3-test-timer-order-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_RECORD_S3_URIS", "true", 1); + setenv("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); + + ctx = flb_create(); + for (index = 0; index < 2; index++) { + input_fds[index] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(input_fds[index] >= 0); + } + flb_input_set(ctx, input_fds[0], "tag", "oldest", NULL); + flb_input_set(ctx, input_fds[1], "tag", "later", 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", "10s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "s3_key_format", "/queue/$TAG", NULL); + flb_output_set(ctx, out_ffd, "static_file_path", "true", NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "1", NULL); + flb_output_set(ctx, out_ffd, "retry_exhausted_action", "delete", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + for (index = 0; index < 2; index++) { + ret = flb_lib_push(ctx, input_fds[index], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + } + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + wait_for_file_count(s3_ctx->stream_active->path, 2); + oldest_file = s3_store_file_get(s3_ctx, "oldest", 6); + later_file = s3_store_file_get(s3_ctx, "later", 5); + TEST_CHECK(oldest_file != NULL); + TEST_CHECK(later_file != NULL); + oldest_file->create_time = time(NULL) - 30; + later_file->create_time = time(NULL) - 20; + + setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + wait_for_s3_call_count("PutObject", 1); + flb_time_msleep(500); + + TEST_CHECK_(get_s3_call_count("PutObject") == 2, + "Expected the healthy tag to upload during another tag's backoff, got %d calls", + get_s3_call_count("PutObject")); + TEST_CHECK_(s3_ctx->retry_time == 0, + "Expected the healthy upload to preserve the global retry reset behavior"); + uri = getenv("TEST_PutObject_URI_1"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest") == 0, + "Expected oldest timed-out chunk first, got %s", + uri ? uri : "(null)"); + uri = getenv("TEST_PutObject_URI_2"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/later") == 0, + "Expected healthy tag during oldest tag's backoff, got %s", + uri ? uri : "(null)"); + + unsetenv("TEST_PUT_OBJECT_ERROR"); + wait_for_s3_call_count("PutObject", 3); + uri = getenv("TEST_PutObject_URI_3"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest") == 0, + "Expected failed tag to retry at its deadline, got %s", + uri ? uri : "(null)"); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_PUT_OBJECT_ERROR"); + unsetenv("TEST_PUT_OBJECT_ERROR_TAG"); + unsetenv("TEST_RECORD_S3_URIS"); + unsetenv("TEST_PutObject_CALL_COUNT"); + unsetenv("TEST_PutObject_URI_1"); + unsetenv("TEST_PutObject_URI_2"); + unsetenv("TEST_PutObject_URI_3"); + flb_free(store_dir); +} + +void flb_test_s3_ordered_construct_error_exhausts_chunk(void) +{ + int ret; + int in_ffd; + int out_ffd; + 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-construct-error-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + 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", "construct-error", 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", "60s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "1", 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); + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + wait_for_file_count(s3_ctx->stream_active->path, 1); + + s3_file = s3_store_file_get(s3_ctx, "construct-error", 15); + TEST_CHECK(s3_file != NULL); + s3_file->create_time = time(NULL) - 61; + setenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR", "true", 1); + + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count_at_most(s3_ctx->stream_active->path, 0); + + TEST_CHECK_(count_files_recursive(s3_ctx->stream_active->path) == 0, + "Expected unreadable queue head to reach terminal cleanup"); + TEST_CHECK_(mk_list_is_empty(&s3_ctx->upload_queue) == 0, + "Expected queue to be empty after construction retry exhaustion"); + TEST_CHECK_(s3_ctx->retry_time == 0, + "Expected retry delay to reset after terminal cleanup"); + + unsetenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR"); + 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_ordered_backoff_does_not_starve_completion(void) +{ + int ret; + int index; + int out_ffd; + int input_fds[2]; + 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-completion-backoff-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + + ctx = flb_create(); + for (index = 0; index < 2; index++) { + input_fds[index] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(input_fds[index] >= 0); + } + flb_input_set(ctx, input_fds[0], "tag", "completing", NULL); + flb_input_set(ctx, input_fds[1], "tag", "backoff", 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", "false", NULL); + flb_output_set(ctx, out_ffd, "compression", "gzip", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "100M", NULL); + flb_output_set(ctx, out_ffd, "upload_chunk_size", "50M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "60s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "5", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + + ret = flb_lib_push(ctx, input_fds[0], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 1); + s3_file = s3_store_file_get(s3_ctx, "completing", 10); + TEST_CHECK(s3_file != NULL); + s3_file->create_time = time(NULL) - 61; + ret = flb_lib_push(ctx, input_fds[0], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_s3_call_count("CompleteMultipartUpload", 1); + + ret = flb_lib_push(ctx, input_fds[1], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 1); + s3_file = s3_store_file_get(s3_ctx, "backoff", 7); + TEST_CHECK(s3_file != NULL); + s3_file->create_time = time(NULL) - 61; + setenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + ret = flb_lib_push(ctx, input_fds[1], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + + wait_for_s3_call_count("CreateMultipartUpload", 2); + wait_for_s3_call_count("CompleteMultipartUpload", 2); + TEST_CHECK_(get_s3_call_count("CompleteMultipartUpload") >= 2, + "Expected pending completion to run while queue head was backing off"); + + unsetenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR"); + unsetenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); unsetenv("TEST_UploadPart_CALL_COUNT"); unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); @@ -495,58 +989,103 @@ void flb_test_s3_upload_part_error(void) flb_free(store_dir); } -void flb_test_s3_complete_upload_error(void) +void flb_test_s3_ordered_shared_upload_retries_safely(void) { int ret; - flb_ctx_t *ctx; int in_ffd; int out_ffd; - char *call_count_str; - int call_count; + flb_ctx_t *ctx; char *store_dir; + char first_file_name[128]; + char second_file_name[128]; + struct flb_s3 *s3_ctx; + struct s3_file *s3_file; - store_dir = create_test_store_directory("/flb-s3-test-upload-err-XXXXXX"); + store_dir = create_test_store_directory("/flb-s3-test-shared-upload-XXXXXX"); TEST_CHECK(store_dir != NULL); if (store_dir == NULL) { return; } - /* mocks calls- signals that we are in test mode */ setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + setenv("TEST_UPLOAD_PART_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); + flb_input_set(ctx, in_ffd, "tag", "shared-upload", 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,"upload_timeout", S3_TEST_UPLOAD_TIMEOUT, NULL); - flb_output_set(ctx, out_ffd,"store_dir", store_dir, NULL); - flb_output_set(ctx, out_ffd,"Retry_Limit", "1", NULL); + 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", "false", NULL); + flb_output_set(ctx, out_ffd, "compression", "gzip", NULL); + flb_output_set(ctx, out_ffd, "total_file_size", "100M", NULL); + flb_output_set(ctx, out_ffd, "upload_chunk_size", "50M", NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "60s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "1", 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); - flb_lib_push(ctx, in_ffd, (char *) JSON_TD , (int) sizeof(JSON_TD) - 1); + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 1); + s3_file = s3_store_file_get(s3_ctx, "shared-upload", 13); + TEST_CHECK(s3_file != NULL); + snprintf(first_file_name, sizeof(first_file_name), "%s", + s3_file->fsf->name); + s3_file->create_time = time(NULL) - 61; - wait_for_s3_call_count("CompleteMultipartUpload", 2); + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_s3_call_count("UploadPart", 1); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - call_count = call_count_str ? atoi(call_count_str) : 0; - TEST_CHECK_(call_count >= 2, - "Expected >= 2 CompleteMultipartUpload calls (retried), got %d", - call_count); + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 2); + s3_file = s3_store_file_get(s3_ctx, "shared-upload", 13); + TEST_CHECK(s3_file != NULL); + snprintf(second_file_name, sizeof(second_file_name), "%s", + s3_file->fsf->name); + TEST_CHECK_(strcmp(first_file_name, second_file_name) != 0, + "Expected two distinct shared-upload chunks"); + s3_file->create_time = time(NULL) - 61; + + wait_for_s3_call_count("UploadPart", 3); + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 2); + wait_for_s3_call_count("UploadPart", 4); + wait_for_file_count_at_most(s3_ctx->stream_active->path, 1); + + TEST_CHECK_(get_s3_call_count("UploadPart") == 4, + "Expected both shared-upload chunks to exhaust safely, got %d attempts", + get_s3_call_count("UploadPart")); + TEST_CHECK_(flb_fstore_file_get(s3_ctx->fs, s3_ctx->stream_active, + first_file_name, strlen(first_file_name)) == NULL, + "Expected first retry-exhausted chunk to be deleted"); + TEST_CHECK_(flb_fstore_file_get(s3_ctx->fs, s3_ctx->stream_active, + second_file_name, strlen(second_file_name)) == NULL, + "Expected second retry-exhausted chunk to be deleted"); + TEST_CHECK_(mk_list_is_empty(&s3_ctx->upload_queue) == 0, + "Expected shared-upload queue to be empty"); flb_stop(ctx); flb_destroy(ctx); + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); + unsetenv("TEST_UPLOAD_PART_ERROR"); unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); unsetenv("TEST_UploadPart_CALL_COUNT"); unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); @@ -1008,7 +1547,7 @@ void flb_test_s3_default_retry_limit(void) /* Wait for the initial attempt and all five default retries. */ flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); - wait_for_s3_call_count("PutObject", 6); + wait_for_s3_call_count_with_timeout("PutObject", 6, 40000); flb_stop(ctx); flb_destroy(ctx); @@ -1089,6 +1628,183 @@ void flb_test_s3_default_retry_exhausted_action_quarantine(void) flb_free(store_dir); } +void flb_test_s3_ordered_index_keeps_global_queue_order(void) +{ + int ret; + int index; + int out_ffd; + int input_fds[2]; + char *uri; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + struct s3_file *oldest_file; + struct s3_file *later_file; + + store_dir = create_test_store_directory("/flb-s3-test-index-order-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_RECORD_S3_URIS", "true", 1); + setenv("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); + + ctx = flb_create(); + for (index = 0; index < 2; index++) { + input_fds[index] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(input_fds[index] >= 0); + } + flb_input_set(ctx, input_fds[0], "tag", "oldest", NULL); + flb_input_set(ctx, input_fds[1], "tag", "later", 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", "10s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "s3_key_format", "/queue/$TAG/$INDEX", NULL); + flb_output_set(ctx, out_ffd, "static_file_path", "true", NULL); + flb_output_set(ctx, out_ffd, "retry_limit", "1", NULL); + flb_output_set(ctx, out_ffd, "retry_exhausted_action", "delete", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + for (index = 0; index < 2; index++) { + ret = flb_lib_push(ctx, input_fds[index], (char *) JSON_TD, + (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + } + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + wait_for_file_count(s3_ctx->stream_active->path, 2); + oldest_file = s3_store_file_get(s3_ctx, "oldest", 6); + later_file = s3_store_file_get(s3_ctx, "later", 5); + TEST_CHECK(oldest_file != NULL); + TEST_CHECK(later_file != NULL); + oldest_file->create_time = time(NULL) - 30; + later_file->create_time = time(NULL) - 20; + + setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + wait_for_s3_call_count("PutObject", 1); + flb_time_msleep(500); + + TEST_CHECK_(get_s3_call_count("PutObject") == 1, + "Expected $INDEX to preserve global stop-at-head ordering, got %d calls", + get_s3_call_count("PutObject")); + uri = getenv("TEST_PutObject_URI_1"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest/0") == 0, + "Expected oldest indexed chunk first, got %s", + uri ? uri : "(null)"); + + unsetenv("TEST_PUT_OBJECT_ERROR"); + wait_for_s3_call_count("PutObject", 3); + uri = getenv("TEST_PutObject_URI_2"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest/0") == 0, + "Expected failed indexed chunk to retry before later tag, got %s", + uri ? uri : "(null)"); + uri = getenv("TEST_PutObject_URI_3"); + TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/later/1") == 0, + "Expected later indexed chunk after retry, got %s", + uri ? uri : "(null)"); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_PUT_OBJECT_ERROR"); + unsetenv("TEST_PUT_OBJECT_ERROR_TAG"); + unsetenv("TEST_RECORD_S3_URIS"); + unsetenv("TEST_PutObject_CALL_COUNT"); + unsetenv("TEST_PutObject_URI_1"); + unsetenv("TEST_PutObject_URI_2"); + unsetenv("TEST_PutObject_URI_3"); + flb_free(store_dir); +} + +void flb_test_s3_empty_upload_queue_file_deleted(void) +{ + int ret; + int in_ffd; + int out_ffd; + int file_count; + char empty_payload = '\0'; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + struct s3_file *s3_file; + struct upload_queue *upload_contents; + + store_dir = create_test_store_directory("/flb-s3-test-empty-queue-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (store_dir == NULL) { + return; + } + + 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", "live", 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); + /* Keep the non-empty control chunk alive beyond the first queue timer tick. */ + flb_output_set(ctx, out_ffd, "upload_timeout", "12s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, 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, "empty", 5, &empty_payload, + 0, time(NULL)); + TEST_CHECK(ret == 0); + + s3_file = s3_store_file_get(s3_ctx, "empty", 5); + TEST_CHECK(s3_file != NULL); + s3_store_file_lock(s3_file); + + upload_contents = flb_calloc(1, sizeof(struct upload_queue)); + TEST_CHECK(upload_contents != NULL); + upload_contents->upload_file = s3_file; + upload_contents->tag = flb_sds_create("empty"); + TEST_CHECK(upload_contents->tag != NULL); + upload_contents->tag_len = 5; + upload_contents->upload_time = -1; + mk_list_add(&upload_contents->_head, &s3_ctx->upload_queue); + + ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + wait_for_file_count(s3_ctx->stream_active->path, 2); + wait_for_file_count_at_most(s3_ctx->stream_active->path, 1); + + file_count = count_files_recursive(s3_ctx->stream_active->path); + TEST_CHECK_(file_count == 1, + "Expected only the non-empty live chunk to remain, got %d files", + file_count); + + 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_near_full_buffer_append_succeeds(void) { int ret; @@ -1159,13 +1875,157 @@ void flb_test_s3_near_full_buffer_append_succeeds(void) flb_free(store_dir); } +void flb_test_s3_startup_index_order_after_failure(void) +{ + int ret; + int index; + int in_ffd; + int out_ffd; + int call_count; + int input_fds[2]; + char tag[32]; + char *uri; + flb_ctx_t *ctx; + char *store_dir; + struct flb_s3 *s3_ctx; + + store_dir = create_test_store_directory("/flb-s3-test-startup-index-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); + setenv("TEST_RECORD_S3_URIS", "true", 1); + + ctx = flb_create(); + for (index = 0; index < 2; index++) { + input_fds[index] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(input_fds[index] >= 0); + snprintf(tag, sizeof(tag), "index-test-%d", index); + flb_input_set(ctx, input_fds[index], "tag", tag, 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, "s3_key_format", "/recovery/$TAG/$INDEX", NULL); + flb_output_set(ctx, out_ffd, "preserve_data_ordering", "false", 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); + for (index = 0; index < 2; index++) { + ret = flb_lib_push(ctx, input_fds[index], + (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + } + wait_for_file_count(s3_ctx->stream_active->path, 2); + + flb_stop(ctx); + flb_destroy(ctx); + flb_time_msleep(1100); + + unsetenv("TEST_PutObject_CALL_COUNT"); + unsetenv("TEST_PutObject_URI_1"); + + ctx = flb_create(); + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "live", 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, "s3_key_format", "/recovery/$TAG/$INDEX", NULL); + flb_output_set(ctx, out_ffd, "preserve_data_ordering", "false", 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); + + call_count = get_s3_call_count("PutObject"); + TEST_CHECK_(call_count == 1, + "Expected $INDEX recovery to stop after one failure, got %d attempts", + call_count); + uri = getenv("TEST_PutObject_URI_1"); + TEST_CHECK_(uri != NULL && + (strcmp(uri, "/fluent/recovery/index-test-0/0") == 0 || + strcmp(uri, "/fluent/recovery/index-test-1/0") == 0), + "Expected failed request to use an original chunk with index 0, got %s", + uri ? uri : "(null)"); + + s3_ctx = get_s3_context(ctx); + TEST_CHECK(s3_ctx != NULL); + TEST_CHECK_(s3_ctx->seq_index == 0, + "Expected failed request to roll index back to 0, got %" PRIu64, + s3_ctx->seq_index); + + unsetenv("TEST_PUT_OBJECT_ERROR"); + 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", 3); + + uri = getenv("TEST_PutObject_URI_2"); + TEST_CHECK_(uri != NULL && + getenv("TEST_PutObject_URI_1") != NULL && + strcmp(uri, getenv("TEST_PutObject_URI_1")) == 0, + "Expected failed chunk retry to keep its tag and index, got %s", + uri ? uri : "(null)"); + uri = getenv("TEST_PutObject_URI_3"); + TEST_CHECK_(uri != NULL && + (strcmp(uri, "/fluent/recovery/index-test-0/1") == 0 || + strcmp(uri, "/fluent/recovery/index-test-1/1") == 0), + "Expected the other original chunk to receive index 1, got %s", + uri ? uri : "(null)"); + TEST_CHECK_(s3_ctx->seq_index == 2, + "Expected two recovered uploads to advance index to 2, got %" PRIu64, + s3_ctx->seq_index); + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); + unsetenv("TEST_PUT_OBJECT_ERROR"); + unsetenv("TEST_RECORD_S3_URIS"); + unsetenv("TEST_PutObject_CALL_COUNT"); + unsetenv("TEST_PutObject_URI_1"); + unsetenv("TEST_PutObject_URI_2"); + unsetenv("TEST_PutObject_URI_3"); + unsetenv("TEST_PutObject_URI_4"); + flb_free(store_dir); +} + void flb_test_s3_startup_buffer_size_accounting(void) { int ret; + int index; int in_ffd; int out_ffd; int call_count; + int sized_down_file_count; int file_count; + int input_fds[S3_TEST_STARTUP_FILE_COUNT]; + char tag[32]; + char old_stream_dir[2048]; uint64_t live_buffer_size; uint64_t restored_buffer_size; flb_ctx_t *ctx; @@ -1182,9 +2042,12 @@ void flb_test_s3_startup_buffer_size_accounting(void) 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); + for (index = 0; index < S3_TEST_STARTUP_FILE_COUNT; index++) { + input_fds[index] = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(input_fds[index] >= 0); + snprintf(tag, sizeof(tag), "startup-test-%d", index); + flb_input_set(ctx, input_fds[index], "tag", tag, NULL); + } out_ffd = flb_output(ctx, (char *) "s3", NULL); TEST_CHECK(out_ffd >= 0); @@ -1202,12 +2065,17 @@ void flb_test_s3_startup_buffer_size_accounting(void) 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); + for (index = 0; index < S3_TEST_STARTUP_FILE_COUNT; index++) { + ret = flb_lib_push(ctx, input_fds[index], + (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + TEST_CHECK(ret >= 0); + } + wait_for_s3_call_count("PutObject", S3_TEST_STARTUP_FILE_COUNT); s3_ctx = get_s3_context(ctx); TEST_CHECK(s3_ctx != NULL); + snprintf(old_stream_dir, sizeof(old_stream_dir), "%s", + s3_ctx->stream_active->path); 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"); @@ -1216,9 +2084,12 @@ void flb_test_s3_startup_buffer_size_accounting(void) 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); + TEST_CHECK_(file_count >= S3_TEST_STARTUP_FILE_COUNT, + "Expected at least %d buffered files to survive the first run, got %d", + S3_TEST_STARTUP_FILE_COUNT, file_count); + + /* Ensure the next process gets a distinct active stream directory. */ + flb_time_msleep(1100); unsetenv("TEST_PutObject_CALL_COUNT"); @@ -1243,8 +2114,17 @@ void flb_test_s3_startup_buffer_size_accounting(void) ret = flb_start(ctx); TEST_CHECK(ret == 0); + call_count = get_s3_call_count("PutObject"); + TEST_CHECK_(call_count == 1, + "Expected ordered startup drain to stop after one failure, got %d attempts", + call_count); + s3_ctx = get_s3_context(ctx); TEST_CHECK(s3_ctx != NULL); + sized_down_file_count = count_sized_down_s3_files(s3_ctx); + TEST_CHECK_(sized_down_file_count >= 2, + "Expected at least two restored down chunks with accounted bytes, got %d", + sized_down_file_count); 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, @@ -1276,16 +2156,20 @@ void flb_test_s3_startup_buffer_size_accounting(void) ret = flb_start(ctx); TEST_CHECK(ret == 0); - wait_for_s3_call_count("PutObject", 1); + wait_for_s3_call_count("PutObject", S3_TEST_STARTUP_FILE_COUNT); + + TEST_CHECK_(test_directory_exists(old_stream_dir) == 0, + "Expected drained startup stream directory to be removed: %s", + old_stream_dir); 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); + wait_for_s3_call_count("PutObject", S3_TEST_STARTUP_FILE_COUNT + 1); call_count = get_s3_call_count("PutObject"); - TEST_CHECK_(call_count == 2, - "Expected startup resend and new upload, got %d PutObject calls", - call_count); + TEST_CHECK_(call_count == S3_TEST_STARTUP_FILE_COUNT + 1, + "Expected %d startup resends and one new upload, got %d PutObject calls", + S3_TEST_STARTUP_FILE_COUNT, call_count); flb_stop(ctx); flb_destroy(ctx); @@ -1304,11 +2188,19 @@ 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 }, + {"empty_upload_queue_file_deleted", flb_test_s3_empty_upload_queue_file_deleted }, {"near_full_buffer_append_succeeds", flb_test_s3_near_full_buffer_append_succeeds }, + {"startup_index_order_after_failure", flb_test_s3_startup_index_order_after_failure }, {"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 }, + {"ordered_retry_uses_backoff_deadline", flb_test_s3_ordered_retry_uses_backoff_deadline }, + {"ordered_timer_isolates_tag_backoff", flb_test_s3_ordered_timer_isolates_tag_backoff }, + {"ordered_index_keeps_global_queue_order", flb_test_s3_ordered_index_keeps_global_queue_order }, + {"ordered_construct_error_exhausts_chunk", flb_test_s3_ordered_construct_error_exhausts_chunk }, + {"ordered_backoff_does_not_starve_completion", flb_test_s3_ordered_backoff_does_not_starve_completion }, + {"ordered_shared_upload_retries_safely", flb_test_s3_ordered_shared_upload_retries_safely }, {"compression_gzip", flb_test_s3_compression_gzip }, {"compression_gzip_putobject", flb_test_s3_compression_gzip_putobject }, {"compression_zstd", flb_test_s3_compression_zstd },