diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h new file mode 100644 index 00000000000..3d888fdc7c2 --- /dev/null +++ b/include/fluent-bit/flb_async_timer.h @@ -0,0 +1,197 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FLB_ASYNC_TIMER_H +#define FLB_ASYNC_TIMER_H + +#ifndef _XOPEN_SOURCE +#define _XOPEN_SOURCE +#endif + +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include +#include +#include +#include + + +void flb_async_timer_destroy(struct flb_out_async_timer *timer); +void flb_async_timer_cleanup(struct mk_list *list); +void flb_output_async_timer_cleanup(struct flb_config *config); +int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int ms, + struct flb_output_instance *o_ins, + char *job_name, + void (*async_cb)(struct flb_config *, void *), + void *data, struct flb_sched_timer **out_timer); +void flb_out_async_timers_print(struct flb_output_instance *ins); +void flb_async_timers_print(struct mk_list *async_timer_list); +int flb_async_timers_size(struct flb_output_instance *ins); +int flb_thread_pool_async_timers_size(struct flb_output_instance *ins); +void flb_thread_pool_async_timers_print(struct flb_output_instance *ins); + +/* + * stores timer coros on the async_timer_list, if the output uses them + */ +struct flb_out_async_timer { + struct flb_config *config; /* FLB context */ + struct flb_output_instance *o_ins; /* output instance */ + struct flb_out_async_timer_cb_data *timer_data; /* callback info */ + struct flb_coro *coro; /* parent coro addr */ + struct mk_list _head; /* Link to async_timer_list */ +}; + +/* + * If the output uses timer coros, then this is used as the callback data + * passed to flb_sched_timer_cb_create + */ +struct flb_out_async_timer_cb_data { + struct flb_output_instance *ins; /* associate coro with this output instance */ + char *job_name; /* used on engine shutdown, print pending "custom" jobs */ + void (*async_cb) (struct flb_config *config, void *data); /* call this output callback in the coro */ + void *data; /* opaque data to pass to the above cb */ +}; + +extern FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); + + +/* The coro callee callback for async timer coros */ +static FLB_INLINE void out_async_timer_cb(void) +{ + struct flb_coro *coro; + struct flb_output_instance *o_ins; + struct flb_out_thread_instance *th_ins; + struct flb_out_async_timer *async_timer; + + + async_timer = (struct flb_out_async_timer *) FLB_TLS_GET(async_timer_coro_params); + if (!async_timer) { + flb_error("[output] no async timer coro params defined, unexpected"); + return; + } + + coro = async_timer->coro; + o_ins = async_timer->o_ins; + + /* Run the callback provided by the output plugin */ + async_timer->timer_data->async_cb(async_timer->config, async_timer->timer_data->data); + + /* move coro to destroy queue */ + if (flb_output_is_threaded(o_ins) == FLB_TRUE) { + th_ins = flb_output_thread_instance_get(); + pthread_mutex_lock(&th_ins->async_timer_mutex); + mk_list_del(&async_timer->_head); + mk_list_add(&async_timer->_head, &th_ins->async_timer_list_destroy); + pthread_mutex_unlock(&th_ins->async_timer_mutex); + } + else { + mk_list_del(&async_timer->_head); + mk_list_add(&async_timer->_head, &o_ins->async_timer_list_destroy); + } + + /* timer coro is complete; yield back to caller/control code */ + flb_coro_yield(coro, FLB_TRUE); +} + +/* + * If the output uses scheduled timers with coroutines, + * this function is used as the callback for flb_sched_timer_cb_create + */ +static FLB_INLINE +void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) +{ + size_t stack_size; + struct flb_coro *coro; + struct flb_out_async_timer *async_timer; + struct flb_out_async_timer *timer_thread_key; + struct flb_out_thread_instance *th_ins; + struct flb_out_async_timer_cb_data *ctx = (struct flb_out_async_timer_cb_data *) data; + struct flb_output_instance *o_ins; + + /* Custom output coroutine info */ + async_timer = (struct flb_out_async_timer*) flb_calloc(1, sizeof(struct flb_out_async_timer)); + if (!async_timer) { + flb_errno(); + return; + } + + /* Create a new co-routine */ + coro = flb_coro_create(async_timer); + if (!coro) { + flb_free(async_timer); + return; + } + + o_ins = ctx->ins; + async_timer->o_ins = o_ins; + async_timer->config = config; + async_timer->coro = coro; + + coro->callee = co_create(config->coro_stack_size, + out_async_timer_cb, &stack_size); + + if (coro->callee == NULL) { + flb_coro_destroy(coro); + flb_free(async_timer); + return; + } + +#ifdef FLB_HAVE_VALGRIND + coro->valgrind_stack_id = \ + VALGRIND_STACK_REGISTER(coro->callee, ((char *) coro->callee) + stack_size); +#endif + + if (o_ins->is_threaded == FLB_TRUE) { + th_ins = flb_output_thread_instance_get(); + pthread_mutex_lock(&th_ins->async_timer_mutex); + mk_list_add(&async_timer->_head, &th_ins->async_timer_list); + pthread_mutex_unlock(&th_ins->async_timer_mutex); + } + else { + mk_list_add(&async_timer->_head, &o_ins->async_timer_list); + } + + /* Same struct used in async_timer_list and in the pthread key, + * Unique memory needed since these are freed separately + */ + timer_thread_key = (struct flb_out_async_timer *) FLB_TLS_GET(async_timer_coro_params); + if (!timer_thread_key) { + timer_thread_key = (struct flb_out_async_timer *) flb_calloc(1, sizeof(struct flb_out_async_timer)); + if (!timer_thread_key) { + flb_errno(); + return; + } + } + + /* copy to thread local storage */ + timer_thread_key->coro = coro; + timer_thread_key->o_ins = o_ins; + timer_thread_key->timer_data = async_timer->timer_data; + + FLB_TLS_SET(async_timer_coro_params, timer_thread_key); + coro->caller = co_active(); + flb_coro_resume(coro); + return; +} + + + +#endif diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index c9172b65d17..b77f5595c0b 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -43,6 +43,7 @@ #include #include #include +#include #include #include #include @@ -408,6 +409,10 @@ struct flb_output_instance { struct mk_list flush_list; struct mk_list flush_list_destroy; + /* similar to flush coroutine list above, timer coroutine list */ + struct mk_list async_timer_list; + struct mk_list async_timer_list_destroy; + /* Keep a reference to the original context this instance belongs to */ struct flb_config *config; }; @@ -460,6 +465,7 @@ struct flb_out_flush_params { }; extern FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); +extern FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); static FLB_INLINE void output_params_set(struct flb_output_flush *out_flush, struct flb_coro *coro, @@ -659,7 +665,11 @@ static inline void flb_output_return(int ret, struct flb_coro *co) { flb_output_flush_prepare_destroy(out_flush); } -/* return the number of co-routines running in the instance */ +/* + * return the number of flush co-routines running in the instance + * Currently, this function is only used for FLB_OUTPUT_NO_MULTIPLEX + * and does not count timer_coros, used by S3 output + */ static inline int flb_output_coros_size(struct flb_output_instance *ins) { int size = 0; diff --git a/include/fluent-bit/flb_output_thread.h b/include/fluent-bit/flb_output_thread.h index 4100c9c1a7e..407a97187f3 100644 --- a/include/fluent-bit/flb_output_thread.h +++ b/include/fluent-bit/flb_output_thread.h @@ -86,7 +86,12 @@ struct flb_out_thread_instance { * 'flushes' running by a threaded instance, then the access to the 'flush_list' * must be protected: we use 'flush_mutex for that purpose. */ - pthread_mutex_t flush_mutex; /* mutex for 'flush_list' */ + pthread_mutex_t flush_mutex; /* mutex for 'flush_list' */ + + /* Same as flush_mutex but for timer coros */ + struct mk_list async_timer_list; /* timer list */ + struct mk_list async_timer_list_destroy; /* timer destroy list */ + pthread_mutex_t async_timer_mutex; /* mutex for async timer lists */ /* List of mapped 'upstream' contexts */ struct mk_list upstreams; @@ -100,7 +105,8 @@ int flb_output_thread_pool_start(struct flb_output_instance *ins); int flb_output_thread_pool_flush(struct flb_task *task, struct flb_output_instance *out_ins, struct flb_config *config); - +int flb_thread_pool_async_timers_size(struct flb_output_instance *ins); +void flb_thread_pool_async_timers_print(struct flb_output_instance *ins); void flb_output_thread_instance_init(); struct flb_out_thread_instance *flb_output_thread_instance_get(); diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index b4556158000..fabb39d854d 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -18,9 +18,12 @@ */ #include +#include +#include #include #include #include +#include #include #include #include @@ -29,6 +32,7 @@ #include #include #include +#include #include #include @@ -45,10 +49,10 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, struct s3_file *chunk, char **out_buf, size_t *out_size); -static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t create_time, +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 is_startup); static void cb_s3_upload(struct flb_config *ctx, void *data); @@ -56,9 +60,10 @@ static struct multipart_upload *get_upload(struct flb_s3 *ctx, const char *tag, int tag_len); static struct multipart_upload *create_upload(struct flb_s3 *ctx, - const char *tag, int tag_len); + const char *tag, int tag_len, + time_t file_first_log_time, + char *input_name); -static void remove_from_queue(struct upload_queue *entry); static struct flb_aws_header content_encoding_header = { .key = "Content-Encoding", @@ -95,6 +100,45 @@ static struct flb_aws_header storage_class_header = { .val_len = 0, }; +static void s3_retry_warn(struct flb_s3 *ctx, const char *tag, + char *input_name, time_t create_time, + int less_than_limit) +{ + struct tm now_time; + char create_time_str[20]; + struct tm *tmp; + + tmp = localtime_r(&create_time, &now_time); + strftime(create_time_str, 20, "%Y-%m-%d %H:%M:%S", tmp); + if (input_name == NULL || strlen(input_name) == 0) { + if (less_than_limit == FLB_TRUE) { + flb_plg_warn(ctx->ins, + "failed to flush chunk tag=%s, create_time=%s, " + "retry issued: (out_id=%d)", + tag, create_time_str, ctx->ins->id); + } + else { + flb_plg_warn(ctx->ins, + "chunk tag=%s, create_time=%s cannot be retried", + tag, create_time_str); + } + } + else { + if (less_than_limit == FLB_TRUE) { + flb_plg_warn(ctx->ins, + "failed to flush chunk tag=%s, create_time=%s, " + "retry issued: input=%s > output=%s (out_id=%d)", + tag, create_time_str, input_name, ctx->ins->name, ctx->ins->id); + } + else { + flb_plg_warn(ctx->ins, + "chunk tag=%s, create_time=%s cannot be retried: " + "input=%s > output=%s", + tag, create_time_str, input_name, ctx->ins->name); + } + } +} + static char *mock_error_response(char *error_env_var) { char *err_val = NULL; @@ -263,6 +307,13 @@ static flb_sds_t concat_path(char *p1, char *p2) { flb_sds_t dir; flb_sds_t tmp; + int len = 0; + + /* remove trailing slash from p1 if needed */ + len = strlen(p1); + if (p1[len - 1] == '/') { + p1[len - 1] = '\0'; + } dir = flb_sds_create_size(64); @@ -321,6 +372,18 @@ static int write_seq_index(char *seq_index_file, uint64_t seq_index) return 0; } +static void s3_decrement_index(struct flb_s3 *ctx) +{ + int ret; + ctx->seq_index--; + + ret = write_seq_index(ctx->seq_index_file, ctx->seq_index); + if (ret < 0) { + flb_plg_error(ctx->ins, "Failed to save decremented $INDEX for s3 key to " + "store_dir after request error"); + } +} + static int init_seq_index(void *context) { int ret; const char *tmp; @@ -354,14 +417,14 @@ static int init_seq_index(void *context) { ctx->seq_index_file = flb_sds_create(ctx->metadata_dir); if (ctx->seq_index_file == NULL) { - flb_plg_error(ctx->ins, "Failed to create sequential index file path"); + flb_plg_error(ctx->ins, "Failed to create $INDEX file path"); flb_errno(); return -1; } tmp = "/seq_index_"; ret = flb_sds_cat_safe(&ctx->seq_index_file, tmp, strlen(tmp)); if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to create sequential index file path"); + flb_plg_error(ctx->ins, "Failed to concat $INDEX file path"); flb_errno(); return -1; } @@ -369,7 +432,7 @@ static int init_seq_index(void *context) { sprintf(tmp_buf, "%d", ctx->ins->id); ret = flb_sds_cat_safe(&ctx->seq_index_file, tmp_buf, strlen(tmp_buf)); if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to create sequential index file path"); + flb_plg_error(ctx->ins, "Failed to concat output ID to $INDEX file path"); flb_errno(); return -1; } @@ -393,11 +456,11 @@ static int init_seq_index(void *context) { else { ret = read_seq_index(ctx->seq_index_file, &ctx->seq_index); if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to read from sequential index " + flb_plg_error(ctx->ins, "Failed to read from $INDEX " "metadata file"); return -1; } - flb_plg_info(ctx->ins, "Successfully recovered index. " + flb_plg_info(ctx->ins, "Successfully recovered existing $INDEX in store_dir. " "Continuing at index=%d", ctx->seq_index); } return 0; @@ -437,8 +500,6 @@ static void s3_context_destroy(struct flb_s3 *ctx) struct mk_list *head; struct mk_list *tmp; struct multipart_upload *m_upload; - struct upload_queue *upload_contents; - if (!ctx) { return; } @@ -490,13 +551,6 @@ static void s3_context_destroy(struct flb_s3 *ctx) multipart_upload_destroy(m_upload); } - 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); - multipart_upload_destroy(upload_contents->m_upload_file); - remove_from_queue(upload_contents); - } - flb_free(ctx); } @@ -527,8 +581,8 @@ static int cb_s3_init(struct flb_output_instance *ins, mk_list_init(&ctx->uploads); mk_list_init(&ctx->upload_queue); - ctx->retry_time = 0; - ctx->upload_queue_success = FLB_FALSE; + pthread_mutex_init(&ctx->upload_queue_mutex, NULL); + pthread_mutex_init(&ctx->cb_flush_mutex, NULL); /* Export context */ flb_output_set_context(ins, ctx); @@ -924,7 +978,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, @@ -949,16 +1003,9 @@ static int cb_s3_init(struct flb_output_instance *ins, cb_s3_upload(config, ctx); } - if (ctx->use_put_object == FLB_TRUE) { - /* - * Run S3 in async mode. - * Multipart uploads don't work with async mode right now in high throughput - * cases. Its not clear why. Realistically, the performance of sync mode - * will be sufficient for most users, and long term we can do the work - * to enable async if needed. - */ - ctx->s3_client->upstream->flags = async_flags; - } + + /* S3 can run in async mode with timer coroutines */ + ctx->s3_client->upstream->flags = async_flags; /* this is done last since in the previous block we make calls to AWS */ ctx->provider->provider_vtable->upstream_set(ctx->provider, ctx->ins); @@ -967,7 +1014,10 @@ static int cb_s3_init(struct flb_output_instance *ins, } /* - * return value is one of FLB_OK, FLB_RETRY, FLB_ERROR + * return value is one of 0, -1, -2. + * 0 means uploaded successfully. + * -1 means failed to upload data, will retry. + * -2 means failed to upload data, but can't retry because reach the retry_limit. * * Chunk is allowed to be NULL */ @@ -981,18 +1031,30 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, int size_check = FLB_FALSE; int part_num_check = FLB_FALSE; int timeout_check = FLB_FALSE; - time_t create_time; int ret; void *payload_buf = NULL; size_t payload_size = 0; size_t preCompress_size = 0; + time_t file_first_log_time = time(NULL); + char* input_name = NULL; + + /* + * When chunk does not exist, file_first_log_time will be the current time. + * This is only for unit tests and prevents unit tests from segfaulting when chunk is + * NULL because if so chunk->first_log_time will be NULl either and will cause + * segfault during the process of put_object upload or mutipart upload. + */ + if (chunk != NULL) { + file_first_log_time = chunk->first_log_time; + input_name = chunk->input_name; + } if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { /* Map payload */ ret = flb_aws_compression_compress(ctx->compression, body, body_size, &payload_buf, &payload_size); if (ret == -1) { flb_plg_error(ctx->ins, "Failed to compress data"); - return FLB_RETRY; + return -1; } else { preCompress_size = body_size; body = (void *) payload_buf; @@ -1017,7 +1079,7 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, if (m_upload == NULL) { if (chunk != NULL && time(NULL) > - (chunk->create_time + ctx->upload_timeout + ctx->retry_time)) { + (chunk->create_time + ctx->upload_timeout)) { /* timeout already reached, just PutObject */ goto put_object; } @@ -1048,49 +1110,37 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, put_object: - /* - * remove chunk from buffer list- needed for async http so that the - * same chunk won't be sent more than once - */ - if (chunk) { - create_time = chunk->create_time; - } - else { - create_time = time(NULL); - } - - ret = s3_put_object(ctx, tag, create_time, body, body_size); + ret = s3_put_object(ctx, tag, file_first_log_time, body, body_size); if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { flb_free(payload_buf); } if (ret < 0) { - /* re-add chunk to list */ if (chunk) { - s3_store_file_unlock(chunk); chunk->failures += 1; + if (ctx->ins->retry_limit >= 0 && chunk->failures > ctx->ins->retry_limit){ + s3_retry_warn(ctx, tag, input_name, file_first_log_time, FLB_FALSE); + return -2; + } + else { + s3_retry_warn(ctx, tag, input_name, file_first_log_time, FLB_TRUE); + return -1; + } } - return FLB_RETRY; + return -1; } - /* data was sent successfully- delete the local buffer */ - if (chunk) { - s3_store_file_delete(ctx, chunk); - } - return FLB_OK; + return 0; multipart: if (init_upload == FLB_TRUE) { - m_upload = create_upload(ctx, tag, tag_len); + m_upload = create_upload(ctx, tag, tag_len, file_first_log_time, input_name); if (!m_upload) { flb_plg_error(ctx->ins, "Could not find or create upload for tag %s", tag); - if (chunk) { - s3_store_file_unlock(chunk); - } if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { flb_free(payload_buf); } - return FLB_RETRY; + return -1; } } @@ -1098,13 +1148,10 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, ret = create_multipart_upload(ctx, m_upload); if (ret < 0) { flb_plg_error(ctx->ins, "Could not initiate multipart upload"); - if (chunk) { - s3_store_file_unlock(chunk); - } if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { flb_free(payload_buf); } - return FLB_RETRY; + return -1; } m_upload->upload_state = MULTIPART_UPLOAD_STATE_CREATED; } @@ -1114,20 +1161,40 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { flb_free(payload_buf); } - m_upload->upload_errors += 1; - /* re-add chunk to list */ if (chunk) { - s3_store_file_unlock(chunk); chunk->failures += 1; + if (ctx->ins->retry_limit >= 0 && chunk->failures > ctx->ins->retry_limit) { + s3_retry_warn(ctx, (char *) chunk->fsf->meta_buf, m_upload->input_name, + chunk->create_time, FLB_FALSE); + /* + * part_number initializes with 1, if the number still is 1 which means + * no data is uploaded and this upload file can be deleted , else set + * as complete. + */ + if (m_upload->part_number == 1) { + mk_list_del(&m_upload->_head); + multipart_upload_destroy(m_upload); + + /* Decrement the $INDEX when no data is uploaded */ + if (ctx->key_fmt_has_seq_index) { + s3_decrement_index(ctx); + } + } + else { + m_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS; + } + return -2; + } + else { + s3_retry_warn(ctx, (char *) chunk->fsf->meta_buf, m_upload->input_name, + chunk->create_time, FLB_TRUE); + return -1; + } } - return FLB_RETRY; + return -1; } + m_upload->part_number += 1; - /* data was sent successfully- delete the local buffer */ - if (chunk) { - s3_store_file_delete(ctx, chunk); - chunk = NULL; - } if (ctx->compression == FLB_AWS_COMPRESS_GZIP) { flb_free(payload_buf); } @@ -1142,7 +1209,7 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, "(the API limit) have been uploaded", m_upload->s3_key); } if (time(NULL) > - (m_upload->init_time + ctx->upload_timeout + ctx->retry_time)) { + (m_upload->init_time + ctx->upload_timeout)) { timeout_check = FLB_TRUE; flb_plg_info(ctx->ins, "Will complete upload for %s because upload_timeout" " has elapsed", m_upload->s3_key); @@ -1156,7 +1223,7 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, m_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS; } - return FLB_OK; + return 0; } @@ -1165,7 +1232,7 @@ 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 is_startup) { struct s3_file *chunk; struct mk_list *tmp; @@ -1185,29 +1252,21 @@ static int put_all_chunks(struct flb_s3 *ctx) if (fs_stream == ctx->stream_upload) { continue; } + /* skip metadata stream */ if (fs_stream == ctx->stream_metadata) { continue; } + /* on startup, we only send old chunks in this routine */ + if (is_startup == FLB_TRUE && fs_stream == ctx->stream_active) { + continue; + } + mk_list_foreach_safe(f_head, tmp, &fs_stream->files) { fsf = mk_list_entry(f_head, struct flb_fstore_file, _head); chunk = fsf->data; - /* Locked chunks are being processed, skip */ - if (chunk->locked == FLB_TRUE) { - continue; - } - - if (chunk->failures >= MAX_UPLOAD_ERRORS) { - flb_plg_warn(ctx->ins, - "Chunk for tag %s failed to send %i times, " - "will not retry", - (char *) fsf->meta_buf, MAX_UPLOAD_ERRORS); - flb_fstore_file_inactive(ctx->fs, fsf); - continue; - } - ret = construct_request_buffer(ctx, NULL, chunk, &buffer, &buffer_size); if (ret < 0) { @@ -1228,18 +1287,47 @@ static int put_all_chunks(struct flb_s3 *ctx) buffer_size = payload_size; } } - ret = s3_put_object(ctx, (const char *) fsf->meta_buf, chunk->create_time, buffer, buffer_size); flb_free(buffer); if (ret < 0) { - s3_store_file_unlock(chunk); chunk->failures += 1; + if (is_startup == FLB_TRUE) { + if (ctx->ins->retry_limit >= 0 && chunk->failures > ctx->ins->retry_limit){ + s3_retry_warn(ctx, (char *) fsf->meta_buf, NULL, + chunk->create_time, FLB_FALSE); + if (chunk->locked == FLB_TRUE) { + /* remove from upload_queue */ + if (chunk->_head.next != NULL && chunk->_head.prev != NULL) { + mk_list_del(&chunk->_head); + } + } + s3_store_file_delete(ctx, chunk); + return -1; + } + else { + s3_retry_warn(ctx, (char *) fsf->meta_buf, NULL, + chunk->create_time, FLB_TRUE); + return -1; + } + } + else { + flb_plg_error(ctx->ins, "Failed to flush chunk tag=%s, " + "chunk will remain buffered on the filesystem " + "in the store_dir.", + (char *) fsf->meta_buf); + } return -1; } /* data was sent successfully- delete the local buffer */ + if (chunk->locked == FLB_TRUE) { + /* remove from upload_queue */ + if (chunk->_head.next != NULL && chunk->_head.prev != NULL) { + mk_list_del(&chunk->_head); + } + } s3_store_file_delete(ctx, chunk); } } @@ -1275,11 +1363,6 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, return -1; } - /* - * lock the chunk from buffer list- needed for async http so that the - * same chunk won't be sent more than once. - */ - s3_store_file_lock(chunk); body = buffered_data; body_size = buffer_size; } @@ -1295,9 +1378,6 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, if (!tmp) { flb_errno(); flb_free(buffered_data); - if (chunk) { - s3_store_file_unlock(chunk); - } return -1; } body = buffered_data = tmp; @@ -1311,7 +1391,7 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, return 0; } -static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t create_time, +static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t file_first_log_time, char *body, size_t body_size) { flb_sds_t s3_key = NULL; @@ -1328,8 +1408,8 @@ static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t create_time flb_sds_t tmp; char final_body_md5[25]; - s3_key = flb_get_s3_key(ctx->s3_key_format, create_time, tag, ctx->tag_delimiters, - ctx->seq_index); + s3_key = flb_get_s3_key(ctx->s3_key_format, file_first_log_time, tag, + ctx->tag_delimiters, ctx->seq_index); if (!s3_key) { flb_plg_error(ctx->ins, "Failed to construct S3 Object Key for %s", tag); return -1; @@ -1391,7 +1471,7 @@ static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t create_time if (ret < 0 && access(ctx->seq_index_file, F_OK) == 0) { ctx->seq_index--; flb_sds_destroy(s3_key); - flb_plg_error(ctx->ins, "Failed to update sequential index metadata file"); + flb_plg_error(ctx->ins, "Failed to update $INDEX tracking file"); return -1; } } @@ -1440,13 +1520,7 @@ static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t create_time decrement_index: if (ctx->key_fmt_has_seq_index) { - ctx->seq_index--; - - ret = write_seq_index(ctx->seq_index_file, ctx->seq_index); - if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to decrement index after request error"); - return -1; - } + s3_decrement_index(ctx); } return -1; } @@ -1485,12 +1559,6 @@ static struct multipart_upload *get_upload(struct flb_s3 *ctx, if (tmp_upload->upload_state == MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS) { continue; } - if (tmp_upload->upload_errors >= MAX_UPLOAD_ERRORS) { - tmp_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS; - flb_plg_error(ctx->ins, "Upload for %s has reached max upload errors", - tmp_upload->s3_key); - continue; - } if (strcmp(tmp_upload->tag, tag) == 0) { m_upload = tmp_upload; break; @@ -1500,8 +1568,9 @@ static struct multipart_upload *get_upload(struct flb_s3 *ctx, return m_upload; } -static struct multipart_upload *create_upload(struct flb_s3 *ctx, - const char *tag, int tag_len) +static struct multipart_upload *create_upload(struct flb_s3 *ctx, const char *tag, + int tag_len, time_t file_first_log_time, + char *input_name) { int ret; struct multipart_upload *m_upload = NULL; @@ -1514,8 +1583,8 @@ static struct multipart_upload *create_upload(struct flb_s3 *ctx, flb_errno(); return NULL; } - s3_key = flb_get_s3_key(ctx->s3_key_format, time(NULL), tag, ctx->tag_delimiters, - ctx->seq_index); + s3_key = flb_get_s3_key(ctx->s3_key_format, file_first_log_time, tag, + ctx->tag_delimiters, ctx->seq_index); if (!s3_key) { flb_plg_error(ctx->ins, "Failed to construct S3 Object Key for %s", tag); flb_free(m_upload); @@ -1532,6 +1601,9 @@ static struct multipart_upload *create_upload(struct flb_s3 *ctx, m_upload->upload_state = MULTIPART_UPLOAD_STATE_NOT_CREATED; m_upload->part_number = 1; m_upload->init_time = time(NULL); + if (input_name != NULL) { + m_upload->input_name = input_name; + } mk_list_add(&m_upload->_head, &ctx->uploads); /* Update file and increment index value right before request */ @@ -1542,7 +1614,7 @@ static struct multipart_upload *create_upload(struct flb_s3 *ctx, if (ret < 0) { ctx->seq_index--; flb_sds_destroy(s3_key); - flb_plg_error(ctx->ins, "Failed to write to sequential index metadata file"); + flb_plg_error(ctx->ins, "Failed to write to $INDEX tracking file"); return NULL; } } @@ -1550,117 +1622,17 @@ static struct multipart_upload *create_upload(struct flb_s3 *ctx, return m_upload; } -/* Adds an entry to upload queue */ -static int add_to_queue(struct flb_s3 *ctx, struct s3_file *upload_file, - struct multipart_upload *m_upload_file, const char *tag, int tag_len) -{ - struct upload_queue *upload_contents; - flb_sds_t tag_cpy; - - /* Create upload contents object and add to upload queue */ - upload_contents = flb_calloc(1, sizeof(struct upload_queue)); - if (upload_contents == NULL) { - flb_plg_error(ctx->ins, "Error allocating memory for upload_queue entry"); - flb_errno(); - return -1; - } - upload_contents->upload_file = upload_file; - upload_contents->m_upload_file = m_upload_file; - upload_contents->tag_len = tag_len; - upload_contents->retry_counter = 0; - upload_contents->upload_time = -1; - - /* Necessary to create separate string for tag to prevent corruption */ - tag_cpy = flb_sds_create_len(tag, tag_len); - if (!tag_cpy) { - flb_errno(); - flb_free(upload_contents); - return -1; - } - upload_contents->tag = tag_cpy; - - - /* Add entry to upload queue */ - mk_list_add(&upload_contents->_head, &ctx->upload_queue); - return 0; -} - -/* Removes an entry from upload_queue */ -void remove_from_queue(struct upload_queue *entry) -{ - mk_list_del(&entry->_head); - flb_sds_destroy(entry->tag); - flb_free(entry); - return; -} - -/* Validity check for upload queue object */ -static int upload_queue_valid(struct upload_queue *upload_contents, time_t now, - void *out_context) -{ - struct flb_s3 *ctx = out_context; - - if (upload_contents == NULL) { - flb_plg_error(ctx->ins, "Error getting entry from upload_queue"); - return -1; - } - if (upload_contents->_head.next == NULL || upload_contents->_head.prev == NULL) { - flb_plg_debug(ctx->ins, "Encountered previously deleted entry in " - "upload_queue. Deleting invalid entry"); - mk_list_del(&upload_contents->_head); - return -1; - } - if (upload_contents->upload_file->locked == FLB_FALSE) { - flb_plg_debug(ctx->ins, "Encountered unlocked file in upload_queue. " - "Exiting"); - return -1; - } - if (upload_contents->upload_file->size <= 0) { - flb_plg_debug(ctx->ins, "Encountered empty chunk file in upload_queue. " - "Deleting empty chunk 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"); - return -1; - } - return 0; -} - -static int send_upload_request(void *out_context, flb_sds_t chunk, - struct s3_file *upload_file, - struct multipart_upload *m_upload_file, - const char *tag, int tag_len) +static int buffer_chunk(void *out_context, struct s3_file *upload_file, + flb_sds_t chunk, int chunk_size, + const char *tag, int tag_len, + time_t file_first_log_time, + char *input_name) { int ret; - char *buffer; - size_t buffer_size; struct flb_s3 *ctx = out_context; - /* Create buffer to upload to S3 */ - ret = construct_request_buffer(ctx, chunk, upload_file, &buffer, &buffer_size); - flb_sds_destroy(chunk); - if (ret < 0) { - flb_plg_error(ctx->ins, "Could not construct request buffer for %s", - upload_file->file_path); - return -1; - } - - /* Upload to S3 */ - ret = upload_data(ctx, upload_file, m_upload_file, buffer, buffer_size, tag, tag_len); - flb_free(buffer); - - return ret; -} - -static int buffer_chunk(void *out_context, struct s3_file *upload_file, flb_sds_t chunk, - int chunk_size, const char *tag, int tag_len) -{ - int ret; - struct flb_s3 *ctx = out_context; - - ret = s3_store_buffer_put(ctx, upload_file, tag, tag_len, chunk, (size_t) chunk_size); + ret = s3_store_buffer_put(ctx, upload_file, tag, + tag_len, chunk, (size_t) chunk_size, file_first_log_time, input_name); flb_sds_destroy(chunk); if (ret < 0) { flb_plg_warn(ctx->ins, "Could not buffer chunk. Data order preservation " @@ -1670,86 +1642,6 @@ static int buffer_chunk(void *out_context, struct s3_file *upload_file, flb_sds_ return 0; } -/* Uploads all chunk files in queue synchronously */ -static void s3_upload_queue(struct flb_config *config, void *out_context) -{ - int ret; - int async_flags; - time_t now; - struct upload_queue *upload_contents; - struct flb_s3 *ctx = out_context; - struct mk_list *tmp; - struct mk_list *head; - - flb_plg_debug(ctx->ins, "Running upload timer callback (upload_queue).."); - - /* 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); - } - - /* upload timer must use sync mode */ - if (ctx->use_put_object == FLB_TRUE) { - async_flags = ctx->s3_client->upstream->flags; - ctx->s3_client->upstream->flags &= ~(FLB_IO_ASYNC); - } - - /* 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); - - now = time(NULL); - - /* Checks if upload_contents is valid */ - ret = upload_queue_valid(upload_contents, now, ctx); - if (ret < 0) { - goto exit; - } - - /* 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) { - remove_from_queue(upload_contents); - ctx->retry_time = 0; - ctx->upload_queue_success = FLB_TRUE; - } - else { - s3_store_file_lock(upload_contents->upload_file); - ctx->upload_queue_success = FLB_FALSE; - - /* If retry limit was reached, discard file and remove file from queue */ - upload_contents->retry_counter++; - if (upload_contents->retry_counter >= MAX_UPLOAD_ERRORS) { - flb_plg_warn(ctx->ins, "Chunk file failed to send %d times, will not " - "retry", upload_contents->retry_counter); - s3_store_file_inactive(ctx, upload_contents->upload_file); - multipart_upload_destroy(upload_contents->m_upload_file); - remove_from_queue(upload_contents); - continue; - } - - /* Retry in N seconds */ - upload_contents->upload_time = now + 2 * upload_contents->retry_counter; - ctx->retry_time += 2 * upload_contents->retry_counter; - flb_plg_debug(ctx->ins, "Failed to upload file in upload_queue. Will not " - "retry for %d seconds", 2 * upload_contents->retry_counter); - break; - } - } - -exit: - /* re-enable async mode */ - if (ctx->use_put_object == FLB_TRUE) { - ctx->s3_client->upstream->flags = async_flags; - } -} static void cb_s3_upload(struct flb_config *config, void *data) { @@ -1764,48 +1656,69 @@ static void cb_s3_upload(struct flb_config *config, void *data) int complete; int ret; time_t now; - int async_flags; - flb_plg_debug(ctx->ins, "Running upload timer callback (cb_s3_upload).."); - - /* upload timer must use sync mode */ - if (ctx->use_put_object == FLB_TRUE) { - async_flags = ctx->s3_client->upstream->flags; - ctx->s3_client->upstream->flags &= ~(FLB_IO_ASYNC); + ret = pthread_mutex_trylock(&ctx->upload_queue_mutex); + if (ret != 0) { + /* don't block the thread, a coro is already flushing */ + return; } now = time(NULL); + + flb_plg_debug(ctx->ins, "Running scheduled timer uploader with coroutines (cb_s3_upload).."); - /* Check all chunks and see if any have timed out */ + /* check chunks in active stream not marked as ready to be sent and see if any are 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 */ + /* Locked chunks are already in the queue, skip */ if (chunk->locked == FLB_TRUE) { continue; } + if (now > (chunk->create_time + ctx->upload_timeout)) { + /* add to upload queue */ + if (chunk->input_name) { + flb_plg_info(ctx->ins, "upload_timeout reached for chunk from %s", + chunk->input_name); + } + s3_store_file_lock(chunk); + mk_list_add(&chunk->_head, &ctx->upload_queue); + } + } + + /* send any chunks that are ready */ + mk_list_foreach_safe(head, tmp, &ctx->upload_queue) { + chunk = mk_list_entry(head, struct s3_file, _head); + fsf = chunk->fsf; + 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); - continue; + if (ctx->preserve_data_ordering == FLB_TRUE) { + break; /* if preserve_data_ordering send in the queue order, do not skip over chunks */ + } else { + 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 (ret == -2 || ret == 0) { + /* if we succeeded or retries expired, delete chunk file and remove from queue */ + mk_list_del(&chunk->_head); + s3_store_file_delete(ctx, chunk); + } + + if (ret < 0) { + if (ctx->preserve_data_ordering == FLB_TRUE) { + break; /* if preserve_data_ordering send in the queue order, do not skip over chunks */ + } } } @@ -1814,10 +1727,11 @@ static void cb_s3_upload(struct flb_config *config, void *data) m_upload = mk_list_entry(head, struct multipart_upload, _head); complete = FLB_FALSE; - if (m_upload->complete_errors >= MAX_UPLOAD_ERRORS) { + if (ctx->ins->retry_limit >= 0 && m_upload->complete_errors > ctx->ins->retry_limit) { flb_plg_error(ctx->ins, - "Upload for %s has reached max completion errors, " - "plugin will give up", m_upload->s3_key); + "Multipart Upload for %s has failed " + "s3:CompleteMultipartUpload more than configured retry_limit, " + "output will give up ", m_upload->s3_key); mk_list_del(&m_upload->_head); continue; } @@ -1829,7 +1743,7 @@ 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)) { flb_plg_info(ctx->ins, "Completing upload for %s because upload_timeout" " has passed", m_upload->s3_key); complete = FLB_TRUE; @@ -1851,9 +1765,7 @@ static void cb_s3_upload(struct flb_config *config, void *data) } } - if (ctx->use_put_object == FLB_TRUE) { - ctx->s3_client->upstream->flags = async_flags; - } + pthread_mutex_unlock(&ctx->upload_queue_mutex); } static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char *data, @@ -2007,14 +1919,17 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char static void unit_test_flush(void *out_context, struct s3_file *upload_file, const char *tag, int tag_len, flb_sds_t chunk, - int chunk_size, struct multipart_upload *m_upload_file) + int chunk_size, struct multipart_upload *m_upload_file, + time_t file_first_log_time, + char *input_name) { int ret; char *buffer; size_t buffer_size; struct flb_s3 *ctx = out_context; - s3_store_buffer_put(ctx, upload_file, tag, tag_len, chunk, (size_t) chunk_size); + s3_store_buffer_put(ctx, upload_file, tag, tag_len, + chunk, (size_t) chunk_size, file_first_log_time, input_name); ret = construct_request_buffer(ctx, chunk, upload_file, &buffer, &buffer_size); if (ret < 0) { flb_plg_error(ctx->ins, "Could not construct request buffer for %s", @@ -2028,11 +1943,9 @@ static void unit_test_flush(void *out_context, struct s3_file *upload_file, FLB_OUTPUT_RETURN(ret); } -static void flush_init(void *out_context) +static void flush_startup_chunks(struct flb_s3 *ctx) { int ret; - struct flb_s3 *ctx = out_context; - struct flb_sched *sched; /* clean up any old buffers found on startup */ if (ctx->has_old_buffers == FLB_TRUE) { @@ -2041,40 +1954,53 @@ 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, "Failed to send locally buffered data left over " "from previous executions; will retry. Buffer=%s", ctx->fs->root_path); - FLB_OUTPUT_RETURN(FLB_RETRY); + } else { + flb_plg_info(ctx->ins, + "Successfully sent all locally buffered data left over " + "from previous executions. Buffer=%s", + ctx->fs->root_path); } } - /* - * create a timer that will run periodically and check if uploads - * are ready for completion - * this is created once on the first flush - */ - if (ctx->timer_created == FLB_FALSE) { - flb_plg_debug(ctx->ins, - "Creating upload timer with frequency %ds", - ctx->timer_ms / 1000); +} + +static void async_timer_cb(struct flb_config *config, void *data) +{ + struct flb_s3 *ctx = data; + + /* ensure that once shutdown begins, no new timer coros run */ + if (config->is_running == FLB_FALSE) { + return; + } + /* upload any ready chunks */ + cb_s3_upload(config, ctx); +} + +static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) +{ + struct flb_sched *sched; + int ret; + + flush_startup_chunks(ctx); + + if (ctx->timer_created == FLB_FALSE) { sched = flb_sched_ctx_get(); - if (ctx->preserve_data_ordering) { - ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, - ctx->timer_ms, s3_upload_queue, ctx, NULL); - } - else { - ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, - ctx->timer_ms, cb_s3_upload, ctx, NULL); - } - if (ret == -1) { + ret = flb_sched_out_async_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, + ctx->timer_ms, ctx->ins, + S3_UPLOAD_JOB_NAME, async_timer_cb, + ctx, NULL); + if (ret < 0) { flb_plg_error(ctx->ins, "Failed to create upload timer"); - FLB_OUTPUT_RETURN(FLB_RETRY); + return; } ctx->timer_created = FLB_TRUE; } @@ -2094,9 +2020,13 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, struct s3_file *upload_file = NULL; struct flb_s3 *ctx = out_context; struct multipart_upload *m_upload_file = NULL; + msgpack_unpacked result; + msgpack_object *obj; + size_t off = 0; + struct flb_time tms; + time_t file_first_log_time = 0; - /* Cleanup old buffers and initialize upload timer */ - flush_init(ctx); + s3_flush_init(config, ctx); /* Process chunk */ if (ctx->log_key) { @@ -2117,93 +2047,85 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, } chunk_size = flb_sds_len(chunk); + pthread_mutex_lock(&ctx->cb_flush_mutex); + /* Get a file candidate matching the given 'tag' */ upload_file = s3_store_file_get(ctx, event_chunk->tag, flb_sds_len(event_chunk->tag)); + if (upload_file == NULL) { + /* unpack msgpack */ + msgpack_unpacked_init(&result); + + /* Get the first record timestamp */ + while (msgpack_unpack_next(&result, + event_chunk->data, + event_chunk->size, &off) == MSGPACK_UNPACK_SUCCESS) { + flb_time_pop_from_msgpack(&tms, &result, &obj); + if (&tms.tm.tv_sec != 0) { + file_first_log_time = tms.tm.tv_sec; + break; + } + } + + msgpack_unpacked_destroy(&result); + } + else { + /* Get file_first_log_time from upload_file */ + file_first_log_time = upload_file->first_log_time; + } + + if (file_first_log_time == 0) { + file_first_log_time = time(NULL); + } + /* Specific to unit tests, will not get called normally */ if (s3_plugin_under_test() == FLB_TRUE) { unit_test_flush(ctx, upload_file, event_chunk->tag, flb_sds_len(event_chunk->tag), - chunk, chunk_size, m_upload_file); + chunk, chunk_size, + m_upload_file, file_first_log_time, i_ins->name); } - /* Discard upload_file if it has failed to upload MAX_UPLOAD_ERRORS times */ - if (upload_file != NULL && upload_file->failures >= MAX_UPLOAD_ERRORS) { - flb_plg_warn(ctx->ins, "File with tag %s failed to send %d times, will not " - "retry", event_chunk->tag, MAX_UPLOAD_ERRORS); - s3_store_file_inactive(ctx, upload_file); - upload_file = NULL; + /* + * Buffer new data in chunk in filesystem and wait for next data from engine + * If successful, data ordering is preserved to be the same as the engine sent us + */ + ret = buffer_chunk(ctx, upload_file, chunk, chunk_size, + event_chunk->tag, flb_sds_len(event_chunk->tag), + file_first_log_time, i_ins->name); + + if (ret < 0) { + pthread_mutex_unlock(&ctx->cb_flush_mutex); + FLB_OUTPUT_RETURN(FLB_RETRY); } /* If upload_timeout has elapsed, upload file */ if (upload_file != NULL && time(NULL) > (upload_file->create_time + ctx->upload_timeout)) { upload_timeout_check = FLB_TRUE; - flb_plg_info(ctx->ins, "upload_timeout reached for %s", - event_chunk->tag); + flb_plg_info(ctx->ins, "upload_timeout reached for chunk from %s, tag=%s", + i_ins->name, event_chunk->tag); } m_upload_file = get_upload(ctx, event_chunk->tag, flb_sds_len(event_chunk->tag)); - if (m_upload_file != NULL && time(NULL) > - (m_upload_file->init_time + ctx->upload_timeout)) { - upload_timeout_check = FLB_TRUE; - flb_plg_info(ctx->ins, "upload_timeout reached for %s", event_chunk->tag); - } - /* If total_file_size has been reached, upload file */ if ((upload_file && upload_file->size + chunk_size > ctx->upload_chunk_size) || (m_upload_file && m_upload_file->bytes + chunk_size > ctx->file_size)) { total_file_size_check = FLB_TRUE; } - /* File is ready for upload */ - if (upload_timeout_check == FLB_TRUE || total_file_size_check == FLB_TRUE) { - if (ctx->preserve_data_ordering == FLB_TRUE) { - /* Buffer last chunk in file and lock file to prevent further changes */ - ret = buffer_chunk(ctx, upload_file, chunk, chunk_size, - event_chunk->tag, flb_sds_len(event_chunk->tag)); - if (ret < 0) { - FLB_OUTPUT_RETURN(FLB_RETRY); - } - s3_store_file_lock(upload_file); - - /* Add chunk file to upload queue */ - ret = add_to_queue(ctx, upload_file, m_upload_file, - event_chunk->tag, flb_sds_len(event_chunk->tag)); - if (ret < 0) { - FLB_OUTPUT_RETURN(FLB_ERROR); - } - - /* Go through upload queue and return error if something went wrong */ - s3_upload_queue(config, ctx); - if (ctx->upload_queue_success == FLB_FALSE) { - ctx->upload_queue_success = FLB_TRUE; - FLB_OUTPUT_RETURN(FLB_ERROR); - } - FLB_OUTPUT_RETURN(FLB_OK); - } - else { - /* Send upload directly without upload queue */ - ret = send_upload_request(ctx, chunk, upload_file, m_upload_file, - event_chunk->tag, - flb_sds_len(event_chunk->tag)); - if (ret < 0) { - FLB_OUTPUT_RETURN(FLB_ERROR); - } - FLB_OUTPUT_RETURN(ret); - } + /* lock chunk file so new new appends. Its ready to be sent. */ + if ((upload_file != NULL) && (upload_timeout_check == FLB_TRUE || total_file_size_check == FLB_TRUE)) { + s3_store_file_lock(upload_file); + /* sends only happen from upload daemon coroutine which iterates over queue */ + mk_list_add(&upload_file->_head, &ctx->upload_queue); } - /* Buffer current chunk in filesystem and wait for next chunk from engine */ - ret = buffer_chunk(ctx, upload_file, chunk, chunk_size, - event_chunk->tag, flb_sds_len(event_chunk->tag)); - if (ret < 0) { - FLB_OUTPUT_RETURN(FLB_RETRY); - } + pthread_mutex_unlock(&ctx->cb_flush_mutex); FLB_OUTPUT_RETURN(FLB_OK); } @@ -2225,7 +2147,7 @@ static int cb_s3_exit(void *data, struct flb_config *config) ctx->s3_client->upstream->flags &= ~(FLB_IO_ASYNC); } 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 0f60aa90f0e..c04d3b5bfeb 100644 --- a/plugins/out_s3/s3.h +++ b/plugins/out_s3/s3.h @@ -25,6 +25,7 @@ #include #include #include +#include /* Upload data to S3 in 5MB chunks */ #define MIN_CHUNKED_UPLOAD_SIZE 5242880 @@ -42,34 +43,14 @@ #define MAX_FILE_SIZE 50000000000 #define MAX_FILE_SIZE_STR "50,000,000,000" +/* Used by engine to print active timer coro's on shutdown */ +#define S3_UPLOAD_JOB_NAME "Upload" + /* Allowed max file size 1 GB for publishing to S3 */ #define MAX_FILE_SIZE_PUT_OBJECT 1000000000 #define DEFAULT_UPLOAD_TIMEOUT 3600 -/* - * If we see repeated errors on an upload/chunk, we will discard it - * This saves us from scenarios where something goes wrong and an upload can - * not proceed (may be some other process completed it or deleted the upload) - * instead of erroring out forever, we eventually discard the upload. - * - * The same is done for chunks, just to be safe, even though realistically - * I can't think of a reason why a chunk could become unsendable. - */ -#define MAX_UPLOAD_ERRORS 5 - -struct upload_queue { - struct s3_file *upload_file; - struct multipart_upload *m_upload_file; - flb_sds_t tag; - int tag_len; - - int retry_counter; - time_t upload_time; - - struct mk_list _head; -}; - struct multipart_upload { flb_sds_t s3_key; flb_sds_t tag; @@ -93,6 +74,9 @@ struct multipart_upload { /* ongoing tracker of how much data has been sent for this upload */ size_t bytes; + /* for s3 retry warn message */ + char *input_name; + struct mk_list _head; /* see note for MAX_UPLOAD_ERRORS */ @@ -156,14 +140,14 @@ struct flb_s3 { struct mk_list uploads; - int preserve_data_ordering; - int upload_queue_success; + /* list of locked chunks that are ready to send */ struct mk_list upload_queue; + int preserve_data_ordering; + size_t file_size; size_t upload_chunk_size; time_t upload_timeout; - time_t retry_time; int timer_created; int timer_ms; @@ -174,6 +158,20 @@ struct flb_s3 { flb_sds_t metadata_dir; flb_sds_t seq_index_file; + /* + * Multiple timer coros can run at the same time, + * (even with single worker case, as one timer might yield and then another start) + * but modifying the pending chunk and upload lists, and deleting S3 store files needs + * to be concurrent safe + */ + pthread_mutex_t upload_queue_mutex; + /* + * If multiple workers are configured, then multiple cb_s3_flush can run at once + * (or cb_s3_flush can run at same time as a timer_coro does something) + * mutex is needed to protect chunk, upload_queue, multipart lists + */ + pthread_mutex_t cb_flush_mutex; + struct flb_output_instance *ins; }; diff --git a/plugins/out_s3/s3_store.c b/plugins/out_s3/s3_store.c index 050734a05f0..b857c1640f0 100644 --- a/plugins/out_s3/s3_store.c +++ b/plugins/out_s3/s3_store.c @@ -125,7 +125,9 @@ struct s3_file *s3_store_file_get(struct flb_s3 *ctx, const char *tag, /* Append data to a new or existing fstore file */ 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) + char *data, size_t bytes, + time_t file_first_log_time, + char* input_name) { int ret; flb_sds_t name; @@ -175,7 +177,9 @@ int s3_store_buffer_put(struct flb_s3 *ctx, struct s3_file *s3_file, return -1; } s3_file->fsf = fsf; + s3_file->first_log_time = file_first_log_time; s3_file->create_time = time(NULL); + s3_file->input_name = input_name; /* Use fstore opaque 'data' reference to keep our context */ fsf->data = s3_file; @@ -241,6 +245,7 @@ static int set_files_context(struct flb_s3 *ctx) continue; } s3_file->fsf = fsf; + s3_file->first_log_time = time(NULL); s3_file->create_time = time(NULL); /* Use fstore opaque 'data' reference to keep our context */ @@ -534,7 +539,3 @@ void s3_store_file_lock(struct s3_file *s3_file) s3_file->locked = FLB_TRUE; } -void s3_store_file_unlock(struct s3_file *s3_file) -{ - s3_file->locked = FLB_FALSE; -} diff --git a/plugins/out_s3/s3_store.h b/plugins/out_s3/s3_store.h index 242d99ab6fc..04539fb50e9 100644 --- a/plugins/out_s3/s3_store.h +++ b/plugins/out_s3/s3_store.h @@ -24,17 +24,22 @@ #include struct s3_file { - int locked; /* locked chunk is busy, cannot write to it */ + int locked; /* locked = no appends to this chunk */ int failures; /* delivery failures */ size_t size; /* file size */ time_t create_time; /* creation time */ + time_t first_log_time; /* first log time */ + char *input_name; /* for s3_retry_warn output message */ flb_sds_t file_path; /* file path */ struct flb_fstore_file *fsf; /* reference to parent flb_fstore_file */ + struct mk_list _head; }; 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); + char *data, size_t bytes, + time_t file_first_log_time, + char *input_name); int s3_store_init(struct flb_s3 *ctx); int s3_store_exit(struct flb_s3 *ctx); diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 256a40bda9f..bf015a86d88 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -6,6 +6,7 @@ set(src flb_mp.c flb_kv.c flb_api.c + flb_async_timer.c flb_csv.c flb_lib.c flb_log.c diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c new file mode 100644 index 00000000000..b8dc078ffd4 --- /dev/null +++ b/src/flb_async_timer.c @@ -0,0 +1,169 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include + +void flb_async_timer_destroy(struct flb_out_async_timer *timer) +{ + mk_list_del(&timer->_head); + flb_coro_destroy(timer->coro); + flb_free(timer); +} + +void flb_async_timer_cleanup(struct mk_list *destroy_list) +{ + struct flb_out_async_timer *async_timer; + struct mk_list *tmp; + struct mk_list *head; + mk_list_foreach_safe(head, tmp, destroy_list) { + async_timer = mk_list_entry(head, struct flb_out_async_timer, _head); + flb_async_timer_destroy(async_timer); + } +} + +void flb_output_async_timer_cleanup(struct flb_config *config) +{ + struct flb_output_instance *o_ins; + struct mk_list *tmp; + struct mk_list *head; + mk_list_foreach_safe(head, tmp, &config->outputs) { + o_ins = mk_list_entry(head, struct flb_output_instance, _head); + flb_async_timer_cleanup(o_ins->async_timer_list_destroy); + } +} + +int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int ms, + struct flb_output_instance *o_ins, + char *job_name, + void (*async_cb)(struct flb_config *, void *), + void *data, struct flb_sched_timer **out_timer) +{ + struct flb_out_async_timer_cb_data *timer_data; + + timer_data = flb_calloc(1, sizeof(struct flb_out_async_timer_cb_data)); + if (!timer_data) { + return; + } + + timer_data->ins = o_ins; + timer_data->job_name = job_name; + timer_data->cb = async_cb; + timer_data->data = data; + + return flb_sched_timer_cb_create(sched, type, ms, flb_out_async_sched_timer_cb, timer_data, NULL); +} + +/* Used in engine flb_running_count */ +int flb_async_timers_size(struct flb_output_instance *ins) +{ + int size = 0; + + if (flb_output_is_threaded(ins) == FLB_TRUE) { + /* + * On threaded mode, we need to count the active co-routines of + * every running thread of the thread pool. + */ + size = flb_thread_pool_async_timers_size(ins); + } + else { + size = mk_list_size(&ins->async_timer_list); + } + + return size; +} + +void flb_async_timers_print(struct mk_list *async_timer_list) +{ + struct flb_out_async_timer *async_timer; + struct mk_list *tmp; + struct mk_list *head; + int n = mk_list_size(async_timer_list); + if (n != 0) { + /* get one coro for the job_name */ + mk_list_foreach_safe(head, tmp, async_timer_list) { + async_timer = mk_list_entry(head, struct flb_out_async_timer, _head); + if (async_timer != NULL) { + flb_info("[task] output=%s still running %d %s(s)", + async_timer->o_ins->alias, n, async_timer->timer_data->job_name); + break; + } + } + } +} + +/* Used in engine flb_running_print */ +void flb_out_async_timers_print(struct flb_output_instance *ins) +{ + if (flb_output_is_threaded(ins) == FLB_TRUE) { + flb_thread_pool_async_timers_print(ins); + } + else { + flb_async_timers_print(&ins->async_timer_list); + } +} + +int flb_thread_pool_async_timers_size(struct flb_output_instance *ins) +{ + int n; + int size = 0; + struct mk_list *head; + struct flb_tp *tp = ins->tp; + struct flb_tp_thread *th; + struct flb_out_thread_instance *th_ins; + + mk_list_foreach(head, &tp->list_threads) { + th = mk_list_entry(head, struct flb_tp_thread, _head); + if (th->status != FLB_THREAD_POOL_RUNNING) { + continue; + } + + th_ins = th->params.data; + + pthread_mutex_lock(&th_ins->flush_mutex); + n = mk_list_size(&th_ins->async_timer_list); + pthread_mutex_unlock(&th_ins->flush_mutex); + size += n; + } + + return size; +} + +void flb_thread_pool_async_timers_print(struct flb_output_instance *ins) +{ + struct mk_list *head; + struct mk_list *tmp; + struct flb_tp *tp = ins->tp; + struct flb_tp_thread *th; + struct flb_out_thread_instance *th_ins; + + mk_list_foreach_safe(head, tmp, &tp->list_threads) { + th = mk_list_entry(head, struct flb_tp_thread, _head); + if (th->status != FLB_THREAD_POOL_RUNNING) { + continue; + } + + th_ins = th->params.data; + pthread_mutex_lock(&th_ins->async_timer_mutex); + flb_async_timers_print(&th_ins->async_timer_list); + pthread_mutex_unlock(&th_ins->async_timer_mutex); + } +} diff --git a/src/flb_engine.c b/src/flb_engine.c index ceb933dfca3..38db343c6f0 100644 --- a/src/flb_engine.c +++ b/src/flb_engine.c @@ -48,6 +48,7 @@ #include #include #include +#include #ifdef FLB_HAVE_METRICS #include @@ -393,6 +394,37 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts, return 0; } +static int flb_running_count(struct flb_config *config) +{ + int tasks = 0, timers = 0, n = 0; + struct mk_list *head; + struct mk_list *tmp; + struct flb_output_instance *o_ins; + + mk_list_foreach_safe(head, tmp, &config->outputs) { + o_ins = mk_list_entry(head, struct flb_output_instance, _head); + n = flb_async_timers_size(o_ins); + timers = timers + n; + } + + tasks = flb_task_running_count(config); + return tasks + timers; +} + +static void flb_running_print(struct flb_config *config) +{ + struct mk_list *head; + struct mk_list *tmp; + struct flb_output_instance *o_ins; + + flb_task_running_print(config); + + mk_list_foreach_safe(head, tmp, &config->outputs) { + o_ins = mk_list_entry(head, struct flb_output_instance, _head); + flb_thread_pool_async_timers_print(o_ins); + } +} + static inline int flb_engine_manager(flb_pipefd_t fd, struct flb_config *config) { int bytes; @@ -551,6 +583,7 @@ int sb_segregate_chunks(struct flb_config *config) int flb_engine_start(struct flb_config *config) { int ret; + int count; uint64_t ts; char tmp[16]; struct flb_time t_flush; @@ -821,19 +854,19 @@ int flb_engine_start(struct flb_config *config) * resources allocated by that co-routine, the best thing is to * wait again for the grace period and re-check again. */ - ret = flb_task_running_count(config); - if (ret > 0 && config->grace_count < config->grace) { + count = flb_running_count(config); + if (count > 0 && config->grace_count < config->grace) { if (config->grace_count == 1) { - flb_task_running_print(config); + flb_running_print(config); } flb_engine_exit(config); } else { - if (ret > 0) { - flb_task_running_print(config); + if (count > 0) { + flb_running_print(config); } - flb_info("[engine] service has stopped (%i pending tasks)", - ret); + flb_info("[engine] service has stopped (%d pending tasks)", + count); ret = config->exit_status_code; flb_engine_shutdown(config); config = NULL; @@ -896,6 +929,7 @@ int flb_engine_start(struct flb_config *config) flb_net_dns_lookup_context_cleanup(&dns_ctx); flb_sched_timer_cleanup(config->sched); flb_upstream_conn_pending_destroy_list(&config->upstreams); + flb_output_async_timer_cleanup(config); /* * depend on main thread to clean up expired message diff --git a/src/flb_output.c b/src/flb_output.c index 07be1bf3314..e5f5537ce5b 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -40,10 +40,12 @@ #include FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); +FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); void flb_output_prepare() { FLB_TLS_INIT(out_flush_params); + FLB_TLS_INIT(async_timer_coro_params); } /* Validate the the output address protocol */ @@ -478,6 +480,11 @@ void flb_output_exit(struct flb_config *config) if (params) { flb_free(params); } + params = FLB_TLS_GET(async_timer_coro_params); + if (params) { + flb_free(params); + } + } static inline int instance_id(struct flb_config *config) @@ -710,6 +717,8 @@ struct flb_output_instance *flb_output_new(struct flb_config *config, mk_list_init(&instance->upstreams); mk_list_init(&instance->flush_list); mk_list_init(&instance->flush_list_destroy); + mk_list_init(&instance->async_timer_list); + mk_list_init(&instance->async_timer_list_destroy); mk_list_add(&instance->_head, &config->outputs); diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index 470aa514cf5..4aa55da009c 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -25,6 +25,7 @@ #include #include #include +#include static pthread_once_t local_thread_instance_init = PTHREAD_ONCE_INIT; FLB_TLS_DEFINE(struct flb_out_thread_instance, local_thread_instance); @@ -180,7 +181,8 @@ static void output_thread(void *data) struct flb_output_instance *ins; struct flb_output_flush *out_flush; struct flb_out_thread_instance *th_ins = data; - struct flb_out_flush_params *params; + struct flb_out_flush_params *flush_params = NULL; + struct flb_out_async_timer *timer_params = NULL; struct flb_net_dns dns_ctx; /* Register thread instance */ @@ -330,9 +332,10 @@ static void output_thread(void *data) /* Destroy upstream connections from the 'pending destroy list' */ flb_upstream_conn_pending_destroy_list(&th_ins->upstreams); flb_sched_timer_cleanup(sched); + flb_async_timer_cleanup(&th_ins->async_timer_list_destroy); /* Check if we should stop the event loop */ - if (stopping == FLB_TRUE && mk_list_size(&th_ins->flush_list) == 0) { + if (stopping == FLB_TRUE && mk_list_size(&th_ins->flush_list) == 0 && mk_list_size(&th_ins->async_timer_list) == 0) { /* * If there are no busy network connections (and no coroutines) its * safe to stop it. @@ -356,11 +359,16 @@ static void output_thread(void *data) upstream_thread_destroy(th_ins); flb_upstream_conn_active_destroy_list(&th_ins->upstreams); flb_upstream_conn_pending_destroy_list(&th_ins->upstreams); + flb_async_timer_cleanup(&th_ins->async_timer_list_destroy); flb_sched_destroy(sched); - params = FLB_TLS_GET(out_flush_params); - if (params) { - flb_free(params); + flush_params = FLB_TLS_GET(out_flush_params); + if (flush_params) { + flb_free(flush_params); + } + timer_params = FLB_TLS_GET(async_timer_coro_params); + if (timer_params) { + flb_free(timer_params); } mk_event_loop_destroy(th_ins->evl); flb_bucket_queue_destroy(th_ins->evl_bktq); @@ -436,7 +444,10 @@ int flb_output_thread_pool_create(struct flb_config *config, th_ins->flush_id = 0; mk_list_init(&th_ins->flush_list); mk_list_init(&th_ins->flush_list_destroy); + mk_list_init(&th_ins->async_timer_list); + mk_list_init(&th_ins->async_timer_list_destroy); pthread_mutex_init(&th_ins->flush_mutex, NULL); + pthread_mutex_init(&th_ins->async_timer_mutex, NULL); mk_list_init(&th_ins->upstreams); upstream_thread_create(th_ins, ins); @@ -503,7 +514,6 @@ int flb_output_thread_pool_coros_size(struct flb_output_instance *ins) struct flb_tp_thread *th; struct flb_out_thread_instance *th_ins; - /* Signal each worker thread that needs to stop doing work */ mk_list_foreach(head, &tp->list_threads) { th = mk_list_entry(head, struct flb_tp_thread, _head); if (th->status != FLB_THREAD_POOL_RUNNING) { diff --git a/src/flb_scheduler.c b/src/flb_scheduler.c index 21028d19dca..1aacf554f11 100644 --- a/src/flb_scheduler.c +++ b/src/flb_scheduler.c @@ -26,6 +26,7 @@ #include #include #include +#include #include #include