Properly stop logger during (re)connect failure

[akerouanton]

PR #77 introduced a new parameter named ForceStopAsyncSend. It can be
used to tell the logger to not try to send all the log messages in its
buffer before closing. Without this parameter, the logger hangs out
whenever it has logs to write and the target Fluentd server is down.

But this new parameter is not enough: the connection is currently lazily
initialized when the logger receive its first message. This blocks the
select reading messages from the queue (until the connection is ready).
Moreover, the connection dialing uses an exponential back-off retry.

Because of that, the logger won't look for messages on stopRunning
channel (the channel introduced by PR #77), either because it's
blocked by the Sleep used for the retry or because the connection
dialing is waiting until dialing timeout (eg. TCP timeout).

To fix these edge cases, the time.Sleep() used for back-off retry has
been transformed into a time.After() and a new stopAsyncConnect channel
has been introduced to stop it. Moreover, the dialer.Dial() call used
to connect to Fluentd has been replaced with dialer.DialContext() and a
cancellable context is now used to stop the call to that method.

Finally, the Fluent.run() method has been adapted to wait for both new
messages on f.pending and stop signal sent to f.stopRunning channel.
Previously, both channels were selected independently, in a procedural
fashion.

This fix is motivated by the issue described in: moby/moby#40063.

[tagomoris]

The motivation is reasonable, so I'm positive to fix that problem.
This kind of patch requires time to review, so I'll be able to review this change tomorrow or early next week.
But anyway, CI is now failing. @NiR- Could you take care of it?

[akerouanton]

But anyway, CI is now failing. @NiR- Could you take care of it?

CI is fixed but actually I'm still doing some manual testing to make sure everything is ok. I'll try to add more test cases 😉

[tagomoris]

@NiR- Great! Please let me know when you finished that.

[angulito]

Any news on this? I'm experiencing the same issue with docker v19.03.6-ce that is happening moby/moby#40063. Thank you!

[akerouanton]

@angulito We first want to merge #83 which is introducing a new way of testing edge cases like the one identified and fixed by this PR.

The new test cases and the fix were first submitted in a single PR, but since the diff was quite large and important, I splitted it in two PRs to make it less difficult for the repo's maintainer(s) to review and accept these changes.

Hopefully this will be included in the next major docker release. In the mean time, I created a logger plugin including this fix. It's working like the original fluentd logger and it's available in akerouanton/fluentd-async-logger.

[angulito]

Thank you for working hard on it! @akerouanton

[akerouanton]

Hey @tagomoris, I took a look at this PR in the past days and I finally found out why tests were failing on the CI. I also overhauled the previous implementation to make the code substantially simpler.

CI was failing because of tests with RequestAck option enabled: the ack hashes contain a randomly generated uint and despite the generator being implicitly seeded with 1 (as specified here), the ack hashes were not stable. I tried to explicitly seed the generator but with no luck. Thus, I decided to introduce a new randomGenerator variable and initialize it with rand.Uint64. This variable is changed in tests to use a function that always return 1. This way, we're sure the ack hashes are stable across time and machines.

On the overhaul part, I decided to remove the newly introduced stopAsyncConnect channel and use a cancellable context instead. This context is passed from newWithDialer() to run() -> write() -> connectOrRetry() -> connect() -> DialWithContext(). In this way, as soon as the context got cancelled, any in-progress dialing got stopped and the errIsClosing error is propagated up to run() by early terminating retriable code paths (connectOrRetry() and write()) . Moreover, in this way, any subsequent calls to write() will fail the same way (for instance if a log is dispatched through f.pending before f.stopRunning closing is dispatched), making the code more robust to channel dispatching order in run(). I also removed the drainEvents condition run() since returning from that function and closing f.pending is already enough to discard queued messages.

This PR is ready to be re-reviewed 🙂

[tagomoris]

I'll try to find time to review this change this or the next week.

[tagomoris]

How about replacing this command with make this "var" to replace the random generator in tests and write the content of the above comment in the test file?
The content is a little complex, and looks almost only about testing.

[tagomoris]

I took a glance at this, and need more time.

[tagomoris]

I reviewed the updated change. The clear problem is, this code will acquire the lock of connection during writing messages and waiting for the ack message. This causes terrible performance degradation if the ack option is enabled. Even without enabling the ack option, the critical section for writing messages could cause performance problems.

And I found another problem (this is not introduced by this change) that arriving ack messages may be disordered, but the current implementation doesn't consider such cases. 🤔

[jchanam]

Hi!

I'd like to know if you were able to move forward with this issue. We're still facing this problem described in moby/moby#40063

Thanks!

[rothn]

We are having this issue as well. Do we know if there's anything preventing a merge?

[hakashya]

Hey @tagomoris,
To confirm, the problems that are preventing the merge are:

1. Message writing critical section causing performance degradation.
2. Unordered ack messages not being handled.

Hey @akerouanton,
I am just wondering if you were working on it?

[tagomoris]

@hakashya Unordered ack messages are not the problem of this change. Of course, it's good if we can solve it.

[PettitWesley]

@akerouanton @tagomoris Hi, I'm Wesley, from the AWS containers team, and I'm also the AWS maintainer of fluent/fluent-bit. We need to get this change out since it fixes the linked Moby/Docker issue.

My proposal is that I will take over this fix. I will preserve @akerouanton's commit so that they get credit. And then add commits on top of it as needed to get this PR to a merge-able state.

To confirm, the issues blocking merge are:

1. Rebase.
2. "Message writing critical section causing performance degradation." I need to read through this code and try to understand it... can someone point me to the piece of the code where this issue is?

If I fix those, we can merge it?

I will begin work on this on Wednesday September 29th.

[tagomoris]

I'm totally ok to merge the change written by 2 contributors (it's almost equal to independent 2 pull-reqs) if it can solve problems.

[akerouanton]

About the overview of the change: I'm not a big fan of long anonymous functions (writer in write()). In my opinion, it suffers readability.

I can rename the current write() method to writeOrRetry() and move the content of the writer closure into a new write() method if you want, like what I did with connect() and connectOrRetry() 🙂

[akerouanton]

Btw, I did not move f.conn.SetWriteDeadline() into connect() method like I proposed in the patch I posted as a comment 2 days ago. However setting the deadline in the write() method could be the source of a bug in sync mode. For instance:

• 1st goroutine send a message which takes a long time to send (eg. tcp exponential backoff got triggered)
• 2nd goroutine send a message whereas 1st message is still in-flight and set the new write deadline, impacting the deadline of the 1st message
• 3rd goroutine send a message and sets the new deadline, again impacting the previous in-flight messages
• etc...

This is due to how SetWriteDeadline() works:

// SetWriteDeadline sets the deadline for future Write calls
// and any currently-blocked Write call.
// Even if write times out, it may return n > 0, indicating that
// some of the data was successfully written.
// A zero value for t means Write will not time out.
SetWriteDeadline(t time.Time) error

Of course this is limited to usecases where many goroutines are used to send logs, but I prefer to point it out.

[akerouanton]

I moved the body of the writer() anynomous function to write() and the retry logic to writeOrRetry(). Also, I replaced time.After() by time.NewTimer() with a defered call to ensure this underlying channel is closed and garbage collected (cf. https://pkg.go.dev/time#After).

[tagomoris]

My 👀 couldn't find any clear problems about locking/consistency on this change. Great work!
I left some comments about comments and naming.

[tagomoris]

👍

[tagomoris]

LGTM!
Let me merge this change and tag v1.7.0 to let users choose this fix only (or v1.6.3 for other recent changes).

[tagomoris]

Done. Thank you for the great work!