Repository: moby/moby (language: go)

Issue:
Docker daemon crashes if fluentd daemon is gone and buffer is full
<!--
If you are reporting a new issue, make sure that we do not have any duplicates
already open. You can ensure this by searching the issue list for this
repository. If there is a duplicate, please close your issue and add a comment
to the existing issue instead.

If you suspect your issue is a bug, please edit your issue description to
include the BUG REPORT INFORMATION shown below. If you fail to provide this
information within 7 days, we cannot debug your issue and will close it. We
will, however, reopen it if you later provide the information.

For more information about reporting issues, see
https://github.com/docker/docker/blob/master/CONTRIBUTING.md#reporting-other-issues

---------------------------------------------------
GENERAL SUPPORT INFORMATION
---------------------------------------------------

The GitHub issue tracker is for bug reports and feature requests.
General support can be found at the following locations:

- Docker Support Forums - https://forums.docker.com
- IRC - irc.freenode.net #docker channel
- Post a question on StackOverflow, using the Docker tag

---------------------------------------------------
BUG REPORT INFORMATION
---------------------------------------------------
Use the commands below to provide key information from your environment:
You do NOT have to include this information if this is a FEATURE REQUEST
-->

Docker daemon crashes if 

- container is configured to log to fluentd
- fluentd server shuts down (network problem or planned shutdown)
- buffer overflows

<!--
Briefly describe the problem you are having in a few paragraphs.
-->

1. Configure logstash with fluentd input on some host
2. run 
```sh
docker run -d --log-driver=fluentd --log-opt fluentd-address=172.17.X.X:4000 --log-opt tag="test" --log-opt fluentd-buffer-limit=10KB --log-opt fluentd-max-retries=2 --name test1 --rm busybox /bin/sh -c 'yes "crashme" '
```
Yes, limits are on purpose low to expose Docker behavior.

3. make sure that Logstash receives messages.
4. stop Logstash
5. on docker host `watch 'docker ps'` shows in matter of seconds that docker daemon is not available.

**Describe the results you received:**

Dockerd process crashed with segfault

```
....repeating lines
time="2017-04-12T17:01:47.079303891Z" level=error msg="Failed to log msg \"crashme\" for logger fluentd: fluent#appendBuffer: Buffer full, limit 10240"
time="2017-04-12T17:01:47.079324142Z" level=error msg="Failed to log msg \"crashme\" for logger fluentd: fluent#appendBuffer: Buffer full, limit 10240"
panic: fluent#reconnect: failed to reconnect!
goroutine 29272 [running]:
panic(0x1630320, 0xc421062010)
/usr/local/go/src/runtime/panic.go:500 +0x1a1
github.com/docker/docker/vendor/github.com/fluent/fluent-logger-golang/fluent.(*Fluent).reconnect(0xc42096a790)
/root/rpmbuild/BUILD/docker-ce/.gopath/src/github.com/docker/docker/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go:276 +0xf8
created by github.com/docker/docker/vendor/github.com/fluent/fluent-logger-golang/fluent.(*Fluent).send
/root/rpmbuild/BUILD/docker-ce/.gopath/src/github.com/docker/docker/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go:290 +0x136
```

**Describe the results you expected:**

Container should terminate, but dockerd should stay alive. Otherwise it seems as very easy DoS to crash docker on any host you have access.

**Additional information you deem important (e.g. issue happens only occasionally):**
100% reproducible with settings above. 

**Output of `docker version`:**

```
Client:
 Version:      17.03.1-ce
 API version:  1.27
 Go version:   go1.7.5
 Git commit:   c6d412e
 Built:        Mon Mar 27 17:05:44 2017
 OS/Arch:      linux/amd64

Server:
 Version:      17.03.1-ce
 API version:  1.27 (minimum version 1.12)
 Go version:   go1.7.5
 Git commit:   c6d412e
 Built:        Mon Mar 27 17:05:44 2017
 OS/Arch:      linux/amd64
 Experimental: false
```

**Output of `docker info`:**

```
Containers: 17
 Running: 1
 Paused: 0
 Stopped: 16
Images: 3
Server Version: 17.03.1-ce
Storage Driver: overlay
 Backing Filesystem: xfs
 Supports d_type: false
Logging Driver: json-file
Cgroup Driver: cgroupfs
Plugins: 
 Volume: local
 Network: bridge host macvlan null overlay
Swarm: inactive
Runtimes: runc
Default Runtime: runc
Init Binary: docker-init
containerd version: 4ab9917febca54791c5f071a9d1f404867857fcc
runc version: 54296cf40ad8143b62dbcaa1d90e520a2136ddfe
init version: 949e6fa
Security Options:
 seccomp
  Profile: default
Kernel Version: 3.10.0-514.6.1.el7.x86_64
Operating System: CentOS Linux 7 (Core)
OSType: linux
Architecture: x86_64
CPUs: 4
Total Memory: 14.69 GiB
ID: 5BXG:QZRZ:L2PI:QVJO:LTFG:JUPJ:ZABK:7LBA:D2G7:WR7K:EO65:SPSH
Docker Root Dir: /var/lib/docker
Debug Mode (client): false
Debug Mode (server): false
Registry: https://index.docker.io/v1/
Experimental: false
Insecure Registries:
 127.0.0.0/8
Live Restore Enabled: false
```

**Additional environment details (AWS, VirtualBox, physical, etc.):**

GCE n1-standard-4 instance


Discussion / hints:
It's not crashed because of a buffer overflow, it's crashing because it's tried to reconnect several times and failed and so is throwing an exception.

Ping @tagomoris 

Ideally this should not ever panic.
@tagomoris Can we just have fluent continuously try to reconnect with a max wait time?
Once the buffer is full it should just block new writes, no?
I wonder how will you handle the troubles when destination Fluentd is going down, you don't find it and Docker daemon will continue to reconnect forever - without crashing it? All logs from containers will disappear, and is it ok for you?
IMO If you can find Fluentd outage, you can start it correctly until docker daemon meets the threshold for retries.
@tagomoris It depends on how fluentd is buffering. When the buffer is full is it lossy or does it block?
Currently docker has a 1MB buffer for each container's stdout/stderr that blocks once it's full.

If the log fails to send (vs just blocked) the entry will be dropped.
If it blocks the buffers will start to fill and eventually block the container process from writing to stdout/stderr.

It's definitely undesirable to crash the daemon.
@cpuguy83 No, it doesn't depend on Fluentd buffer configuration. Fluentd forward input plugin will accept messages via TCP connection even when internal buffer is full. Of course, these messages will be lost, but it's not related with any clients (including logging driver).
For such cases, Fluentd's log warns about it. It's problem about Fluentd, not logging drivers.

But if Fluentd is not running, logging driver should tell/notify/warn it to users in any way. For example, containers don't run if it's configured to use Fluentd logging driver and destination isn't running.
For me consistent behavior would be to stop the container. Exactly in the way it does on startup.

Seems that log loss is inevitable in case if fluentd disapeared for a long time.

But in no way it should result in dockerd crash, because that takes down all containers on the host.
@tagomoris problem is if i have just one container which uses fluentd and there isa problem with fluentd - whole docker crashes. 

That stops all other containers which have nothing to do with fluentd and totally contradicts promise of isolation of containers from each other.
IIRC, `panic` just stops to execute a container when I implemented this logging driver at first (Because there are no docker-daemon at that time).
Anyway, it sounds reasonable that whole docker crash is horrible (even for any reasons).

So, the ideal steps are:

1. add option to avoid calling `panic` in fluent-logger-golang (and retry forever)
2. enable that option in logging driver

I'll comment about further details for fluent-logger-golang in https://github.com/fluent/fluent-logger-golang/issues/18
But would this stop the container? Because I see value in scenario, that container gets stopped/destroyed in absence on fluentd (say very small buffer) and they autohealing restarts container somewhere else - thus minimizing loss of logs. 

If it continues trying forever - what would happen to container at that moment? Logs silently discarded? Process writing to stdout blocks?
@gasparch If the call to `Log()` actually fails the log will be dropped and a error will appear in the docker daemon logs.

If the call to `Log()` blocks, the buffers will start filling up until the process can't write to the stream anymore.
There's also support for more of a ring buffer that just buffers until it's full and then starts dropping messages.

So the failure won't be silent.
It may help if instead of just logging an error if we increment some failure count in our metrics so it can be tracked more easily.
thanks, monitoring docker's own logs can solve that problem. 

Still I do not know if I like such architectural solution or prefer solution when software stops working if one component fails (like docker does when it cannot connect to fluentd on container creation). 
It looks more consistent for me "no requested logging" => "no running software" approach, but Docker's user base opinion may be totally different :) so I do not know what is good for majority of users.

But any solution which does not crash docker is better than current one :)

Hi,
I am currently facing this problem on a multi-tenant swarm cluster i'm operating.
How have your thinkings evolved about this issue?
AMHO, "no requested logging" => "no running software" approach cannot apply in my case as each tenant sends its logs to services external to the cluster and I just cannot have docker daemon crash because of that.
Until a satisfaying approach is found, would it just be possible to accept -1 value in the fluentd-max-retries parameter (which is compliant with fluent-logger-golang), that would let the operator decide the behaviour of docker daemon in that case?
make retry count ridicoulosly high and monitor Docker own logs for logging errors :) 

other way - stop using Docker altogether :P :P :P okay, just switch to another logging driver ;)
I've run into this on Docker 18+ and had to use these options to stop the docker daemon crashing:

        ```
        "fluentd-buffer-limit": "4g",
        "fluentd-max-retries": "144",
        "fluentd-retry-wait": 600,
        "fluentd-async-connect": "true"
        ```
Is there a better recommended solution thus far ?
any update on this one should we use fluentd or filebeat using with k8 and ELK.

Which one has less issues ?
There is no guarantee that filebeat from inside container will be able to ship all logs if container is suddenly killed ;) All hassle with `fluentd` for me was to provide more reliable ways to ship logs. But see above :( 

Fix patch:
diff --git a/daemon/logger/fluentd/fluentd.go b/daemon/logger/fluentd/fluentd.go
index 829b606c9f3f8..997c01c65257e 100644
--- a/daemon/logger/fluentd/fluentd.go
+++ b/daemon/logger/fluentd/fluentd.go
@@ -242,6 +242,7 @@ func parseConfig(cfg map[string]string) (fluent.Config, error) {
 		AsyncConnect:       asyncConnect,
 		SubSecondPrecision: subSecondPrecision,
 		RequestAck:         requestAck,
+		ForceStopAsyncSend: async || asyncConnect,
 	}
 
 	return config, nil
diff --git a/vendor.conf b/vendor.conf
index dbf3e9979a268..503c9d229d7a7 100644
--- a/vendor.conf
+++ b/vendor.conf
@@ -106,7 +106,7 @@ github.com/godbus/dbus/v5                           37bf87eef99d69c4f1d3528bd66e
 github.com/Graylog2/go-gelf                         1550ee647df0510058c9d67a45c56f18911d80b8 # v2 branch
 
 # fluent-logger-golang deps
-github.com/fluent/fluent-logger-golang              7a6c9dcd7f14c2ed5d8c55c11b894e5455ee311b # v1.4.0
+github.com/fluent/fluent-logger-golang              0b652e850a9140d0b1db6390d8925d0601e952db # v1.8.0
 github.com/philhofer/fwd                            bb6d471dc95d4fe11e432687f8b70ff496cf3136 # v1.0.0
 github.com/tinylib/msgp                             af6442a0fcf6e2a1b824f70dd0c734f01e817751 # v1.1.0
 
diff --git a/vendor/github.com/fluent/fluent-logger-golang/README.md b/vendor/github.com/fluent/fluent-logger-golang/README.md
index 5930602619082..554619a31cd0e 100644
--- a/vendor/github.com/fluent/fluent-logger-golang/README.md
+++ b/vendor/github.com/fluent/fluent-logger-golang/README.md
@@ -1,7 +1,8 @@
 fluent-logger-golang
 ====
 
-[![Build Status](https://travis-ci.org/fluent/fluent-logger-golang.png?branch=master)](https://travis-ci.org/fluent/fluent-logger-golang)
+[![Build Status](https://github.com/fluent/fluent-logger-golang/actions/workflows/ci.yaml/badge.svg?branch=master)](https://github.com/fluent/fluent-logger-golang/actions)
+[![GoDoc](https://godoc.org/github.com/fluent/fluent-logger-golang/fluent?status.svg)](https://godoc.org/github.com/fluent/fluent-logger-golang/fluent)
 
 ## A structured event logger for Fluentd (Golang)
 
@@ -19,8 +20,6 @@ Install the package with `go get` and use `import` to include it in your project
 import "github.com/fluent/fluent-logger-golang/fluent"
 ```
 
-GoDoc: http://godoc.org/github.com/fluent/fluent-logger-golang/fluent
-
 ## Example
 
 ```go
@@ -29,7 +28,7 @@ package main
 import (
   "github.com/fluent/fluent-logger-golang/fluent"
   "fmt"
-  "time"
+  //"time"
 )
 
 func main() {
@@ -59,22 +58,125 @@ func main() {
 f := fluent.New(fluent.Config{FluentPort: 80, FluentHost: "example.com"})
 ```
 
+### FluentNetwork
+
+Specify the network protocol. The supported values are:
+
+ * "tcp" (use `FluentHost` and `FluentPort`)
+ * "tls" (use`FluentHost` and `FluentPort`)
+ * "unix" (use `FluentSocketPath`)
+
+The default is "tcp".
+
+### FluentHost
+
+Specify a hostname or IP address as a string for the destination of the "tcp" protocol.
+The default is "127.0.0.1".
+
+### FluentPort
+
+Specify the TCP port of the destination. The default is 24224.
+
+### FluentSocketPath
+
+Specify the unix socket path when `FluentNetwork` is "unix".
+
+### Timeout
+
+Set the timeout value of `time.Duration` to connect to the destination.
+The default is 3 seconds.
+
 ### WriteTimeout
 
-Sets the timeout for Write call of logger.Post.
+Sets the timeout value of `time.Duration` for Write call of `logger.Post`.
 Since the default is zero value, Write will not time out.
 
+### BufferLimit
+
+Sets the number of events buffered on the memory. Records will be stored in memory up to this number. If the buffer is full, the call to record logs will fail.
+The default is 8192.
+
+### RetryWait
+
+Set the duration of the initial wait for the first retry, in milliseconds. The actual retry wait will be `r * 1.5^(N-1)` (r: this value, N: the number of retries).
+The default is 500.
+
+### MaxRetry
+
+Sets the maximum number of retries. If the number of retries become larger than this value, the write/send operation will fail.
+The default is 13.
+
+### MaxRetryWait
+
+The maximum duration of wait between retries, in milliseconds. If the calculated retry wait is larger than this value, the actual retry wait will be this value.
+The default is 60,000 (60 seconds).
+
+### TagPrefix
+
+Sets the prefix string of the tag. Prefix will be appended with a dot `.`, like `ppp.tag` (ppp: the value of this parameter, tag: the tag string specified in a call).
+The default is blank.
+
 ### Async
 
 Enable asynchronous I/O (connect and write) for sending events to Fluentd.
 The default is false.
 
+### ForceStopAsyncSend
+
+When Async is enabled, immediately discard the event queue on close() and return (instead of trying MaxRetry times for each event in the queue before returning)
+The default is false.
+
+### AsyncResultCallback
+
+When Async is enabled, if this is callback is provided, it will be called on every write to Fluentd. The callback function
+takes two arguments - a `[]byte` of the message that was to be sent and an `error`. If the `error` is not nil this means the 
+delivery of the message was unsuccessful.
+
+### SubSecondPrecision
+
+Enable time encoding as EventTime, which contains sub-second precision values. The messages encoded with this option can be received only by Fluentd v0.14 or later.
+The default is false.
+
+### MarshalAsJson
+
+Enable Json data marshaling to send messages using Json format (instead of the standard MessagePack). It is supported by Fluentd `in_forward` plugin.
+The default is false.
+
 ### RequestAck
 
 Sets whether to request acknowledgment from Fluentd to increase the reliability
 of the connection. The default is false.
 
+### TlsInsecureSkipVerify
+
+Skip verifying the server certificate. Useful for development and testing. The default is false.
+
+## FAQ
+
+### Does this logger support the features of Fluentd Forward Protocol v1?
+
+"the features" includes heartbeat messages (for TCP keepalive), TLS transport and shared key authentication.
+
+This logger doesn't support those features. Patches are welcome!
+
+### Is it allowed to call `Fluent.Post()` after connection close?
+
+Before v1.8.0, the Fluent logger silently reopened connections whenever
+`Fluent.Post()` was called.
+
+```go
+logger, _ := fluent.New(fluent.Config{})
+logger.Post(tag, data)
+logger.Close()
+logger.Post(tag, data)  /* reopen connection */
+```
+
+However, this behavior was confusing, in particular when multiple goroutines
+were involved. Starting v1.8.0, the logger no longer accepts `Fluent.Post()`
+after `Fluent.Close()`, and instead returns a "Logger already closed" error.
+
 ## Tests
 ```
+
 go test
 ```
diff --git a/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go b/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go
index 5bbd52668ea27..c2e1b25955586 100644
--- a/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go
+++ b/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go
@@ -1,10 +1,13 @@
 package fluent
 
 import (
+	"context"
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"math"
+	"math/rand"
 	"net"
 	"os"
 	"reflect"
@@ -15,8 +18,8 @@ import (
 	"bytes"
 	"encoding/base64"
 	"encoding/binary"
+
 	"github.com/tinylib/msgp/msgp"
-	"math/rand"
 )
 
 const (
@@ -34,21 +37,30 @@ const (
 	// Default sub-second precision value to false since it is only compatible
 	// with fluentd versions v0.14 and above.
 	defaultSubSecondPrecision = false
+
+	// Default value whether to skip checking insecure certs on TLS connections.
+	defaultTlsInsecureSkipVerify = false
 )
 
+// randomGenerator is used by getUniqueId to generate ack hashes. Its value is replaced
+// during tests with a deterministic function.
+var randomGenerator = rand.Uint64
+
 type Config struct {
-	FluentPort       int           `json:"fluent_port"`
-	FluentHost       string        `json:"fluent_host"`
-	FluentNetwork    string        `json:"fluent_network"`
-	FluentSocketPath string        `json:"fluent_socket_path"`
-	Timeout          time.Duration `json:"timeout"`
-	WriteTimeout     time.Duration `json:"write_timeout"`
-	BufferLimit      int           `json:"buffer_limit"`
-	RetryWait        int           `json:"retry_wait"`
-	MaxRetry         int           `json:"max_retry"`
-	MaxRetryWait     int           `json:"max_retry_wait"`
-	TagPrefix        string        `json:"tag_prefix"`
-	Async            bool          `json:"async"`
+	FluentPort          int           `json:"fluent_port"`
+	FluentHost          string        `json:"fluent_host"`
+	FluentNetwork       string        `json:"fluent_network"`
+	FluentSocketPath    string        `json:"fluent_socket_path"`
+	Timeout             time.Duration `json:"timeout"`
+	WriteTimeout        time.Duration `json:"write_timeout"`
+	BufferLimit         int           `json:"buffer_limit"`
+	RetryWait           int           `json:"retry_wait"`
+	MaxRetry            int           `json:"max_retry"`
+	MaxRetryWait        int           `json:"max_retry_wait"`
+	TagPrefix           string        `json:"tag_prefix"`
+	Async               bool          `json:"async"`
+	ForceStopAsyncSend  bool          `json:"force_stop_async_send"`
+	AsyncResultCallback func(data []byte, err error)
 	// Deprecated: Use Async instead
 	AsyncConnect  bool `json:"async_connect"`
 	MarshalAsJSON bool `json:"marshal_as_json"`
@@ -61,6 +73,21 @@ type Config struct {
 	// respond with an acknowledgement. This option improves the reliability
 	// of the message transmission.
 	RequestAck bool `json:"request_ack"`
+
+	// Flag to skip verifying insecure certs on TLS connections
+	TlsInsecureSkipVerify bool `json: "tls_insecure_skip_verify"`
+}
+
+type ErrUnknownNetwork struct {
+	network string
+}
+
+func (e *ErrUnknownNetwork) Error() string {
+	return "unknown network " + e.network
+}
+
+func NewErrUnknownNetwork(network string) error {
+	return &ErrUnknownNetwork{network}
 }
 
 type msgToSend struct {
@@ -71,15 +98,35 @@ type msgToSend struct {
 type Fluent struct {
 	Config
 
-	pending chan *msgToSend
-	wg      sync.WaitGroup
-
-	muconn sync.Mutex
+	dialer dialer
+	// stopRunning is used in async mode to signal to run() it should abort.
+	stopRunning chan struct{}
+	// cancelDialings is used by Close() to stop any in-progress dialing.
+	cancelDialings context.CancelFunc
+	pending        chan *msgToSend
+	pendingMutex   sync.RWMutex
+	closed         bool
+	wg             sync.WaitGroup
+
+	muconn sync.RWMutex
 	conn   net.Conn
 }
 
+type dialer interface {
+	DialContext(ctx context.Context, network, address string) (net.Conn, error)
+}
+
 // New creates a new Logger.
-func New(config Config) (f *Fluent, err error) {
+func New(config Config) (*Fluent, error) {
+	if config.Timeout == 0 {
+		config.Timeout = defaultTimeout
+	}
+	return newWithDialer(config, &net.Dialer{
+		Timeout: config.Timeout,
+	})
+}
+
+func newWithDialer(config Config, d dialer) (f *Fluent, err error) {
 	if config.FluentNetwork == "" {
 		config.FluentNetwork = defaultNetwork
 	}
@@ -92,9 +139,6 @@ func New(config Config) (f *Fluent, err error) {
 	if config.FluentSocketPath == "" {
 		config.FluentSocketPath = defaultSocketPath
 	}
-	if config.Timeout == 0 {
-		config.Timeout = defaultTimeout
-	}
 	if config.WriteTimeout == 0 {
 		config.WriteTimeout = defaultWriteTimeout
 	}
@@ -110,20 +154,36 @@ func New(config Config) (f *Fluent, err error) {
 	if config.MaxRetryWait == 0 {
 		config.MaxRetryWait = defaultMaxRetryWait
 	}
+	if !config.TlsInsecureSkipVerify {
+		config.TlsInsecureSkipVerify = defaultTlsInsecureSkipVerify
+	}
 	if config.AsyncConnect {
 		fmt.Fprintf(os.Stderr, "fluent#New: AsyncConnect is now deprecated, please use Async instead")
 		config.Async = config.Async || config.AsyncConnect
 	}
+
 	if config.Async {
+		ctx, cancel := context.WithCancel(context.Background())
+
 		f = &Fluent{
-			Config:  config,
-			pending: make(chan *msgToSend, config.BufferLimit),
+			Config:         config,
+			dialer:         d,
+			stopRunning:    make(chan struct{}),
+			cancelDialings: cancel,
+			pending:        make(chan *msgToSend, config.BufferLimit),
+			pendingMutex:   sync.RWMutex{},
+			muconn:         sync.RWMutex{},
 		}
+
 		f.wg.Add(1)
-		go f.run()
+		go f.run(ctx)
 	} else {
-		f = &Fluent{Config: config}
-		err = f.connect()
+		f = &Fluent{
+			Config: config,
+			dialer: d,
+			muconn: sync.RWMutex{},
+		}
+		err = f.connect(context.Background())
 	}
 	return
 }
@@ -176,13 +236,18 @@ func (f *Fluent) PostWithTime(tag string, tm time.Time, message interface{}) err
 		fields := msgtype.NumField()
 		for i := 0; i < fields; i++ {
 			field := msgtype.Field(i)
+			value := msg.FieldByIndex(field.Index)
+			// ignore unexported fields
+			if !value.CanInterface() {
+				continue
+			}
 			name := field.Name
 			if n1 := field.Tag.Get("msg"); n1 != "" {
 				name = n1
 			} else if n2 := field.Tag.Get("codec"); n2 != "" {
 				name = n2
 			}
-			kv[name] = msg.FieldByIndex(field.Index).Interface()
+			kv[name] = value.Interface()
 		}
 		return f.EncodeAndPostData(tag, tm, kv)
 	}
@@ -219,8 +284,12 @@ func (f *Fluent) postRawData(msg *msgToSend) error {
 	if f.Config.Async {
 		return f.appendBuffer(msg)
 	}
+
 	// Synchronous write
-	return f.write(msg)
+	if f.closed {
+		return fmt.Errorf("fluent#postRawData: Logger already closed")
+	}
+	return f.writeWithRetry(context.Background(), msg)
 }
 
 // For sending forward protocol adopted JSON
@@ -254,7 +323,7 @@ func getUniqueID(timeUnix int64) (string, error) {
 		enc.Close()
 		return "", err
 	}
-	if err := binary.Write(enc, binary.LittleEndian, rand.Uint64()); err != nil {
+	if err := binary.Write(enc, binary.LittleEndian, randomGenerator()); err != nil {
 		enc.Close()
 		return "", err
 	}
@@ -290,18 +359,55 @@ func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (msg
 	return
 }
 
-// Close closes the connection, waiting for pending logs to be sent
+// Close closes the connection, waiting for pending logs to be sent. If the client is
+// running in async mode, the run() goroutine exits before Close() returns.
 func (f *Fluent) Close() (err error) {
 	if f.Config.Async {
+		f.pendingMutex.Lock()
+		if f.closed {
+			f.pendingMutex.Unlock()
+			return nil
+		}
+		f.closed = true
+		f.pendingMutex.Unlock()
+
+		if f.Config.ForceStopAsyncSend {
+			close(f.stopRunning)
+			f.cancelDialings()
+		}
+
 		close(f.pending)
-		f.wg.Wait()
+		// If ForceStopAsyncSend is false, all logs in the channel have to be sent
+		// before closing the connection. At this point closed is true so no more
+		// logs are written to the channel and f.pending has been closed, so run()
+		// goroutine will exit as soon as all logs in the channel are sent.
+		if !f.Config.ForceStopAsyncSend {
+			f.wg.Wait()
+		}
 	}
+
+	f.muconn.Lock()
 	f.close()
+	f.closed = true
+	f.muconn.Unlock()
+
+	// If ForceStopAsyncSend is true, we shall close the connection before waiting for
+	// run() goroutine to exit to be sure we aren't waiting on ack message that might
+	// never come (eg. because fluentd server is down). However we want to be sure the
+	// run() goroutine stops before returning from Close().
+	if f.Config.ForceStopAsyncSend {
+		f.wg.Wait()
+	}
 	return
 }
 
 // appendBuffer appends data to buffer with lock.
 func (f *Fluent) appendBuffer(msg *msgToSend) error {
+	f.pendingMutex.RLock()
+	defer f.pendingMutex.RUnlock()
+	if f.closed {
+		return fmt.Errorf("fluent#appendBuffer: Logger already closed")
+	}
 	select {
 	case f.pending <- msg:
 	default:
@@ -310,42 +416,114 @@ func (f *Fluent) appendBuffer(msg *msgToSend) error {
 	return nil
 }
 
-// close closes the connection.
+// close closes the connection. Callers should take care of locking muconn first.
 func (f *Fluent) close() {
-	f.muconn.Lock()
 	if f.conn != nil {
 		f.conn.Close()
 		f.conn = nil
 	}
-	f.muconn.Unlock()
 }
 
-// connect establishes a new connection using the specified transport.
-func (f *Fluent) connect() (err error) {
-
+// connect establishes a new connection using the specified transport. Caller should
+// take care of locking muconn first.
+func (f *Fluent) connect(ctx context.Context) (err error) {
 	switch f.Config.FluentNetwork {
 	case "tcp":
-		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort), f.Config.Timeout)
+		f.conn, err = f.dialer.DialContext(ctx,
+			f.Config.FluentNetwork,
+			f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort))
+	case "tls":
+		tlsConfig := &tls.Config{InsecureSkipVerify: f.Config.TlsInsecureSkipVerify}
+		f.conn, err = tls.DialWithDialer(
+			&net.Dialer{Timeout: f.Config.Timeout},
+			"tcp",
+			f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort), tlsConfig,
+		)
 	case "unix":
-		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentSocketPath, f.Config.Timeout)
+		f.conn, err = f.dialer.DialContext(ctx,
+			f.Config.FluentNetwork,
+			f.Config.FluentSocketPath)
 	default:
-		err = net.UnknownNetworkError(f.Config.FluentNetwork)
+		err = NewErrUnknownNetwork(f.Config.FluentNetwork)
 	}
+
 	return err
 }
 
-func (f *Fluent) run() {
+var errIsClosing = errors.New("fluent logger is closing")
+
+// Caller should take care of locking muconn first.
+func (f *Fluent) connectWithRetry(ctx context.Context) error {
+	// A Time channel is used instead of time.Sleep() to avoid blocking this
+	// goroutine during way too much time (because of the exponential back-off
+	// retry).
+	// time.NewTimer() is used instead of time.After() to avoid leaking the
+	// timer channel (cf. https://pkg.go.dev/time#After).
+	timeout := time.NewTimer(time.Duration(0))
+	defer func() {
+		// timeout.Stop() is called in a function literal instead of being
+		// defered directly as it's re-assigned below when the retry loop spins.
+		timeout.Stop()
+	}()
+
+	for i := 0; i < f.Config.MaxRetry; i++ {
+		select {
+		case <-timeout.C:
+			err := f.connect(ctx)
+			if err == nil {
+				return nil
+			}
+
+			if _, ok := err.(*ErrUnknownNetwork); ok {
+				return err
+			}
+			if err == context.Canceled {
+				return errIsClosing
+			}
+
+			waitTime := f.Config.RetryWait * e(defaultReconnectWaitIncreRate, float64(i-1))
+			if waitTime > f.Config.MaxRetryWait {
+				waitTime = f.Config.MaxRetryWait
+			}
+
+			timeout = time.NewTimer(time.Duration(waitTime) * time.Millisecond)
+		case <-ctx.Done():
+			return errIsClosing
+		}
+	}
+
+	return fmt.Errorf("could not connect to fluentd after %d retries", f.Config.MaxRetry)
+}
+
+// run is the goroutine used to unqueue and write logs in async mode. That
+// goroutine is meant to run during the whole life of the Fluent logger.
+func (f *Fluent) run(ctx context.Context) {
 	for {
 		select {
 		case entry, ok := <-f.pending:
+			// f.stopRunning is closed before f.pending only when ForceStopAsyncSend
+			// is enabled. Otherwise, f.pending is closed when Close() is called.
 			if !ok {
 				f.wg.Done()
 				return
 			}
-			err := f.write(entry)
-			if err != nil {
+
+			err := f.writeWithRetry(ctx, entry)
+			if err != nil && err != errIsClosing {
 				fmt.Fprintf(os.Stderr, "[%s] Unable to send logs to fluentd, reconnecting...\n", time.Now().Format(time.RFC3339))
 			}
+			if f.AsyncResultCallback != nil {
+				var data []byte
+				if entry != nil {
+					data = entry.data
+				}
+				f.AsyncResultCallback(data, err)
+			}
+		case <-f.stopRunning:
+			fmt.Fprintf(os.Stderr, "[%s] Discarding queued events...\n", time.Now().Format(time.RFC3339))
+
+			f.wg.Done()
+			return
 		}
 	}
 }
@@ -354,55 +532,85 @@ func e(x, y float64) int {
 	return int(math.Pow(x, y))
 }
 
-func (f *Fluent) write(msg *msgToSend) error {
-
+func (f *Fluent) writeWithRetry(ctx context.Context, msg *msgToSend) error {
 	for i := 0; i < f.Config.MaxRetry; i++ {
+		if retry, err := f.write(ctx, msg); !retry {
+			return err
+		}
+	}
+
+	return fmt.Errorf("fluent#write: failed to write after %d attempts", f.Config.MaxRetry)
+}
+
+// write writes the provided msg to fluentd server. Its first return values is
+// a bool indicating whether the write should be retried.
+// This method relies on function literals to execute muconn.Unlock or
+// muconn.RUnlock in deferred calls to ensure the mutex is unlocked even in
+// the case of panic recovering.
+func (f *Fluent) write(ctx context.Context, msg *msgToSend) (bool, error) {
+	closer := func() {
+		f.muconn.Lock()
+		defer f.muconn.Unlock()
+
+		f.close()
+	}
 
-		// Connect if needed
+	if err := func() (err error) {
 		f.muconn.Lock()
+		defer f.muconn.Unlock()
+
 		if f.conn == nil {
-			err := f.connect()
-			if err != nil {
-				f.muconn.Unlock()
-				waitTime := f.Config.RetryWait * e(defaultReconnectWaitIncreRate, float64(i-1))
-				if waitTime > f.Config.MaxRetryWait {
-					waitTime = f.Config.MaxRetryWait
-				}
-				time.Sleep(time.Duration(waitTime) * time.Millisecond)
-				continue
-			}
+			err = f.connectWithRetry(ctx)
+		}
+
+		return err
+	}(); err != nil {
+		// Here, we don't want to retry the write since connectWithRetry already
+		// retries Config.MaxRetry times to connect.
+		return false, fmt.Errorf("fluent#write: %v", err)
+	}
+
+	if err := func() (err error) {
+		f.muconn.RLock()
+		defer f.muconn.RUnlock()
+
+		if f.conn == nil {
+			return fmt.Errorf("connection has been closed before writing to it")
 		}
-		f.muconn.Unlock()
 
-		// We're connected, write msg
 		t := f.Config.WriteTimeout
 		if time.Duration(0) < t {
 			f.conn.SetWriteDeadline(time.Now().Add(t))
 		} else {
 			f.conn.SetWriteDeadline(time.Time{})
 		}
-		_, err := f.conn.Write(msg.data)
-		if err != nil {
-			f.close()
+
+		_, err = f.conn.Write(msg.data)
+		return err
+	}(); err != nil {
+		closer()
+		return true, fmt.Errorf("fluent#write: %v", err)
+	}
+
+	// Acknowledgment check
+	if msg.ack != "" {
+		resp := &AckResp{}
+		var err error
+		if f.Config.MarshalAsJSON {
+			dec := json.NewDecoder(f.conn)
+			err = dec.Decode(resp)
 		} else {
-			// Acknowledgment check
-			if msg.ack != "" {
-				resp := &AckResp{}
-				if f.Config.MarshalAsJSON {
-					dec := json.NewDecoder(f.conn)
-					err = dec.Decode(resp)
-				} else {
-					r := msgp.NewReader(f.conn)
-					err = resp.DecodeMsg(r)
-				}
-				if err != nil || resp.Ack != msg.ack {
-					f.close()
-					continue
-				}
-			}
-			return err
+			r := msgp.NewReader(f.conn)
+			err = resp.DecodeMsg(r)
+		}
+
+		if err != nil || resp.Ack != msg.ack {
+			fmt.Fprintf(os.Stderr, "fluent#write: message ack (%s) doesn't match expected one (%s). Closing connection...", resp.Ack, msg.ack)
+
+			closer()
+			return true, err
 		}
 	}
 
-	return fmt.Errorf("fluent#write: failed to reconnect, max retry: %v", f.Config.MaxRetry)
+	return false, nil
 }
diff --git a/vendor/github.com/fluent/fluent-logger-golang/fluent/proto.go b/vendor/github.com/fluent/fluent-logger-golang/fluent/proto.go
index 76fc860ee357d..95faa9624a77f 100644
--- a/vendor/github.com/fluent/fluent-logger-golang/fluent/proto.go
+++ b/vendor/github.com/fluent/fluent-logger-golang/fluent/proto.go
@@ -3,6 +3,7 @@
 package fluent
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/tinylib/msgp/msgp"
@@ -93,8 +94,19 @@ func (t *EventTime) MarshalBinaryTo(b []byte) error {
 	return nil
 }
 
-// UnmarshalBinary is not implemented since decoding messages is not supported
-// by this library.
+// Although decoding messages is not officially supported by this library,
+// UnmarshalBinary is implemented for testing and general completeness.
 func (t *EventTime) UnmarshalBinary(b []byte) error {
+	if len(b) != length {
+		return fmt.Errorf("Invalid EventTime byte length: %d", len(b))
+	}
+
+	sec := (int32(b[0]) << 24) | (int32(b[1]) << 16)
+	sec = sec | (int32(b[2]) << 8) | int32(b[3])
+
+	nsec := (int32(b[4]) << 24) | (int32(b[5]) << 16)
+	nsec = nsec | (int32(b[6]) << 8) | int32(b[7])
+
+	*t = EventTime(time.Unix(int64(sec), int64(nsec)))
 	return nil
 }


Test patch: