S3 Stability Refactor + Scheduled timer coroutines => AWS Distro PR #27

PettitWesley · 2023-09-26T20:51:49Z

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

Example configuration file for the change
Debug log output from testing the change

Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

Attached local packaging test output showing all targets (including any new ones) build.

Documentation

Documentation required for this feature

Backporting

Backport to latest stable release.

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.

Signed-off-by: Wesley Pettit <[email protected]>

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 <[email protected]>

Signed-off-by: Wesley Pettit <[email protected]>

matthewfala

First pass just observing your code. I left some comments for my understanding which will be deleted when I understand the code better and can make more helpful comments.

So far it's looking good in terms log logic. In terms of organization, I suppose Leonardo would need to weigh into that. I'm not fully understanding the single thread vs multi workers since I haven't read that part closely. Also need to still read the timer coro deletion logic section once added to the delete list.

matthewfala · 2023-09-29T22:24:56Z

plugins/out_s3/s3.c

    int ret;
    void *payload_buf = NULL;
    size_t payload_size = 0;
    size_t preCompress_size = 0;
+    time_t file_first_log_time = time(NULL);


Clay's PR is back from the dead? This looks familiar. I think we previously removed this commit since it was causing instability right?

Yes and the goal of the S3 refactor is to bring back all the S3 changes/features and redo and retest them.

matthewfala · 2023-09-29T22:26:28Z

plugins/out_s3/s3.c

@@ -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)) {


How is the retry timer logic changing?

matthewfala · 2023-09-29T22:32:55Z

src/flb_output.c

@@ -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);


Do you need to set the FLB_TLS timer_coro_params to NULL to avoid a double free?

No. This only occurs once during shutdown.

matthewfala · 2023-09-30T00:16:21Z

plugins/out_s3/s3.c

@@ -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) {


Note: this section get's the file_first_log_time from (1) a log with a timestamp, (2) archived timestamp (3) current time.

matthewfala · 2023-09-30T00:36:21Z

plugins/out_s3/s3.c

-    /* 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 "


Logic change, does this explain it correctly?:

Unify upload failure deletion logic to the callback function

Just buffer the chunk here.

Callback handles all uploads and failure handling

Flush just buffers

matthewfala · 2023-09-30T00:41:25Z

plugins/out_s3/s3.c

+        ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM,
+                                        ctx->timer_ms, flb_output_coro_timer_cb, timer_data, NULL);


Todo: Check if this function is run multiple times if timer_data will be overwritten.

matthewfala · 2023-09-30T01:07:57Z

plugins/out_s3/s3.c

+    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);


This is threadsafe right? Like, what if two workers add to the list at the same time?

OH I was thinking we'd still only support 1 worker for S3... but yea if I add a lock here then I think we can probably support multiple... the timer coros have a lock for the ready to upload chunk list...

All of the code that interacts with any list in cb_s3_flush needs to be protected.

The chunk list, the upload list, these can't be iterated over if another thread is modifying them.

Added pthread_mutex_lock(&ctx->cb_flush_mutex); at beginning of cb_s3_flush

matthewfala · 2023-09-30T01:17:59Z

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)


Used exclusively for graceful shutdowns

matthewfala · 2023-09-30T02:48:05Z

include/fluent-bit/flb_output.h

+        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);


Todo: understand destroy list logic better

Ah yea... I forgot to implement deletion, its in the design: fluent#7466

matthewfala · 2023-09-30T02:48:58Z

include/fluent-bit/flb_output.h

+
+    FLB_TLS_SET(timer_coro_params, params);
+    coro->caller = co_active();
+    flb_coro_resume(coro);


It appears that this will begin the callback from the start. From this point on the network events will take over control

Yea that's the intention

matthewfala

Review 2/3 - This is a closer review of the timer logic (did not review s3 changes in detail): Looks good for the most part in terms of logic! Missing cleanup/timer delete logic. Wrote some organizational and data structure recommendations.

matthewfala · 2023-10-03T22:42:08Z

plugins/out_s3/s3.c

+        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);


Amazing that this works. Pretty complicated logic.
This is slightly akward. Most of the timer is exposed in the fluent bit core lib. However an integral part of the timer functionality (creating the async timer) is implemented at the plugin level.

I would recommend extracting to a function such as:
flb_sched_async_timer_cb_create(ins, job_name, async_cb, data, etc)

Would you please follow up on this comment?

Sure I can do this

matthewfala · 2023-10-03T23:04:53Z

include/fluent-bit/flb_output.h

+    FLB_TLS_SET(timer_coro_params, params);
+    coro->caller = co_active();
+    flb_coro_resume(coro);
+    return;


Why do you have a return at the bottom of your function? If it's best practices then that's good, but I'll have to adjust how I write void functions in the future.

Oops, I think its not best practice?

matthewfala · 2023-10-03T23:15:45Z

include/fluent-bit/flb_output.h

+    }
+
+    /* timer coro is complete; yield back to caller/control code */
+    flb_coro_yield(coro, FLB_TRUE);


Delete logic still to be implemented likely in engine and output_thread event loop clean up section.

Ah yea... I forgot to implement deletion, its in the design: fluent#7466

matthewfala · 2023-10-04T00:04:10Z

plugins/out_s3/s3.h

+     * but modifying the pending chunk and upload lists, and deleting S3 store files needs
+     * to be concurrent safe
+     */
+    pthread_mutex_t flush_mutex;


This name is slightly confusing since it shares the name of the core flush_mutex on the output thread. Consider renaming to upload_queue_mutex

matthewfala · 2023-10-04T00:10:55Z

include/fluent-bit/flb_output.h

+/*
+ * stores timer coros on the timer_coro_list, if the output uses them
+ */
+struct flb_output_timer_coro {


The data structures look good! But in terms of organization, would you consider renaming all flb_output_timer_... to flb_async_timer_... and moving it to a new core file called flb_async_timer.x.

This will highlight that your functions operate as a layer on top of the existing coroutine logic, and does not have interweaved logic. (calls existing coro functions, existing coro and flush functions do not call the new functions)

Ok. I think output is still needed in the name since in the future we could have timer coros for other plugin types as well I think.

So I think I'd do flb_async_timer.c and flb_out_async_timer...

I think generally the convention is to prefix with flb_<file_name>_<function_name>. So the file should either be called flb_out_async_timer.c/h or the function should be called flb_async_timer_x (see flb_regex.h)

Is "out" used to indicate that the timer is only intended to be used in the output plugin? I believe the timer can be scheduled anywhere? That might be incorrect actually, or is it?

matthewfala · 2023-10-04T01:08:53Z

include/fluent-bit/flb_output_thread.h

+    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' */


This review isn't for style but incase we forget later, the comments here may be copied from flush_... and pends update

/* flush context list */

matthewfala · 2023-10-04T01:14:01Z

include/fluent-bit/flb_output.h

+ * 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 {


Recommended rename to flb_output_coro_timer_initialization_data or better flb_async_timer_initialization_data

How about flb_out_async_timer_cb_data since its for the callback..

It would be great for this function name to include the word initialization to differentiate it from the user callback and user provided data void *data;. This datastructure is explicitly used to initialize the async timer, but is not for the user provided callback. Also to differentiate from the coro cb async_timer_cb

Name should align with:
https://github.com/PettitWesley/fluent-bit/pull/27/files#r1345008308

These two pointers go together - they are responsible for initializing an async_timer:

flb_async_timer_initialization_data

flb_async_timer_initialization_cb

Likewise we have for the user provided callback the same convention:

cb

data

matthewfala · 2023-10-04T01:16:11Z

include/fluent-bit/flb_output.h

+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 */


Consider renaming to async_cb to differentiate from the standard scheduled timer cb

matthewfala · 2023-10-04T01:46:15Z

src/flb_engine.c

+    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;
+    }


Consider extracting function to flb_async_timer_running_count(config) for style. No strong convictions on this one.

matthewfala · 2023-10-04T01:54:59Z

src/flb_engine.c

+    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);
+    }


Consider extracting function to flb_async_timer_running_print(config) for style. No strong convictions on this one.

Signed-off-by: Wesley Pettit <[email protected]>

PettitWesley · 2023-10-19T21:25:51Z

@matthewfala So I like the idea of storing the async timer lists on the scheduler struct, which could make it generic and not associated with outputs.

However, I am seeing a problem this idea.

I have methods flb_thread_pool_async_timers_size and flb_thread_pool_async_timers_print : https://github.com/PettitWesley/fluent-bit/pull/27/files#diff-27b244cc7913bcd78c3395ca61c3f13822acfdf90e384a171b7eee0c73a01f6aR124

These need to collect all active timers. This is used to print timers on shutdown that are still running, and to wait the graceful shutdown until they are done.

The sched struct right now is one per event loop AFAICT. SO one in the engine, one in each output thread. But there's no reference between the output thread and the sched instance. There's no centralized list of all sched instances. I need something like that in order for these functions to work. Basically, this has to be tied to the fact that its running on some output worker event loop. I think the simplest solution would be to link the thread that the sched is for together- by adding a reference on the thread to its sched.

PettitWesley · 2023-10-20T03:31:12Z

There are two key ways I need to track async timer coros:

On shutdown we must track how many pending coros there are, and right now there’s no way to grab all of the sched structures to count these new coros. A sched is associated with an event loop, so both the engine and each worker has one.
S3 output needs to create one scheduled timer in each of its output threads, so that it can get a coro every interval N in each worker thread so each worker can perform upload work (including doing gzip compression, which takes a lot of CPU and is the main reason customers ask for multi-workers).

For Problem 1, I want to propose:

The output thread has a reference to its sched. This means you can iterate over the outputs and each's output threads and then get each sched and get each timer coro. Like this: 6d3ab25

PettitWesley · 2023-10-23T16:01:58Z

S3 output needs to create one scheduled timer in each of its output threads, so that it can get a coro every interval N in each worker thread so each worker can perform upload work (including doing gzip compression, which takes a lot of CPU and is the main reason customers ask for multi-workers).

With flb_output_thread_instance_get the output can get the pointer to the current worker thread that its running in. It can store these addresses in an array same length as number of workers. Each flush, it checks if the worker ID is new and not in its list, if so, it starts a new scheduled timer in that worker.

PettitWesley · 2023-10-23T17:29:29Z

Eduardo, wants it to be called flb_coro_timer

PettitWesley · 2023-10-26T20:45:01Z

src/flb_async_timer.c

+
+    mk_list_foreach(head, &tp->list_threads) {
+        th = mk_list_entry(head, struct flb_tp_thread, _head);
+        if (th->status != FLB_THREAD_POOL_RUNNING) {


One interesting thing, FLB_THREAD_POOL_RUNNING just tells you the thread has been started, the FLB_THREAD_POOL_STOPPED value is never set.

github-actions · 2024-01-25T02:12:20Z

This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.

PettitWesley had a problem deploying to pr September 26, 2023 20:52 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 20:52 — with GitHub Actions Inactive

PettitWesley had a problem deploying to pr September 26, 2023 21:11 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 21:11 — with GitHub Actions Inactive

PettitWesley had a problem deploying to pr September 26, 2023 22:26 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 22:26 — with GitHub Actions Inactive

PettitWesley force-pushed the s3-refactor-timer-coros-awspr branch from aec4756 to 19892e0 Compare September 26, 2023 22:27

PettitWesley had a problem deploying to pr September 26, 2023 22:27 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 22:27 — with GitHub Actions Inactive

PettitWesley added 4 commits September 26, 2023 15:42

output: add support for scheduled timer jobs with coroutines

5562a2b

Signed-off-by: Wesley Pettit <[email protected]>

output_thread: add support for scheduled timer jobs with coroutines

fdda9d5

Signed-off-by: Wesley Pettit <[email protected]>

engine: print and track scheduled timer coroutines on shutdown

a4407b6

Signed-off-by: Wesley Pettit <[email protected]>

PettitWesley force-pushed the s3-refactor-timer-coros-awspr branch from 19892e0 to c73fef0 Compare September 26, 2023 22:48

PettitWesley changed the base branch from 2_31_10_all_cherry_picks to 2_31_12_all_cherry_picks September 26, 2023 22:48

PettitWesley changed the base branch from 2_31_12_all_cherry_picks to 2.31.12-all-cherrypicks September 26, 2023 22:49

output: fix variable name in output_pre_timer_cb

5202199

Signed-off-by: Wesley Pettit <[email protected]>

PettitWesley had a problem deploying to pr September 26, 2023 23:25 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 23:25 — with GitHub Actions Inactive

out_s3: remove unneeded info message

814c8c7

Signed-off-by: Wesley Pettit <[email protected]>

PettitWesley had a problem deploying to pr September 26, 2023 23:28 — with GitHub Actions Failure

PettitWesley had a problem deploying to pr September 26, 2023 23:29 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr September 26, 2023 23:29 — with GitHub Actions Inactive

matthewfala reviewed Sep 30, 2023

View reviewed changes

matthewfala suggested changes Oct 4, 2023

View reviewed changes

PettitWesley had a problem deploying to pr October 17, 2023 23:51 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 17, 2023 23:51 — with GitHub Actions Inactive

Add flb_async_timer.c to CMakelists.txt

3c27f4f

Signed-off-by: Wesley Pettit <[email protected]>

PettitWesley had a problem deploying to pr October 18, 2023 00:03 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 18, 2023 00:03 — with GitHub Actions Inactive

wip

26de4ea

PettitWesley had a problem deploying to pr October 18, 2023 21:55 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 18, 2023 21:55 — with GitHub Actions Inactive

wip

93b2f8e

PettitWesley had a problem deploying to pr October 18, 2023 21:58 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 18, 2023 21:58 — with GitHub Actions Inactive

wip

5504b1e

PettitWesley had a problem deploying to pr October 18, 2023 23:00 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 18, 2023 23:00 — with GitHub Actions Inactive

wip

6605425

PettitWesley had a problem deploying to pr October 18, 2023 23:01 — with GitHub Actions Failure

PettitWesley temporarily deployed to pr October 18, 2023 23:01 — with GitHub Actions Inactive

PettitWesley commented Oct 26, 2023

View reviewed changes

github-actions bot added the Stale label Jan 25, 2024

PettitWesley removed the Stale label Mar 12, 2024

		ret = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_PERM,
		ctx->timer_ms, flb_output_coro_timer_cb, timer_data, NULL);

S3 Stability Refactor + Scheduled timer coroutines => AWS Distro PR #27

Are you sure you want to change the base?

S3 Stability Refactor + Scheduled timer coroutines => AWS Distro PR #27

Conversation

PettitWesley commented Sep 26, 2023

matthewfala left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewfala left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewfala Oct 6, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewfala Oct 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

PettitWesley commented Oct 19, 2023

PettitWesley commented Oct 20, 2023

PettitWesley commented Oct 23, 2023

PettitWesley commented Oct 23, 2023

Choose a reason for hiding this comment

github-actions bot commented Jan 25, 2024

matthewfala left a comment •

edited

Loading

matthewfala Oct 6, 2023 •

edited

Loading

matthewfala Oct 7, 2023 •

edited

Loading