From 5562a2bfd607029c3c89bd063ebc6e2fb72a7341 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 15:42:21 -0700 Subject: [PATCH 01/22] output: add support for scheduled timer jobs with coroutines Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_output.h | 213 +++++++++++++++++++++++++++++++- src/flb_output.c | 7 ++ 2 files changed, 219 insertions(+), 1 deletion(-) diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index c9172b65d17..1444ec9c59a 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -408,6 +408,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 timer_coro_list; + struct mk_list timer_coro_list_destroy; + /* Keep a reference to the original context this instance belongs to */ struct flb_config *config; }; @@ -428,6 +432,28 @@ struct flb_output_flush { struct mk_list _head; /* Link to flb_task->threads */ }; +/* + * stores timer coros on the timer_coro_list, if the output uses them + */ +struct flb_output_timer_coro { + struct flb_config *config; /* FLB context */ + struct flb_output_instance *o_ins; /* output instance */ + struct flb_output_coro_timer_data *timer_data; /* callback info */ + struct flb_coro *coro; /* parent coro addr */ + struct mk_list _head; /* Link to timer_coro_list */ +}; + +/* + * If the output uses timer coros, then this is used as the callback data + * passed to flb_sched_timer_cb_create + */ +struct flb_output_coro_timer_data { + struct flb_output_instance *ins; /* associate coro with this output instance */ + flb_sds_t job_name; /* used on engine shutdown, print pending "custom" jobs */ + void (*cb) (struct flb_config *config, void *data); /* call this output callback in the coro */ + void *data; /* opaque data to pass to the above cb */ +}; + static FLB_INLINE int flb_output_is_threaded(struct flb_output_instance *ins) { return ins->is_threaded; @@ -443,6 +469,19 @@ static FLB_INLINE void flb_output_flush_destroy(struct flb_output_flush *out_flu flb_free(out_flush); } +/* + * See below note for flb_out_flush_params + * this is equivalent for timer coroutines + */ +struct flb_out_timer_coro_params { + struct flb_output_timer_coro *output_timer; /* output flush */ + struct flb_output_coro_timer_data *timer_data; /* callback info */ + struct flb_config *config; /* Fluent Bit context */ + struct flb_coro *coro; /* coroutine context */ +}; + +extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, timer_coro_params); + /* * libco do not support parameters in the entrypoint function due to the * complexity of implementation in terms of architecture and compiler, but @@ -526,6 +565,125 @@ static FLB_INLINE void output_pre_cb_flush(void) persisted_params.config); } +/* same as above but for timer coros */ +static FLB_INLINE void output_pre_timer_cb(void) +{ + struct flb_coro *coro; + struct flb_out_timer_coro_params *params; + struct flb_output_instance *o_ins; + struct flb_out_thread_instance *th_ins; + struct flb_output_timer_coro *timer_coro; + + + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(timer_coro_params); + if (!params) { + flb_error("[output] no timer coro params defined, unexpected"); + return; + } + + coro = params->coro; + timer_coro = params->output_timer; + o_ins = params->output_timer->o_ins; + + /* Run the callback provided by the output plugin */ + timer_data->cb(params->config, params->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->timer_mutex); + mk_list_del(&timer_coro->_head); + mk_list_add(&timer_coro->_head, &th_ins->timer_coro_list_destroy); + pthread_mutex_unlock(&th_ins->timer_mutex); + } + else { + mk_list_del(&timer_coro->_head); + mk_list_add(&timer_coro->_head, &o_ins->timer_coro_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_output_coro_timer_cb(struct flb_config *config, void *data) +{ + size_t stack_size; + struct flb_coro *coro; + struct flb_output_timer_coro *timer_coro; + struct flb_out_thread_instance *th_ins; + struct flb_output_coro_timer_data *ctx = (struct flb_output_coro_timer_data *) data; + struct flb_out_timer_coro_params *params; + struct flb_output_instance *o_ins; + + /* Custom output coroutine info */ + timer_coro = (struct flb_output_timer_coro *) flb_calloc(1, sizeof(struct flb_output_timer_coro)); + if (!timer_coro) { + flb_errno(); + return; + } + + /* Create a new co-routine */ + coro = flb_coro_create(timer_coro); + if (!coro) { + flb_free(timer_coro); + return; + } + + o_ins = ctx->ins; + timer_coro->o_ins = o_ins; + timer_coro->config = config; + timer_coro->coro = coro; + + coro->callee = co_create(config->coro_stack_size, + output_pre_timer_cb, &stack_size); + + if (coro->callee == NULL) { + flb_coro_destroy(coro); + flb_free(timer_coro); + 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->timer_mutex); + mk_list_add(&timer_coro->_head, &th_ins->timer_coro_list); + pthread_mutex_unlock(&th_ins->timer_mutex); + } + else { + mk_list_add(&timer_coro->_head, &o_ins->timer_coro_list); + } + + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(timer_coro_params); + if (!params) { + params = (struct flb_out_timer_coro_params *) flb_calloc(1, sizeof(struct flb_out_flush_params)); + if (!params) { + flb_errno(); + return; + } + } + + /* Callback parameters in order */ + params->output_timer = timer_coro; + params->timer_data = ctx; + params->config = config; + params->coro = coro; + + FLB_TLS_SET(timer_coro_params, params); + coro->caller = co_active(); + flb_coro_resume(coro); + return; +} + void flb_output_flush_prepare_destroy(struct flb_output_flush *out_flush); int flb_output_flush_id_get(struct flb_output_instance *ins); @@ -659,7 +817,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; @@ -678,6 +840,55 @@ static inline int flb_output_coros_size(struct flb_output_instance *ins) return size; } +/* Used in engine flb_running_count */ +static inline int flb_output_timer_coros_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_output_thread_pool_timer_coros_size(ins); + } + else { + size = mk_list_size(&ins->timer_coro_list); + } + + return size; +} + +static inline void flb_timer_coros_print(struct mk_list *timer_coro_list) +{ + struct flb_output_timer_coro *timer_coro; + struct mk_list *tmp; + struct mk_list *head; + int n = mk_list_size(timer_coro_list); + if (n != 0) { + /* get one coro for the job_name */ + mk_list_foreach_safe(head, tmp, timer_coro_list) { + timer_coro = mk_list_entry(head, struct flb_output_timer_coro, _head); + if (timer_coro != NULL) { + flb_info("[task] output=%s still running %d %s(s)", + timer_coro->o_ins->alias, n, timer_coro->timer_data->job_name); + break; + } + } + } +} + +/* Used in engine flb_running_print */ +static inline void flb_output_timer_coros_print(struct flb_output_instance *ins) +{ + if (flb_output_is_threaded(ins) == FLB_TRUE) { + flb_output_thread_pool_timer_coros_print(ins); + } + else { + flb_timer_coros_print(&ins->timer_coro_list); + } +} + static inline void flb_output_return_do(int x) { struct flb_coro *coro; diff --git a/src/flb_output.c b/src/flb_output.c index 07be1bf3314..2ba9c9a0af9 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_timer_coro_params, timer_coro_params); void flb_output_prepare() { FLB_TLS_INIT(out_flush_params); + FLB_TLS_INIT(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(timer_coro_params); + if (params) { + flb_free(params); + } + } static inline int instance_id(struct flb_config *config) From fdda9d537acc1a1b84986aa1bb65363325d4c334 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 15:43:27 -0700 Subject: [PATCH 02/22] output_thread: add support for scheduled timer jobs with coroutines Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_output_thread.h | 10 +++- src/flb_output_thread.c | 66 +++++++++++++++++++++++--- 2 files changed, 68 insertions(+), 8 deletions(-) diff --git a/include/fluent-bit/flb_output_thread.h b/include/fluent-bit/flb_output_thread.h index 4100c9c1a7e..38a59276025 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 timer_coro_list; /* flush context list */ + struct mk_list timer_coro_list_destroy; /* flust context destroy list */ + pthread_mutex_t timer_mutex; /* mutex for 'flush_list' */ /* 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_output_thread_pool_timer_coros_size(struct flb_output_instance *ins); +void flb_output_thread_pool_timer_coros_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/src/flb_output_thread.c b/src/flb_output_thread.c index 470aa514cf5..9dafddf68b5 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -180,7 +180,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_timer_coro_params *timer_params = NULL; struct flb_net_dns dns_ctx; /* Register thread instance */ @@ -332,7 +333,7 @@ static void output_thread(void *data) flb_sched_timer_cleanup(sched); /* 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->timer_coro_list) == 0) { /* * If there are no busy network connections (and no coroutines) its * safe to stop it. @@ -358,9 +359,13 @@ static void output_thread(void *data) flb_upstream_conn_pending_destroy_list(&th_ins->upstreams); 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(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 +441,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->timer_coro_list); + mk_list_init(&th_ins->timer_coro_list_destroy); pthread_mutex_init(&th_ins->flush_mutex, NULL); + pthread_mutex_init(&th_ins->timer_mutex, NULL); mk_list_init(&th_ins->upstreams); upstream_thread_create(th_ins, ins); @@ -494,6 +502,53 @@ int flb_output_thread_pool_create(struct flb_config *config, return 0; } +int flb_output_thread_pool_timer_coros_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->timer_coro_list); + pthread_mutex_unlock(&th_ins->flush_mutex); + size += n; + } + + return size; +} + +void flb_output_thread_pool_timer_coros_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->timer_mutex); + flb_timer_coros_print(&th_ins->timer_coro_list); + pthread_mutex_unlock(&th_ins->timer_mutex); + } +} + int flb_output_thread_pool_coros_size(struct flb_output_instance *ins) { int n; @@ -503,7 +558,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) { From a4407b6ca843384ce48aa8d30e6b58025ea64ee5 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 15:44:03 -0700 Subject: [PATCH 03/22] engine: print and track scheduled timer coroutines on shutdown Signed-off-by: Wesley Pettit --- src/flb_engine.c | 46 +++++++++++++++++++++++++++++++++++++++------- 1 file changed, 39 insertions(+), 7 deletions(-) diff --git a/src/flb_engine.c b/src/flb_engine.c index ceb933dfca3..4d45840ac1f 100644 --- a/src/flb_engine.c +++ b/src/flb_engine.c @@ -393,6 +393,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_output_timer_coros_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_output_timer_coros_print(o_ins); + } +} + static inline int flb_engine_manager(flb_pipefd_t fd, struct flb_config *config) { int bytes; @@ -551,6 +582,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 +853,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; From c73fef00b11d78e2eb1f8f1c15496978a6ee79d2 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 15:45:08 -0700 Subject: [PATCH 04/22] out_s3: stability refactor This commit combines many changes designed to improve S3 stability: - code clean up and refactoring for readability - re-implement preserve_data_ordering code to fix several bugs and simplify logic. - all uploads now happen outside of cb_flush - uploads performed with new scheduled timer jobs with coroutines - S3 always uses async IO, with mutex lock to protect it from concurrency issues - remove trailing slash in store_dir to prevent double // in final path Signed-off-by: Wesley Pettit --- plugins/out_s3/s3.c | 756 +++++++++++++++++--------------------- plugins/out_s3/s3.h | 43 +-- plugins/out_s3/s3_store.c | 12 +- plugins/out_s3/s3_store.h | 9 +- 4 files changed, 376 insertions(+), 444 deletions(-) diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index b4556158000..51b42fc5e4a 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 @@ -45,10 +48,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 +59,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 +99,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 +306,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 +371,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; @@ -437,8 +499,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 +550,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 +580,7 @@ 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->flush_mutex, NULL); /* Export context */ flb_output_set_context(ins, ctx); @@ -924,7 +976,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 +1001,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 +1012,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 +1029,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 +1077,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 +1108,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 +1146,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 +1159,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 +1207,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 +1221,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 +1230,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 +1250,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 +1285,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 +1361,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 +1376,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 +1389,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 +1406,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; @@ -1440,13 +1518,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 +1557,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 +1566,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 +1581,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 +1599,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 */ @@ -1550,117 +1620,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) -{ - 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) +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; 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 +1640,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 +1654,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->flush_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 +1725,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 +1741,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 +1763,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->flush_mutex); } static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char *data, @@ -2007,14 +1917,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 +1941,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 +1952,71 @@ 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 timer_coro_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; + struct flb_output_coro_timer_data *timer_data = NULL; + flb_sds_t job_name; + 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); + job_name = flb_sds_create(S3_UPLOAD_JOB_NAME); + if (!job_name) { + return; } - else { - ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, - ctx->timer_ms, cb_s3_upload, ctx, NULL); + + timer_data = flb_calloc(1, sizeof(struct flb_output_coro_timer_data)); + if (!timer_data) { + flb_sds_destroy(job_name); + return; } - if (ret == -1) { + + timer_data->ins = ctx->ins; + timer_data->job_name = job_name; + timer_data->cb = timer_coro_cb; + timer_data->data = ctx; + + ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, + ctx->timer_ms, flb_output_coro_timer_cb, timer_data, NULL); + if (ret < 0) { flb_plg_error(ctx->ins, "Failed to create upload timer"); - FLB_OUTPUT_RETURN(FLB_RETRY); + flb_free(timer_data); + flb_sds_destroy(job_name); + return; } ctx->timer_created = FLB_TRUE; } @@ -2094,9 +2036,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) { @@ -2122,88 +2068,76 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, 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) { + 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); - } FLB_OUTPUT_RETURN(FLB_OK); } @@ -2225,7 +2159,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..0be62ddaa27 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,13 @@ struct flb_s3 { flb_sds_t metadata_dir; flb_sds_t seq_index_file; + /* + * Multiple timer coros can run at the same time, + * but modifying the pending chunk and upload lists, and deleting S3 store files needs + * to be concurrent safe + */ + pthread_mutex_t 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..34f10e1ea17 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 */ @@ -413,6 +418,7 @@ int s3_store_file_inactive(struct flb_s3 *ctx, struct s3_file *s3_file) int s3_store_file_delete(struct flb_s3 *ctx, struct s3_file *s3_file) { struct flb_fstore_file *fsf; + flb_plg_info(ctx->ins, "s3_store_file_delete chunk=%p", s3_file); fsf = s3_file->fsf; ctx->current_buffer_size -= s3_file->size; @@ -534,7 +540,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); From 52021996fd8d17618450cfb17b701cfb412b0dd6 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 16:24:35 -0700 Subject: [PATCH 05/22] output: fix variable name in output_pre_timer_cb Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_output.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index 1444ec9c59a..941ea6e03ee 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -586,7 +586,7 @@ static FLB_INLINE void output_pre_timer_cb(void) o_ins = params->output_timer->o_ins; /* Run the callback provided by the output plugin */ - timer_data->cb(params->config, params->timer_data->data); + params->timer_data->cb(params->config, params->timer_data->data); /* move coro to destroy queue */ if (flb_output_is_threaded(o_ins) == FLB_TRUE) { From 814c8c7447d9489b291a4d4074a573f5e51ea53c Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 26 Sep 2023 16:28:36 -0700 Subject: [PATCH 06/22] out_s3: remove unneeded info message Signed-off-by: Wesley Pettit --- plugins/out_s3/s3_store.c | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/out_s3/s3_store.c b/plugins/out_s3/s3_store.c index 34f10e1ea17..b857c1640f0 100644 --- a/plugins/out_s3/s3_store.c +++ b/plugins/out_s3/s3_store.c @@ -418,7 +418,6 @@ int s3_store_file_inactive(struct flb_s3 *ctx, struct s3_file *s3_file) int s3_store_file_delete(struct flb_s3 *ctx, struct s3_file *s3_file) { struct flb_fstore_file *fsf; - flb_plg_info(ctx->ins, "s3_store_file_delete chunk=%p", s3_file); fsf = s3_file->fsf; ctx->current_buffer_size -= s3_file->size; From c710a19e94fa353348a6bc7f2247ed34b3588024 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Fri, 6 Oct 2023 15:13:18 -0700 Subject: [PATCH 07/22] WIP: renaming Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_async_timer.h | 35 +++++++++++++++ include/fluent-bit/flb_output.h | 62 +++++++++++++------------- include/fluent-bit/flb_output_thread.h | 4 +- plugins/out_s3/s3.c | 4 +- src/flb_output.c | 6 +-- src/flb_output_thread.c | 12 ++--- 6 files changed, 79 insertions(+), 44 deletions(-) create mode 100644 include/fluent-bit/flb_async_timer.h diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h new file mode 100644 index 00000000000..8c48b70ddec --- /dev/null +++ b/include/fluent-bit/flb_async_timer.h @@ -0,0 +1,35 @@ +/* -*- 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 + + + +#endif diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index 941ea6e03ee..92c1b4a0810 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -409,8 +409,8 @@ struct flb_output_instance { struct mk_list flush_list_destroy; /* similar to flush coroutine list above, timer coroutine list */ - struct mk_list timer_coro_list; - struct mk_list timer_coro_list_destroy; + 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; @@ -433,24 +433,24 @@ struct flb_output_flush { }; /* - * stores timer coros on the timer_coro_list, if the output uses them + * stores timer coros on the async_timer_list, if the output uses them */ -struct flb_output_timer_coro { +struct flb_out_async_timer{ struct flb_config *config; /* FLB context */ struct flb_output_instance *o_ins; /* output instance */ - struct flb_output_coro_timer_data *timer_data; /* callback info */ + struct flb_out_async_timer_cb_data *timer_data; /* callback info */ struct flb_coro *coro; /* parent coro addr */ - struct mk_list _head; /* Link to timer_coro_list */ + 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_output_coro_timer_data { +struct flb_out_async_timer_cb_data { struct flb_output_instance *ins; /* associate coro with this output instance */ flb_sds_t job_name; /* used on engine shutdown, print pending "custom" jobs */ - void (*cb) (struct flb_config *config, void *data); /* call this output callback in the coro */ + 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 */ }; @@ -474,13 +474,13 @@ static FLB_INLINE void flb_output_flush_destroy(struct flb_output_flush *out_flu * this is equivalent for timer coroutines */ struct flb_out_timer_coro_params { - struct flb_output_timer_coro *output_timer; /* output flush */ - struct flb_output_coro_timer_data *timer_data; /* callback info */ + struct flb_out_async_timer*output_timer; /* output flush */ + struct flb_out_async_timer_cb_data *timer_data; /* callback info */ struct flb_config *config; /* Fluent Bit context */ struct flb_coro *coro; /* coroutine context */ }; -extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, timer_coro_params); +extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, out_async_timer_param); /* * libco do not support parameters in the entrypoint function due to the @@ -566,16 +566,16 @@ static FLB_INLINE void output_pre_cb_flush(void) } /* same as above but for timer coros */ -static FLB_INLINE void output_pre_timer_cb(void) +static FLB_INLINE void out_async_timer_cb(void) { struct flb_coro *coro; struct flb_out_timer_coro_params *params; struct flb_output_instance *o_ins; struct flb_out_thread_instance *th_ins; - struct flb_output_timer_coro *timer_coro; + struct flb_out_async_timer*timer_coro; - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(timer_coro_params); + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(out_async_timer_param); if (!params) { flb_error("[output] no timer coro params defined, unexpected"); return; @@ -593,12 +593,12 @@ static FLB_INLINE void output_pre_timer_cb(void) th_ins = flb_output_thread_instance_get(); pthread_mutex_lock(&th_ins->timer_mutex); mk_list_del(&timer_coro->_head); - mk_list_add(&timer_coro->_head, &th_ins->timer_coro_list_destroy); + mk_list_add(&timer_coro->_head, &th_ins->async_timer_list_destroy); pthread_mutex_unlock(&th_ins->timer_mutex); } else { mk_list_del(&timer_coro->_head); - mk_list_add(&timer_coro->_head, &o_ins->timer_coro_list_destroy); + mk_list_add(&timer_coro->_head, &o_ins->async_timer_list_destroy); } /* timer coro is complete; yield back to caller/control code */ @@ -610,18 +610,18 @@ static FLB_INLINE void output_pre_timer_cb(void) * this function is used as the callback for flb_sched_timer_cb_create */ static FLB_INLINE -void flb_output_coro_timer_cb(struct flb_config *config, void *data) +void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) { size_t stack_size; struct flb_coro *coro; - struct flb_output_timer_coro *timer_coro; + struct flb_out_async_timer*timer_coro; struct flb_out_thread_instance *th_ins; - struct flb_output_coro_timer_data *ctx = (struct flb_output_coro_timer_data *) data; + struct flb_out_async_timer_cb_data *ctx = (struct flb_out_async_timer_cb_data *) data; struct flb_out_timer_coro_params *params; struct flb_output_instance *o_ins; /* Custom output coroutine info */ - timer_coro = (struct flb_output_timer_coro *) flb_calloc(1, sizeof(struct flb_output_timer_coro)); + timer_coro = (struct flb_out_async_timer*) flb_calloc(1, sizeof(struct flb_output_timer_coro)); if (!timer_coro) { flb_errno(); return; @@ -640,7 +640,7 @@ void flb_output_coro_timer_cb(struct flb_config *config, void *data) timer_coro->coro = coro; coro->callee = co_create(config->coro_stack_size, - output_pre_timer_cb, &stack_size); + out_async_timer_cb, &stack_size); if (coro->callee == NULL) { flb_coro_destroy(coro); @@ -656,14 +656,14 @@ void flb_output_coro_timer_cb(struct flb_config *config, void *data) if (o_ins->is_threaded == FLB_TRUE) { th_ins = flb_output_thread_instance_get(); pthread_mutex_lock(&th_ins->timer_mutex); - mk_list_add(&timer_coro->_head, &th_ins->timer_coro_list); + mk_list_add(&timer_coro->_head, &th_ins->async_timer_list); pthread_mutex_unlock(&th_ins->timer_mutex); } else { - mk_list_add(&timer_coro->_head, &o_ins->timer_coro_list); + mk_list_add(&timer_coro->_head, &o_ins->async_timer_list); } - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(timer_coro_params); + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(out_async_timer_param); if (!params) { params = (struct flb_out_timer_coro_params *) flb_calloc(1, sizeof(struct flb_out_flush_params)); if (!params) { @@ -678,7 +678,7 @@ void flb_output_coro_timer_cb(struct flb_config *config, void *data) params->config = config; params->coro = coro; - FLB_TLS_SET(timer_coro_params, params); + FLB_TLS_SET(out_async_timer_param, params); coro->caller = co_active(); flb_coro_resume(coro); return; @@ -853,21 +853,21 @@ static inline int flb_output_timer_coros_size(struct flb_output_instance *ins) size = flb_output_thread_pool_timer_coros_size(ins); } else { - size = mk_list_size(&ins->timer_coro_list); + size = mk_list_size(&ins->async_timer_list); } return size; } -static inline void flb_timer_coros_print(struct mk_list *timer_coro_list) +static inline void flb_timer_coros_print(struct mk_list *async_timer_list) { - struct flb_output_timer_coro *timer_coro; + struct flb_out_async_timer*timer_coro; struct mk_list *tmp; struct mk_list *head; - int n = mk_list_size(timer_coro_list); + int n = mk_list_size(async_timer_list); if (n != 0) { /* get one coro for the job_name */ - mk_list_foreach_safe(head, tmp, timer_coro_list) { + mk_list_foreach_safe(head, tmp, async_timer_list) { timer_coro = mk_list_entry(head, struct flb_output_timer_coro, _head); if (timer_coro != NULL) { flb_info("[task] output=%s still running %d %s(s)", @@ -885,7 +885,7 @@ static inline void flb_output_timer_coros_print(struct flb_output_instance *ins) flb_output_thread_pool_timer_coros_print(ins); } else { - flb_timer_coros_print(&ins->timer_coro_list); + flb_timer_coros_print(&ins->async_timer_list); } } diff --git a/include/fluent-bit/flb_output_thread.h b/include/fluent-bit/flb_output_thread.h index 38a59276025..266d835aa24 100644 --- a/include/fluent-bit/flb_output_thread.h +++ b/include/fluent-bit/flb_output_thread.h @@ -89,8 +89,8 @@ struct flb_out_thread_instance { pthread_mutex_t flush_mutex; /* mutex for 'flush_list' */ /* Same as flush_mutex but for timer coros */ - struct mk_list timer_coro_list; /* flush context list */ - struct mk_list timer_coro_list_destroy; /* flust context destroy list */ + struct mk_list async_timer_list; /* flush context list */ + struct mk_list async_timer_list_destroy; /* flust context destroy list */ pthread_mutex_t timer_mutex; /* mutex for 'flush_list' */ /* List of mapped 'upstream' contexts */ diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index 51b42fc5e4a..a55f82d5fde 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -1985,7 +1985,7 @@ static void timer_coro_cb(struct flb_config *config, void *data) static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) { struct flb_sched *sched; - struct flb_output_coro_timer_data *timer_data = NULL; + struct flb_out_async_timer_cb_data *timer_data = NULL; flb_sds_t job_name; int ret; @@ -2011,7 +2011,7 @@ static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) timer_data->data = ctx; ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, - ctx->timer_ms, flb_output_coro_timer_cb, timer_data, NULL); + ctx->timer_ms, flb_out_async_sched_timer_cb, timer_data, NULL); if (ret < 0) { flb_plg_error(ctx->ins, "Failed to create upload timer"); flb_free(timer_data); diff --git a/src/flb_output.c b/src/flb_output.c index 2ba9c9a0af9..4b415cd8933 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -40,12 +40,12 @@ #include FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); -FLB_TLS_DEFINE(struct flb_out_timer_coro_params, timer_coro_params); +FLB_TLS_DEFINE(struct flb_out_timer_coro_params, out_async_timer_param); void flb_output_prepare() { FLB_TLS_INIT(out_flush_params); - FLB_TLS_INIT(timer_coro_params); + FLB_TLS_INIT(out_async_timer_param); } /* Validate the the output address protocol */ @@ -480,7 +480,7 @@ void flb_output_exit(struct flb_config *config) if (params) { flb_free(params); } - params = FLB_TLS_GET(timer_coro_params); + params = FLB_TLS_GET(out_async_timer_param); if (params) { flb_free(params); } diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index 9dafddf68b5..bc2882d7364 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -333,7 +333,7 @@ static void output_thread(void *data) flb_sched_timer_cleanup(sched); /* Check if we should stop the event loop */ - if (stopping == FLB_TRUE && mk_list_size(&th_ins->flush_list) == 0 && mk_list_size(&th_ins->timer_coro_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. @@ -363,7 +363,7 @@ static void output_thread(void *data) if (flush_params) { flb_free(flush_params); } - timer_params = FLB_TLS_GET(timer_coro_params); + timer_params = FLB_TLS_GET(out_async_timer_param); if (timer_params) { flb_free(timer_params); } @@ -441,8 +441,8 @@ 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->timer_coro_list); - mk_list_init(&th_ins->timer_coro_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->timer_mutex, NULL); mk_list_init(&th_ins->upstreams); @@ -520,7 +520,7 @@ int flb_output_thread_pool_timer_coros_size(struct flb_output_instance *ins) th_ins = th->params.data; pthread_mutex_lock(&th_ins->flush_mutex); - n = mk_list_size(&th_ins->timer_coro_list); + n = mk_list_size(&th_ins->async_timer_list); pthread_mutex_unlock(&th_ins->flush_mutex); size += n; } @@ -544,7 +544,7 @@ void flb_output_thread_pool_timer_coros_print(struct flb_output_instance *ins) th_ins = th->params.data; pthread_mutex_lock(&th_ins->timer_mutex); - flb_timer_coros_print(&th_ins->timer_coro_list); + flb_timer_coros_print(&th_ins->async_timer_list); pthread_mutex_unlock(&th_ins->timer_mutex); } } From 3d8d3268f56daed6f077da41ff45a4d2c3888b3e Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 11 Oct 2023 14:33:17 -0700 Subject: [PATCH 08/22] wip --- include/fluent-bit/flb_async_timer.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 8c48b70ddec..aded3a966a7 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -31,5 +31,7 @@ #include +int flb_out_async_timer_cleanup(struct mk_list *list); + #endif From d152dd67db084dd4ba138012ad07e4dfe96565b6 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 11 Oct 2023 16:44:55 -0700 Subject: [PATCH 09/22] output: init timer coroutine lists Signed-off-by: Wesley Pettit --- src/flb_output.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/flb_output.c b/src/flb_output.c index 4b415cd8933..99cb6c52d2a 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -717,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); From b087a0d87480fb2a98891e12598335ac99cf4f0d Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Mon, 16 Oct 2023 13:53:53 -0700 Subject: [PATCH 10/22] out_s3: add to all index error messages to be clear Signed-off-by: Wesley Pettit --- plugins/out_s3/s3.c | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index a55f82d5fde..d806e6df485 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -416,14 +416,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; } @@ -431,7 +431,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; } @@ -455,11 +455,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; @@ -1469,7 +1469,7 @@ static int s3_put_object(struct flb_s3 *ctx, const char *tag, time_t file_first_ 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; } } @@ -1612,7 +1612,7 @@ static struct multipart_upload *create_upload(struct flb_s3 *ctx, const char *ta 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; } } From 382068f1f7f62863a20279a4111316304857bfc3 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Mon, 16 Oct 2023 16:15:58 -0700 Subject: [PATCH 11/22] WIP: flb_sched_out_async_timer_cb_create and cb_flush_mutex Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_async_timer.h | 6 ++++ include/fluent-bit/flb_output.h | 10 +++--- plugins/out_s3/s3.c | 39 ++++++++-------------- plugins/out_s3/s3.h | 9 ++++- src/flb_async_timer.c | 49 ++++++++++++++++++++++++++++ src/flb_output.c | 6 ++-- src/flb_output_thread.c | 2 +- 7 files changed, 85 insertions(+), 36 deletions(-) create mode 100644 src/flb_async_timer.c diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index aded3a966a7..68d5e1ab470 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -29,9 +29,15 @@ #endif #include +#include int flb_out_async_timer_cleanup(struct mk_list *list); +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); #endif diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index 92c1b4a0810..87b116aca28 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -435,7 +435,7 @@ struct flb_output_flush { /* * stores timer coros on the async_timer_list, if the output uses them */ -struct flb_out_async_timer{ +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 */ @@ -480,7 +480,7 @@ struct flb_out_timer_coro_params { struct flb_coro *coro; /* coroutine context */ }; -extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, out_async_timer_param); +extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, async_timer_coro_params); /* * libco do not support parameters in the entrypoint function due to the @@ -575,7 +575,7 @@ static FLB_INLINE void out_async_timer_cb(void) struct flb_out_async_timer*timer_coro; - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(out_async_timer_param); + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(async_timer_coro_params); if (!params) { flb_error("[output] no timer coro params defined, unexpected"); return; @@ -663,7 +663,7 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) mk_list_add(&timer_coro->_head, &o_ins->async_timer_list); } - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(out_async_timer_param); + params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(async_timer_coro_params); if (!params) { params = (struct flb_out_timer_coro_params *) flb_calloc(1, sizeof(struct flb_out_flush_params)); if (!params) { @@ -678,7 +678,7 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) params->config = config; params->coro = coro; - FLB_TLS_SET(out_async_timer_param, params); + FLB_TLS_SET(async_timer_coro_params, params); coro->caller = co_active(); flb_coro_resume(coro); return; diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index d806e6df485..88357610ffd 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -580,7 +580,8 @@ static int cb_s3_init(struct flb_output_instance *ins, mk_list_init(&ctx->uploads); mk_list_init(&ctx->upload_queue); - pthread_mutex_init(&ctx->flush_mutex, NULL); + pthread_mutex_init(&ctx->upload_queue_mutex, NULL); + pthread_mutex_init(&ctx->cb_flush_mutex, NULL); /* Export context */ flb_output_set_context(ins, ctx); @@ -1655,7 +1656,7 @@ static void cb_s3_upload(struct flb_config *config, void *data) int ret; time_t now; - ret = pthread_mutex_trylock(&ctx->flush_mutex); + ret = pthread_mutex_trylock(&ctx->upload_queue_mutex); if (ret != 0) { /* don't block the thread, a coro is already flushing */ return; @@ -1763,7 +1764,7 @@ static void cb_s3_upload(struct flb_config *config, void *data) } } - pthread_mutex_unlock(&ctx->flush_mutex); + pthread_mutex_unlock(&ctx->upload_queue_mutex); } static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char *data, @@ -1969,7 +1970,7 @@ static void flush_startup_chunks(struct flb_s3 *ctx) } -static void timer_coro_cb(struct flb_config *config, void *data) +static void async_timer_cb(struct flb_config *config, void *data) { struct flb_s3 *ctx = data; @@ -1985,8 +1986,6 @@ static void timer_coro_cb(struct flb_config *config, void *data) static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) { struct flb_sched *sched; - struct flb_out_async_timer_cb_data *timer_data = NULL; - flb_sds_t job_name; int ret; flush_startup_chunks(ctx); @@ -1994,28 +1993,12 @@ static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) if (ctx->timer_created == FLB_FALSE) { sched = flb_sched_ctx_get(); - job_name = flb_sds_create(S3_UPLOAD_JOB_NAME); - if (!job_name) { - return; - } - - timer_data = flb_calloc(1, sizeof(struct flb_output_coro_timer_data)); - if (!timer_data) { - flb_sds_destroy(job_name); - return; - } - - timer_data->ins = ctx->ins; - timer_data->job_name = job_name; - timer_data->cb = timer_coro_cb; - timer_data->data = ctx; - - ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM, - ctx->timer_ms, flb_out_async_sched_timer_cb, timer_data, NULL); + 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_free(timer_data); - flb_sds_destroy(job_name); return; } ctx->timer_created = FLB_TRUE; @@ -2063,6 +2046,8 @@ 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, @@ -2111,6 +2096,7 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, file_first_log_time, i_ins->name); if (ret < 0) { + pthread_mutex_unlock(&ctx->cb_flush_mutex); FLB_OUTPUT_RETURN(FLB_RETRY); } @@ -2138,6 +2124,7 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk, mk_list_add(&upload_file->_head, &ctx->upload_queue); } + pthread_mutex_unlock(&ctx->cb_flush_mutex); FLB_OUTPUT_RETURN(FLB_OK); } diff --git a/plugins/out_s3/s3.h b/plugins/out_s3/s3.h index 0be62ddaa27..c04d3b5bfeb 100644 --- a/plugins/out_s3/s3.h +++ b/plugins/out_s3/s3.h @@ -160,10 +160,17 @@ struct flb_s3 { /* * 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 flush_mutex; + 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/src/flb_async_timer.c b/src/flb_async_timer.c new file mode 100644 index 00000000000..3b52e7f898b --- /dev/null +++ b/src/flb_async_timer.c @@ -0,0 +1,49 @@ +/* -*- 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 + +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) +{ + flb_sds_t job_name; + struct flb_out_async_timer_cb_data *timer_data; + + job_name = flb_sds_create(job_name); + if (!job_name) { + return; + } + + timer_data = flb_calloc(1, sizeof(struct flb_out_async_timer_cb_data)); + if (!timer_data) { + flb_sds_destroy(job_name); + 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); +} \ No newline at end of file diff --git a/src/flb_output.c b/src/flb_output.c index 99cb6c52d2a..1d3516fd233 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -40,12 +40,12 @@ #include FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); -FLB_TLS_DEFINE(struct flb_out_timer_coro_params, out_async_timer_param); +FLB_TLS_DEFINE(struct flb_out_timer_coro_params, async_timer_coro_params); void flb_output_prepare() { FLB_TLS_INIT(out_flush_params); - FLB_TLS_INIT(out_async_timer_param); + FLB_TLS_INIT(async_timer_coro_params); } /* Validate the the output address protocol */ @@ -480,7 +480,7 @@ void flb_output_exit(struct flb_config *config) if (params) { flb_free(params); } - params = FLB_TLS_GET(out_async_timer_param); + params = FLB_TLS_GET(async_timer_coro_params); if (params) { flb_free(params); } diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index bc2882d7364..d5065767a67 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -363,7 +363,7 @@ static void output_thread(void *data) if (flush_params) { flb_free(flush_params); } - timer_params = FLB_TLS_GET(out_async_timer_param); + timer_params = FLB_TLS_GET(async_timer_coro_params); if (timer_params) { flb_free(timer_params); } From 5fccd08b3f63728fac342a6cd3dbeed4a25b1e58 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:01:28 -0700 Subject: [PATCH 12/22] refactor async timers into their own code file Signed-off-by: Wesley Pettit --- include/fluent-bit/flb_async_timer.h | 154 ++++++++++++++++++- include/fluent-bit/flb_output.h | 204 +------------------------ include/fluent-bit/flb_output_thread.h | 10 +- plugins/out_s3/s3.c | 2 +- src/flb_async_timer.c | 127 ++++++++++++++- src/flb_engine.c | 5 +- src/flb_output.c | 2 +- src/flb_output_thread.c | 53 +------ 8 files changed, 294 insertions(+), 263 deletions(-) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 68d5e1ab470..6598a95f008 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -32,12 +32,164 @@ #include -int flb_out_async_timer_cleanup(struct mk_list *list); +void flb_async_timer_destroy(struct flb_out_async_timer *timer); +int flb_async_timer_cleanup(struct mk_list *list); +int 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 */ + flb_sds_t 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 (!params) { + flb_error("[output] no 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 (!params) { + 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 87b116aca28..665b325a2c4 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -432,28 +432,6 @@ struct flb_output_flush { struct mk_list _head; /* Link to flb_task->threads */ }; -/* - * 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 */ - flb_sds_t 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 */ -}; - static FLB_INLINE int flb_output_is_threaded(struct flb_output_instance *ins) { return ins->is_threaded; @@ -469,19 +447,6 @@ static FLB_INLINE void flb_output_flush_destroy(struct flb_output_flush *out_flu flb_free(out_flush); } -/* - * See below note for flb_out_flush_params - * this is equivalent for timer coroutines - */ -struct flb_out_timer_coro_params { - struct flb_out_async_timer*output_timer; /* output flush */ - struct flb_out_async_timer_cb_data *timer_data; /* callback info */ - struct flb_config *config; /* Fluent Bit context */ - struct flb_coro *coro; /* coroutine context */ -}; - -extern FLB_TLS_DEFINE(struct flb_out_timer_coro_params, async_timer_coro_params); - /* * libco do not support parameters in the entrypoint function due to the * complexity of implementation in terms of architecture and compiler, but @@ -499,6 +464,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, @@ -565,125 +531,6 @@ static FLB_INLINE void output_pre_cb_flush(void) persisted_params.config); } -/* same as above but for timer coros */ -static FLB_INLINE void out_async_timer_cb(void) -{ - struct flb_coro *coro; - struct flb_out_timer_coro_params *params; - struct flb_output_instance *o_ins; - struct flb_out_thread_instance *th_ins; - struct flb_out_async_timer*timer_coro; - - - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(async_timer_coro_params); - if (!params) { - flb_error("[output] no timer coro params defined, unexpected"); - return; - } - - coro = params->coro; - timer_coro = params->output_timer; - o_ins = params->output_timer->o_ins; - - /* Run the callback provided by the output plugin */ - params->timer_data->cb(params->config, params->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->timer_mutex); - mk_list_del(&timer_coro->_head); - mk_list_add(&timer_coro->_head, &th_ins->async_timer_list_destroy); - pthread_mutex_unlock(&th_ins->timer_mutex); - } - else { - mk_list_del(&timer_coro->_head); - mk_list_add(&timer_coro->_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*timer_coro; - 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_out_timer_coro_params *params; - struct flb_output_instance *o_ins; - - /* Custom output coroutine info */ - timer_coro = (struct flb_out_async_timer*) flb_calloc(1, sizeof(struct flb_output_timer_coro)); - if (!timer_coro) { - flb_errno(); - return; - } - - /* Create a new co-routine */ - coro = flb_coro_create(timer_coro); - if (!coro) { - flb_free(timer_coro); - return; - } - - o_ins = ctx->ins; - timer_coro->o_ins = o_ins; - timer_coro->config = config; - timer_coro->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(timer_coro); - 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->timer_mutex); - mk_list_add(&timer_coro->_head, &th_ins->async_timer_list); - pthread_mutex_unlock(&th_ins->timer_mutex); - } - else { - mk_list_add(&timer_coro->_head, &o_ins->async_timer_list); - } - - params = (struct flb_out_timer_coro_params *) FLB_TLS_GET(async_timer_coro_params); - if (!params) { - params = (struct flb_out_timer_coro_params *) flb_calloc(1, sizeof(struct flb_out_flush_params)); - if (!params) { - flb_errno(); - return; - } - } - - /* Callback parameters in order */ - params->output_timer = timer_coro; - params->timer_data = ctx; - params->config = config; - params->coro = coro; - - FLB_TLS_SET(async_timer_coro_params, params); - coro->caller = co_active(); - flb_coro_resume(coro); - return; -} - void flb_output_flush_prepare_destroy(struct flb_output_flush *out_flush); int flb_output_flush_id_get(struct flb_output_instance *ins); @@ -840,55 +687,6 @@ static inline int flb_output_coros_size(struct flb_output_instance *ins) return size; } -/* Used in engine flb_running_count */ -static inline int flb_output_timer_coros_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_output_thread_pool_timer_coros_size(ins); - } - else { - size = mk_list_size(&ins->async_timer_list); - } - - return size; -} - -static inline void flb_timer_coros_print(struct mk_list *async_timer_list) -{ - struct flb_out_async_timer*timer_coro; - 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) { - timer_coro = mk_list_entry(head, struct flb_output_timer_coro, _head); - if (timer_coro != NULL) { - flb_info("[task] output=%s still running %d %s(s)", - timer_coro->o_ins->alias, n, timer_coro->timer_data->job_name); - break; - } - } - } -} - -/* Used in engine flb_running_print */ -static inline void flb_output_timer_coros_print(struct flb_output_instance *ins) -{ - if (flb_output_is_threaded(ins) == FLB_TRUE) { - flb_output_thread_pool_timer_coros_print(ins); - } - else { - flb_timer_coros_print(&ins->async_timer_list); - } -} - static inline void flb_output_return_do(int x) { struct flb_coro *coro; diff --git a/include/fluent-bit/flb_output_thread.h b/include/fluent-bit/flb_output_thread.h index 266d835aa24..407a97187f3 100644 --- a/include/fluent-bit/flb_output_thread.h +++ b/include/fluent-bit/flb_output_thread.h @@ -89,9 +89,9 @@ struct flb_out_thread_instance { pthread_mutex_t flush_mutex; /* mutex for 'flush_list' */ /* Same as flush_mutex but for timer coros */ - struct mk_list async_timer_list; /* flush context list */ - struct mk_list async_timer_list_destroy; /* flust context destroy list */ - pthread_mutex_t timer_mutex; /* mutex for 'flush_list' */ + 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; @@ -105,8 +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_output_thread_pool_timer_coros_size(struct flb_output_instance *ins); -void flb_output_thread_pool_timer_coros_print(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); 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 88357610ffd..8a573a152e3 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -1996,7 +1996,7 @@ static void s3_flush_init(struct flb_config *config, struct flb_s3 *ctx) 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) + ctx, NULL); if (ret < 0) { flb_plg_error(ctx->ins, "Failed to create upload timer"); return; diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index 3b52e7f898b..f912e2a51f2 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -20,6 +20,35 @@ #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); +} + +int 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); + } +} + +int 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, @@ -46,4 +75,100 @@ int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int m timer_data->data = data; return flb_sched_timer_cb_create(sched, type, ms, flb_out_async_sched_timer_cb, timer_data, NULL); -} \ No newline at end of file +} + +/* 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 4d45840ac1f..af9c9667e9b 100644 --- a/src/flb_engine.c +++ b/src/flb_engine.c @@ -402,7 +402,7 @@ static int flb_running_count(struct flb_config *config) mk_list_foreach_safe(head, tmp, &config->outputs) { o_ins = mk_list_entry(head, struct flb_output_instance, _head); - n = flb_output_timer_coros_size(o_ins); + n = flb_async_timers_size(o_ins); timers = timers + n; } @@ -420,7 +420,7 @@ static void flb_running_print(struct flb_config *config) mk_list_foreach_safe(head, tmp, &config->outputs) { o_ins = mk_list_entry(head, struct flb_output_instance, _head); - flb_output_timer_coros_print(o_ins); + flb_thread_pool_async_timers_print(o_ins); } } @@ -928,6 +928,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 1d3516fd233..e5f5537ce5b 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -40,7 +40,7 @@ #include FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); -FLB_TLS_DEFINE(struct flb_out_timer_coro_params, async_timer_coro_params); +FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); void flb_output_prepare() { diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index d5065767a67..10b84246664 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -181,7 +181,7 @@ static void output_thread(void *data) struct flb_output_flush *out_flush; struct flb_out_thread_instance *th_ins = data; struct flb_out_flush_params *flush_params = NULL; - struct flb_out_timer_coro_params *timer_params = NULL; + struct flb_out_async_timer *timer_params = NULL; struct flb_net_dns dns_ctx; /* Register thread instance */ @@ -331,6 +331,7 @@ 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 && mk_list_size(&th_ins->async_timer_list) == 0) { @@ -357,6 +358,7 @@ 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); flush_params = FLB_TLS_GET(out_flush_params); @@ -444,7 +446,7 @@ int flb_output_thread_pool_create(struct flb_config *config, 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->timer_mutex, NULL); + pthread_mutex_init(&th_ins->async_timer_mutex, NULL); mk_list_init(&th_ins->upstreams); upstream_thread_create(th_ins, ins); @@ -502,53 +504,6 @@ int flb_output_thread_pool_create(struct flb_config *config, return 0; } -int flb_output_thread_pool_timer_coros_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_output_thread_pool_timer_coros_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->timer_mutex); - flb_timer_coros_print(&th_ins->async_timer_list); - pthread_mutex_unlock(&th_ins->timer_mutex); - } -} - int flb_output_thread_pool_coros_size(struct flb_output_instance *ins) { int n; From 19dff4c0c00e8edb478f6f17ace7e405b3c92eb0 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:10:04 -0700 Subject: [PATCH 13/22] Add missing imports of #include --- include/fluent-bit/flb_async_timer.h | 1 + plugins/out_s3/s3.c | 1 + src/flb_engine.c | 1 + src/flb_scheduler.c | 1 + 4 files changed, 4 insertions(+) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 6598a95f008..80cd1325a05 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -30,6 +30,7 @@ #include #include +#include void flb_async_timer_destroy(struct flb_out_async_timer *timer); diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index 8a573a152e3..fabb39d854d 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -32,6 +32,7 @@ #include #include #include +#include #include #include diff --git a/src/flb_engine.c b/src/flb_engine.c index af9c9667e9b..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 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 From 3adf2046bb5597efe2469d8b95cc0db29008e6d0 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:12:38 -0700 Subject: [PATCH 14/22] wip --- include/fluent-bit/flb_async_timer.h | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 80cd1325a05..aff5da788cf 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -82,8 +82,8 @@ static FLB_INLINE void out_async_timer_cb(void) async_timer = (struct flb_out_async_timer *) FLB_TLS_GET(async_timer_coro_params); - if (!params) { - flb_error("[output] no timer coro params defined, unexpected"); + if (!async_timer) { + flb_error("[output] no async timer coro params defined, unexpected"); return; } @@ -174,7 +174,7 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) 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 (!params) { + if (!timer_thread_key) { flb_errno(); return; } From de4aa1bffd01585235d75316ed73ea7c5991fa35 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:14:20 -0700 Subject: [PATCH 15/22] wip --- src/flb_output_thread.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index 10b84246664..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); From c74dc5f8523dd4b0c723f39e9cdc9ffe69caac16 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:46:13 -0700 Subject: [PATCH 16/22] wip --- src/flb_async_timer.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index f912e2a51f2..2f3605a5fdd 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -19,6 +19,7 @@ #include #include +#include void flb_async_timer_destroy(struct flb_out_async_timer *timer) { From 9a5fc2afe61d93089375d55c3f990f608911e74f Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 16:51:05 -0700 Subject: [PATCH 17/22] wip --- include/fluent-bit/flb_async_timer.h | 1 - src/flb_async_timer.c | 1 - 2 files changed, 2 deletions(-) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index aff5da788cf..397991f0e43 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -28,7 +28,6 @@ #define _GNU_SOURCE #endif -#include #include #include diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index 2f3605a5fdd..d25de075764 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -18,7 +18,6 @@ */ #include -#include #include void flb_async_timer_destroy(struct flb_out_async_timer *timer) From 3c27f4f0e5437736438ec45995f80ca45b49e318 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Tue, 17 Oct 2023 17:03:07 -0700 Subject: [PATCH 18/22] Add flb_async_timer.c to CMakelists.txt Signed-off-by: Wesley Pettit --- src/CMakeLists.txt | 1 + 1 file changed, 1 insertion(+) 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 From 26de4ea9a7a4888f65a3ebe22681fa7d2ec4b6ed Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 18 Oct 2023 14:54:54 -0700 Subject: [PATCH 19/22] wip --- include/fluent-bit/flb_async_timer.h | 2 +- src/flb_async_timer.c | 9 +-------- 2 files changed, 2 insertions(+), 9 deletions(-) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 397991f0e43..d55ad1c7b83 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -63,7 +63,7 @@ struct flb_out_async_timer { */ struct flb_out_async_timer_cb_data { struct flb_output_instance *ins; /* associate coro with this output instance */ - flb_sds_t job_name; /* used on engine shutdown, print pending "custom" jobs */ + 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 */ }; diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index d25de075764..7494772aa75 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -43,7 +43,7 @@ int 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) { + 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); } @@ -55,17 +55,10 @@ int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int m void (*async_cb)(struct flb_config *, void *), void *data, struct flb_sched_timer **out_timer) { - flb_sds_t job_name; struct flb_out_async_timer_cb_data *timer_data; - job_name = flb_sds_create(job_name); - if (!job_name) { - return; - } - timer_data = flb_calloc(1, sizeof(struct flb_out_async_timer_cb_data)); if (!timer_data) { - flb_sds_destroy(job_name); return; } From 93b2f8eb1f35ebde01112df72d13a813f3f573cd Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 18 Oct 2023 14:58:11 -0700 Subject: [PATCH 20/22] wip --- include/fluent-bit/flb_async_timer.h | 4 ++-- src/flb_async_timer.c | 6 ++++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index d55ad1c7b83..6002e9552a5 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -33,8 +33,8 @@ void flb_async_timer_destroy(struct flb_out_async_timer *timer); -int flb_async_timer_cleanup(struct mk_list *list); -int flb_output_async_timer_cleanup(struct flb_config *config); +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, diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index 7494772aa75..b8dc078ffd4 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -18,6 +18,8 @@ */ #include +#include +#include #include void flb_async_timer_destroy(struct flb_out_async_timer *timer) @@ -27,7 +29,7 @@ void flb_async_timer_destroy(struct flb_out_async_timer *timer) flb_free(timer); } -int flb_async_timer_cleanup(struct mk_list *destroy_list) +void flb_async_timer_cleanup(struct mk_list *destroy_list) { struct flb_out_async_timer *async_timer; struct mk_list *tmp; @@ -38,7 +40,7 @@ int flb_async_timer_cleanup(struct mk_list *destroy_list) } } -int flb_output_async_timer_cleanup(struct flb_config *config) +void flb_output_async_timer_cleanup(struct flb_config *config) { struct flb_output_instance *o_ins; struct mk_list *tmp; From 5504b1e490b03f5d929a47fde8ac2941c00d51b4 Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 18 Oct 2023 16:00:30 -0700 Subject: [PATCH 21/22] wip --- include/fluent-bit/flb_output.h | 1 + 1 file changed, 1 insertion(+) diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index 665b325a2c4..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 From 6605425399b4ce8b3c524536d914061fdabc8cdc Mon Sep 17 00:00:00 2001 From: Wesley Pettit Date: Wed, 18 Oct 2023 16:01:17 -0700 Subject: [PATCH 22/22] wip --- include/fluent-bit/flb_async_timer.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 6002e9552a5..3d888fdc7c2 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -30,6 +30,8 @@ #include #include +#include +#include void flb_async_timer_destroy(struct flb_out_async_timer *timer);