Reworked Asynchronous mode: Fluentd Post data is no longer blocking in asynchronous mode

- Configuration AsyncConnect is now named Async.
- Errors during reconnect no longer cause panic.
- Synchronous mode is now always synchronous (reconnect synchronously).

Signed-off-by: alexlry <alexandre@keymantics.com>

Author: alexlry

fluent/fluent.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"os"
 	"reflect"
 	"strconv"
 	"sync"
@@ -21,7 +22,7 @@ const (
 	defaultPort                   = 24224
 	defaultTimeout                = 3 * time.Second
 	defaultWriteTimeout           = time.Duration(0) // Write() will not time out
-	defaultBufferLimit            = 8 * 1024 * 1024
+	defaultBufferLimit            = 8 * 1024
 	defaultRetryWait              = 500
 	defaultMaxRetryWait           = 60000
 	defaultMaxRetry               = 13
@@ -43,8 +44,10 @@ type Config struct {
 	MaxRetry         int           `json:"max_retry"`
 	MaxRetryWait     int           `json:"max_retry_wait"`
 	TagPrefix        string        `json:"tag_prefix"`
-	AsyncConnect     bool          `json:"async_connect"`
-	MarshalAsJSON    bool          `json:"marshal_as_json"`
+	Async            bool          `json:"async"`
+	// Deprecated: Use Async instead
+	AsyncConnect  bool `json:"async_connect"`
+	MarshalAsJSON bool `json:"marshal_as_json"`
 
 	// Sub-second precision timestamps are only possible for those using fluentd
 	// v0.14+ and serializing their messages with msgpack.
@@ -54,12 +57,11 @@ type Config struct {
 type Fluent struct {
 	Config
 
-	mubuff  sync.Mutex
-	pending []byte
+	pending chan []byte
+	wg      sync.WaitGroup
 
-	muconn       sync.Mutex
-	conn         net.Conn
-	reconnecting bool
+	muconn sync.Mutex
+	conn   net.Conn
 }
 
 // New creates a new Logger.
@@ -95,10 +97,17 @@ func New(config Config) (f *Fluent, err error) {
 		config.MaxRetryWait = defaultMaxRetryWait
 	}
 	if config.AsyncConnect {
-		f = &Fluent{Config: config, reconnecting: true}
-		go f.reconnect()
+		config.Async = config.Async || config.AsyncConnect
+	}
+	if config.Async {
+		f = &Fluent{
+			Config:  config,
+			pending: make(chan []byte, config.BufferLimit),
+		}
+		f.wg.Add(1)
+		go f.run()
 	} else {
-		f = &Fluent{Config: config, reconnecting: false}
+		f = &Fluent{Config: config}
 		err = f.connect()
 	}
 	return
@@ -192,14 +201,11 @@ func (f *Fluent) PostRawData(data []byte) {
 }
 
 func (f *Fluent) postRawData(data []byte) error {
-	if err := f.appendBuffer(data); err != nil {
-		return err
+	if f.Config.Async {
+		return f.appendBuffer(data)
 	}
-	if err := f.send(); err != nil {
-		f.close()
-		return err
-	}
-	return nil
+	// Synchronous write
+	return f.write(data)
 }
 
 // For sending forward protocol adopted JSON
@@ -232,23 +238,23 @@ func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (data
 	return
 }
 
-// Close closes the connection.
+// Close closes the connection, waiting for pending logs to be sent
 func (f *Fluent) Close() (err error) {
-	if len(f.pending) > 0 {
-		err = f.send()
+	if f.Config.Async {
+		close(f.pending)
+		f.wg.Wait()
 	}
 	f.close()
 	return
 }
 
 // appendBuffer appends data to buffer with lock.
 func (f *Fluent) appendBuffer(data []byte) error {
-	f.mubuff.Lock()
-	defer f.mubuff.Unlock()
-	if len(f.pending)+len(data) > f.Config.BufferLimit {
-		return errors.New(fmt.Sprintf("fluent#appendBuffer: Buffer full, limit %v", f.Config.BufferLimit))
+	select {
+	case f.pending <- data:
+	default:
+		return fmt.Errorf("fluent#appendBuffer: Buffer full, limit %v", f.Config.BufferLimit)
 	}
-	f.pending = append(f.pending, data...)
 	return nil
 }
 
@@ -264,8 +270,6 @@ func (f *Fluent) close() {
 
 // connect establishes a new connection using the specified transport.
 func (f *Fluent) connect() (err error) {
-	f.muconn.Lock()
-	defer f.muconn.Unlock()
 
 	switch f.Config.FluentNetwork {
 	case "tcp":
@@ -275,66 +279,63 @@ func (f *Fluent) connect() (err error) {
 	default:
 		err = net.UnknownNetworkError(f.Config.FluentNetwork)
 	}
+	return err
+}
 
-	if err == nil {
-		f.reconnecting = false
+func (f *Fluent) run() {
+	for {
+		select {
+		case entry, ok := <-f.pending:
+			if !ok {
+				f.wg.Done()
+				return
+			}
+			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))
+			}
+		}
 	}
-	return
 }
 
 func e(x, y float64) int {
 	return int(math.Pow(x, y))
 }
 
-func (f *Fluent) reconnect() {
-	for i := 0; ; i++ {
-		err := f.connect()
-		if err == nil {
-			f.send()
-			return
-		}
-		if i == f.Config.MaxRetry {
-			// TODO: What we can do when connection failed MaxRetry times?
-			panic("fluent#reconnect: failed to reconnect!")
-		}
-		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)
-	}
-}
-
-func (f *Fluent) send() error {
-	f.muconn.Lock()
-	defer f.muconn.Unlock()
-
-	if f.conn == nil {
-		if f.reconnecting == false {
-			f.reconnecting = true
-			go f.reconnect()
+func (f *Fluent) write(data []byte) error {
+
+	for i := 0; i < f.Config.MaxRetry; i++ {
+
+		// Connect if needed
+		f.muconn.Lock()
+		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
+			}
 		}
-		return errors.New("fluent#send: can't send logs, client is reconnecting")
-	}
-
-	f.mubuff.Lock()
-	defer f.mubuff.Unlock()
+		f.muconn.Unlock()
 
-	var err error
-	if len(f.pending) > 0 {
+		// We're connected, write data
 		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(f.pending)
+		_, err := f.conn.Write(data)
 		if err != nil {
-			f.conn.Close()
-			f.conn = nil
+			f.close()
 		} else {
-			f.pending = f.pending[:0]
+			return err
 		}
 	}
-	return err
+
+	return errors.New("fluent#write: failed to reconnect")
 }

fluent/fluent_test.go

@@ -140,16 +140,16 @@ func Test_New_itShouldUseConfigValuesFromMashalAsJSONArgument(t *testing.T) {
 }
 
 func Test_send_WritePendingToConn(t *testing.T) {
-	f := &Fluent{Config: Config{}, reconnecting: false}
+	f, _ := New(Config{Async: true})
 
 	conn := &Conn{}
 	f.conn = conn
 
 	msg := "This is test writing."
 	bmsg := []byte(msg)
-	f.pending = append(f.pending, bmsg...)
+	f.pending <- bmsg
 
-	err := f.send()
+	err := f.write(bmsg)
 	if err != nil {
 		t.Error(err)
 	}
@@ -159,10 +159,12 @@ func Test_send_WritePendingToConn(t *testing.T) {
 	if string(rcv) != msg {
 		t.Errorf("got %s, except %s", string(rcv), msg)
 	}
+
+	f.Close()
 }
 
 func Test_MarshalAsMsgpack(t *testing.T) {
-	f := &Fluent{Config: Config{}, reconnecting: false}
+	f := &Fluent{Config: Config{}}
 
 	conn := &Conn{}
 	f.conn = conn
@@ -193,7 +195,6 @@ func Test_SubSecondPrecision(t *testing.T) {
 		Config: Config{
 			SubSecondPrecision: true,
 		},
-		reconnecting: false,
 	}
 	fluent.conn = &Conn{}
 
@@ -215,7 +216,7 @@ func Test_SubSecondPrecision(t *testing.T) {
 }
 
 func Test_MarshalAsJSON(t *testing.T) {
-	f := &Fluent{Config: Config{MarshalAsJSON: true}, reconnecting: false}
+	f := &Fluent{Config: Config{MarshalAsJSON: true}}
 
 	conn := &Conn{}
 	f.conn = conn
@@ -250,11 +251,11 @@ func TestJsonConfig(t *testing.T) {
 		FluentSocketPath: "/var/tmp/fluent.sock",
 		Timeout:          3000,
 		WriteTimeout:     6000,
-		BufferLimit:      200,
+		BufferLimit:      10,
 		RetryWait:        5,
 		MaxRetry:         3,
 		TagPrefix:        "fluent",
-		AsyncConnect:     false,
+		Async:            false,
 		MarshalAsJSON:    true,
 	}
 
@@ -276,8 +277,8 @@ func TestAsyncConnect(t *testing.T) {
 	ch := make(chan result, 1)
 	go func() {
 		config := Config{
-			FluentPort:   8888,
-			AsyncConnect: true,
+			FluentPort: 8888,
+			Async:      true,
 		}
 		f, err := New(config)
 		ch <- result{f: f, err: err}
@@ -291,14 +292,14 @@ func TestAsyncConnect(t *testing.T) {
 		}
 		res.f.Close()
 	case <-time.After(time.Millisecond * 500):
-		t.Error("AsyncConnect must not block")
+		t.Error("Async must not block")
 	}
 }
 
 func Test_PostWithTimeNotTimeOut(t *testing.T) {
 	f, err := New(Config{
 		FluentPort:    6666,
-		AsyncConnect:  false,
+		Async:         false,
 		MarshalAsJSON: true, // easy to check equality
 	})
 	if err != nil {
@@ -343,7 +344,7 @@ func Test_PostWithTimeNotTimeOut(t *testing.T) {
 func Test_PostMsgpMarshaler(t *testing.T) {
 	f, err := New(Config{
 		FluentPort:    6666,
-		AsyncConnect:  false,
+		Async:         false,
 		MarshalAsJSON: true, // easy to check equality
 	})
 	if err != nil {

fluent/testdata/config.json

@@ -5,10 +5,10 @@
   "fluent_socket_path":"/var/tmp/fluent.sock",
   "timeout":3000,
   "write_timeout":6000,
-  "buffer_limit":200,
+  "buffer_limit":10,
   "retry_wait":5,
   "max_retry":3,
   "tag_prefix":"fluent",
-  "async_connect": false,
+  "async": false,
   "marshal_as_json": true
 }