vendor: github.com/fluent/fluent-logger-golang 1.6.1

Updates the fluent logger library. Namely this fixes a couple places
where the library could panic when closing and writing to channels.

see https://github.com/fluent/fluent-logger-golang/pull/93 and
https://github.com/fluent/fluent-logger-golang/pull/95

closes #40829
closes #32567

Signed-off-by: Cam <gh@sparr.email>

vendor.conf

@@ -105,7 +105,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              b9b7fb02ccfee8ba4e69aa87386820c2bf24fd11 # v1.6.1
 github.com/philhofer/fwd                            bb6d471dc95d4fe11e432687f8b70ff496cf3136 # v1.0.0
 github.com/tinylib/msgp                             af6442a0fcf6e2a1b824f70dd0c734f01e817751 # v1.1.0
 

vendor/github.com/fluent/fluent-logger-golang/README.md

@@ -2,6 +2,7 @@ fluent-logger-golang
 ====
 
 [![Build Status](https://travis-ci.org/fluent/fluent-logger-golang.png?branch=master)](https://travis-ci.org/fluent/fluent-logger-golang)
+[![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,21 +58,92 @@ func main() {
 f := fluent.New(fluent.Config{FluentPort: 80, FluentHost: "example.com"})
 ```
 
+### FluentNetwork
+
+Specify the network protocol, as "tcp" (use `FluentHost` and `FluentPort`) or "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.
+
+### 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.
 
+## 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!
+
 ## Tests
 ```
 go test

vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go

@@ -15,8 +15,9 @@ import (
 	"bytes"
 	"encoding/base64"
 	"encoding/binary"
-	"github.com/tinylib/msgp/msgp"
 	"math/rand"
+
+	"github.com/tinylib/msgp/msgp"
 )
 
 const (
@@ -37,18 +38,19 @@ const (
 )
 
 type Config struct {
-	FluentPort       int           `json:"fluent_port"`
+	FluentPort         int           `json:"fluent_port"`
-	FluentHost       string        `json:"fluent_host"`
+	FluentHost         string        `json:"fluent_host"`
-	FluentNetwork    string        `json:"fluent_network"`
+	FluentNetwork      string        `json:"fluent_network"`
-	FluentSocketPath string        `json:"fluent_socket_path"`
+	FluentSocketPath   string        `json:"fluent_socket_path"`
-	Timeout          time.Duration `json:"timeout"`
+	Timeout            time.Duration `json:"timeout"`
-	WriteTimeout     time.Duration `json:"write_timeout"`
+	WriteTimeout       time.Duration `json:"write_timeout"`
-	BufferLimit      int           `json:"buffer_limit"`
+	BufferLimit        int           `json:"buffer_limit"`
-	RetryWait        int           `json:"retry_wait"`
+	RetryWait          int           `json:"retry_wait"`
-	MaxRetry         int           `json:"max_retry"`
+	MaxRetry           int           `json:"max_retry"`
-	MaxRetryWait     int           `json:"max_retry_wait"`
+	MaxRetryWait       int           `json:"max_retry_wait"`
-	TagPrefix        string        `json:"tag_prefix"`
+	TagPrefix          string        `json:"tag_prefix"`
-	Async            bool          `json:"async"`
+	Async              bool          `json:"async"`
+	ForceStopAsyncSend bool          `json:"force_stop_async_send"`
 	// Deprecated: Use Async instead
 	AsyncConnect  bool `json:"async_connect"`
 	MarshalAsJSON bool `json:"marshal_as_json"`
@@ -63,6 +65,18 @@ type Config struct {
 	RequestAck bool `json:"request_ack"`
 }
 
+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 {
 	data []byte
 	ack  string
@@ -71,15 +85,32 @@ type msgToSend struct {
 type Fluent struct {
 	Config
 
-	pending chan *msgToSend
+	dialer       dialer
-	wg      sync.WaitGroup
+	stopRunning  chan bool
+	pending      chan *msgToSend
+	pendingMutex sync.RWMutex
+	chanClosed   bool
+	wg           sync.WaitGroup
 
 	muconn sync.Mutex
 	conn   net.Conn
 }
 
 // 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,
+	})
+}
+
+type dialer interface {
+	Dial(string, string) (net.Conn, error)
+}
+
+func newWithDialer(config Config, d dialer) (f *Fluent, err error) {
 	if config.FluentNetwork == "" {
 		config.FluentNetwork = defaultNetwork
 	}
@@ -92,9 +123,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
 	}
@@ -114,15 +142,22 @@ func New(config Config) (f *Fluent, err error) {
 		fmt.Fprintf(os.Stderr, "fluent#New: AsyncConnect is now deprecated, please use Async instead")
 		config.Async = config.Async || config.AsyncConnect
 	}
+
 	if config.Async {
 		f = &Fluent{
-			Config:  config,
+			Config:       config,
-			pending: make(chan *msgToSend, config.BufferLimit),
+			dialer:       d,
+			pending:      make(chan *msgToSend, config.BufferLimit),
+			pendingMutex: sync.RWMutex{},
+			stopRunning:  make(chan bool, 1),
 		}
 		f.wg.Add(1)
 		go f.run()
 	} else {
-		f = &Fluent{Config: config}
+		f = &Fluent{
+			Config: config,
+			dialer: d,
+		}
 		err = f.connect()
 	}
 	return
@@ -292,16 +327,32 @@ func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (msg
 
 // Close closes the connection, waiting for pending logs to be sent
 func (f *Fluent) Close() (err error) {
+	defer f.close(f.conn)
 	if f.Config.Async {
+		f.pendingMutex.Lock()
+		if f.chanClosed {
+			f.pendingMutex.Unlock()
+			return nil
+		}
+		f.chanClosed = true
+		f.pendingMutex.Unlock()
+		if f.Config.ForceStopAsyncSend {
+			f.stopRunning <- true
+			close(f.stopRunning)
+		}
 		close(f.pending)
 		f.wg.Wait()
 	}
-	f.close()
+	return nil
-	return
 }
 
 // appendBuffer appends data to buffer with lock.
 func (f *Fluent) appendBuffer(msg *msgToSend) error {
+	f.pendingMutex.RLock()
+	defer f.pendingMutex.RUnlock()
+	if f.chanClosed {
+		return fmt.Errorf("fluent#appendBuffer: Logger already closed")
+	}
 	select {
 	case f.pending <- msg:
 	default:
@@ -311,9 +362,9 @@ func (f *Fluent) appendBuffer(msg *msgToSend) error {
 }
 
 // close closes the connection.
-func (f *Fluent) close() {
+func (f *Fluent) close(c net.Conn) {
 	f.muconn.Lock()
-	if f.conn != nil {
+	if f.conn != nil && f.conn == c {
 		f.conn.Close()
 		f.conn = nil
 	}
@@ -322,19 +373,24 @@ func (f *Fluent) close() {
 
 // connect establishes a new connection using the specified transport.
 func (f *Fluent) connect() (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.Dial(
+			f.Config.FluentNetwork,
+			f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort))
 	case "unix":
-		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentSocketPath, f.Config.Timeout)
+		f.conn, err = f.dialer.Dial(
+			f.Config.FluentNetwork,
+			f.Config.FluentSocketPath)
 	default:
-		err = net.UnknownNetworkError(f.Config.FluentNetwork)
+		err = NewErrUnknownNetwork(f.Config.FluentNetwork)
 	}
 	return err
 }
 
 func (f *Fluent) run() {
+	drainEvents := false
+	var emitEventDrainMsg sync.Once
 	for {
 		select {
 		case entry, ok := <-f.pending:
@@ -342,11 +398,22 @@ func (f *Fluent) run() {
 				f.wg.Done()
 				return
 			}
+			if drainEvents {
+				emitEventDrainMsg.Do(func() { fmt.Fprintf(os.Stderr, "[%s] Discarding queued events...\n", time.Now().Format(time.RFC3339)) })
+				continue
+			}
 			err := f.write(entry)
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "[%s] Unable to send logs to fluentd, reconnecting...\n", time.Now().Format(time.RFC3339))
 			}
 		}
+		select {
+		case stopRunning, ok := <-f.stopRunning:
+			if stopRunning || !ok {
+				drainEvents = true
+			}
+		default:
+		}
 	}
 }
 
@@ -355,48 +422,56 @@ func e(x, y float64) int {
 }
 
 func (f *Fluent) write(msg *msgToSend) error {
-
+	var c net.Conn
 	for i := 0; i < f.Config.MaxRetry; i++ {
-
+		c = f.conn
 		// Connect if needed
-		f.muconn.Lock()
+		if c == nil {
-		if f.conn == nil {
+			f.muconn.Lock()
-			err := f.connect()
+			if f.conn == nil {
-			if err != nil {
+				err := f.connect()
-				f.muconn.Unlock()
+				if err != nil {
-				waitTime := f.Config.RetryWait * e(defaultReconnectWaitIncreRate, float64(i-1))
+					f.muconn.Unlock()
-				if waitTime > f.Config.MaxRetryWait {
+
-					waitTime = f.Config.MaxRetryWait
+					if _, ok := err.(*ErrUnknownNetwork); ok {
+						// do not retry on unknown network error
+						break
+					}
+					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
 				}
-				time.Sleep(time.Duration(waitTime) * time.Millisecond)
-				continue
 			}
+			c = f.conn
+			f.muconn.Unlock()
 		}
-		f.muconn.Unlock()
 
 		// We're connected, write msg
 		t := f.Config.WriteTimeout
 		if time.Duration(0) < t {
-			f.conn.SetWriteDeadline(time.Now().Add(t))
+			c.SetWriteDeadline(time.Now().Add(t))
 		} else {
-			f.conn.SetWriteDeadline(time.Time{})
+			c.SetWriteDeadline(time.Time{})
 		}
-		_, err := f.conn.Write(msg.data)
+		_, err := c.Write(msg.data)
 		if err != nil {
-			f.close()
+			f.close(c)
 		} else {
 			// Acknowledgment check
 			if msg.ack != "" {
 				resp := &AckResp{}
 				if f.Config.MarshalAsJSON {
-					dec := json.NewDecoder(f.conn)
+					dec := json.NewDecoder(c)
 					err = dec.Decode(resp)
 				} else {
-					r := msgp.NewReader(f.conn)
+					r := msgp.NewReader(c)
 					err = resp.DecodeMsg(r)
 				}
 				if err != nil || resp.Ack != msg.ack {
-					f.close()
+					f.close(c)
 					continue
 				}
 			}

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
+// Although decoding messages is not officially supported by this library,
-// 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
 }