Make stage_size & stage computation thread safe

[hlakshmi]

Which issue(s) this PR fixes:
#2712
Fixes #

What this PR does / why we need it:
When writing chunks to buffer, the stage_size computation can be thread-unsafe as mentioned in the bug which can result in BufferOverflowErrors even though the buffer is not actually full. This happened on a live long running fluentd. The method buffer.write is supposed to write to a chunk atmost once and hence stage_size should be added for a chunk only once but due to write_step_by_step the block which updates the value of staged_bytesize may be called multiple times and the chunk can be rollbacked but the value of staged_bytesize is not reverted causing the stage_size to be more than the actual value.

This PR addresses this issue.

Docs Changes:

Release Note:

[repeatedly]

Thanks!. We will check issue and patch later.

[hlakshmi]

@repeatedly Can we get this merged into the upcoming release if it looks good as this is causing a critical issue of data loss. We have tested this in large scale environment and the fix seems to work. Thanks!

[ganmacs]

Sorry for the delay. I added a comment.

[ganmacs]

I think the following situation can happen. what do you think of this?

1. staged_bytesizes_by_chunk stores chunk1
2. rollback happens in write_step_by_step (ShouldRetry raises)
3. another thread euqueues chunk1 before this thread enters into

   fluentd/lib/fluent/plugin/buffer.rb

   Line 673 in 065751f

   chunk = get_next_chunk.call

4. create new chunk(chunk2) in

   fluentd/lib/fluent/plugin/buffer.rb

   Line 663 in 065751f

   synchronize{ @stage[metadata] ||= generate_chunk(metadata).staged! }

5. staged_bytesizes_by_chunk stores chunk2 (still remtains chunk1 in staged_bytesizes_by_chunk)

[hlakshmi]

@ganmacs Steps (3) can't happen because enqueue_chunk can't get a lock and waits until the lock is released by the thread which is writing to the chunk.

The lock is acquired here: https://github.com/fluent/fluentd/blob/master/lib/fluent/plugin/buffer.rb#L279
And released here:
https://github.com/fluent/fluentd/blob/master/lib/fluent/plugin/buffer.rb#L319

So enqueue_chunk can happen only after the commit is done.
Hope this helps.

[ganmacs]

Make sense. You're right. other thread can get the chunk but can't enter

fluentd/lib/fluent/plugin/buffer.rb

Line 674 in 065751f

chunk.synchronize do

.

[ganmacs]

👏

[ganmacs]

Make sense. You're right. other thread can get the chunk but can't enter

fluentd/lib/fluent/plugin/buffer.rb

Line 674 in 065751f

chunk.synchronize do

.

[repeatedly]

Thanks!