Add support for concurrent append in out_file

[fujimotos]

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

What this PR does / why we need it:

The basic problem is that Fluentd can run a plugin in multiple workers,
but there is no mechanism that allows workers to coordinate each other.

For this reason, plugins are often suffers race condition issues on multi-worker
mode (e.g. #3805, fluent/fluent-plugin-s3#391).

This can be simply resolvable by implementing a inter-worker lock.
Here is how I implemented it:

• ServerModule prepares a lockfile directory and share that directory
  path with its workers (through ENV).

• Now, Workers can define critical sessions by creating a lockfile
  in that directory.

• The actual lock was held by flock, so it's automatically freed by
  OS even when Worker gets killed in the middle of critical session.

WIth this merged, we will be able to run out_file without race
conditions among workers.

Docs Changes:

Required

Release Note:

[daipom]

Thanks for the implementation!
I think adding per-worker locking as a common feature of Fluentd would be a very elegant solution.
I will see this in detail.

[daipom]

The overall implementation direction is fine for me.
I have commented on the points I noticed.

I have confirmed that the current implementation solves this problem in my environment.

[daipom]

I have confirmed that this works correctly on Windows too.

• Windows10 (10.0.19043)

The results of the reproducing way in #3805 are as follows. (3 times each)

• This branch: d31631e

0 / 7706 lines are broken.
0 / 18388 lines are broken.
0 / 9933 lines are broken.

• Fluentd v1.15.0: 79ead40

3 / 6786 lines are broken.
11 / 12245 lines are broken.
10 / 12283 lines are broken.

[fujimotos]

Unit test added.

[ashie]

Hmm, behavior of GitHub seems strange.
I just merged this by merge button and it seems correctly applied to master branch, but this PR isn't closed yet.

[ashie]

Probably it's caused by this incident of GitHub: https://www.githubstatus.com/incidents/dsj40hwxnldc

[ashie]

The merged commits aren't broken, so I close this PR.
Sorry for that.

[fujimotos]

Testing with NFS storages

I've set up a NFS server and spent this morning testing this fix.

I configured Fluentd to perform an intensive concurrent writes, and run
the same test using v1.15.0 and master HEAD (20 times each). Here is
the result:

#	0	1	2	3	4	5	6	7	8	9
v1.15.0	o	x	o	x	o	o	x	o	x	x
(cont)	x	o	o	o	o	o	o	o	x	o
master	o	o	o	o	o	o	o	o	o	o
(cont)	o	o	o	o	o	o	o	o	o	o

o = No corruption, x = Corrupted output

To put it short, v1.15.0 corrupted the output 7 times out of 20, while
the master HEAD never did. So I believe this fix works fine not only
on local disks, but also on NFS storages.

Test Procedure Details

System

• Ubuntu 20.04 on AWS
• Created NFSv4 server on /opt/nfs.
• NFS Server was nfs-kernel-server v1:1.3.4-2.5ubuntu3.4

Procedure

1. Launch Fluentd with bundle exec fluentd -c fluent.conf
2. Run send-record.sh
3. Shutdown Fluentd
4. Run check.py on /opt/nfs/test.*.log

fluentd.conf

<system>
  workers 8
</system>
<source>
  @type forward
</source>
<match **>
  @type file
  path "/opt/nfs/test.log"
  append true
  <buffer>
    @type "file"
    path "buffer/"
    flush_mode interval
    flush_interval 60m
    flush_at_shutdown true
  </buffer>
</match>

send-records.sh

msg=`yes 1 | head -n 65536 | tr -d "\n"`
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
echo "{\"message\": \"$msg\"}" | bundle exec fluent-cat test.log &
wait

check.py

import sys
import json

i = 0
for line in sys.stdin:
    try:
        json.loads(line.strip().split(maxsplit=2)[2])
    except Exception:
        print(i, line)
        raise
    i += 1
print(i)