Merge pull request #19439 from pcarrier/pcarrier/19438-async-connections-to-fluentd

fluentd logger: support all options besides Unix sockets

daemon/logger/fluentd/fluentd.go

@@ -8,10 +8,12 @@ import (
 	"net"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/docker/daemon/logger/loggerutils"
+	"github.com/docker/go-units"
 	"github.com/fluent/fluent-logger-golang/fluent"
 )
 
@@ -24,10 +26,25 @@ type fluentd struct {
 }
 
 const (
-	name               = "fluentd"
+	name = "fluentd"
-	defaultHostName    = "localhost"
+
+	defaultHost        = "127.0.0.1"
 	defaultPort        = 24224
-	defaultBufferLimit = 1 * 1024 * 1024 // 1M buffer by default
+	defaultBufferLimit = 1024 * 1024
+	defaultTagPrefix   = "docker"
+
+	// logger tries to reconnect 2**32 - 1 times
+	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
+	defaultRetryWait              = 1000
+	defaultTimeout                = 3 * time.Second
+	defaultMaxRetries             = math.MaxInt32
+	defaultReconnectWaitIncreRate = 1.5
+
+	addressKey      = "fluentd-address"
+	bufferLimitKey  = "fluentd-buffer-limit"
+	retryWaitKey    = "fluentd-retry-wait"
+	maxRetriesKey   = "fluentd-max-retries"
+	asyncConnectKey = "fluentd-async-connect"
 )
 
 func init() {
@@ -43,7 +60,7 @@ func init() {
 // the context. Supported context configuration variables are
 // fluentd-address & fluentd-tag.
 func New(ctx logger.Context) (logger.Logger, error) {
-	host, port, err := parseAddress(ctx.Config["fluentd-address"])
+	host, port, err := parseAddress(ctx.Config[addressKey])
 	if err != nil {
 		return nil, err
 	}
@@ -52,24 +69,58 @@ func New(ctx logger.Context) (logger.Logger, error) {
 	if err != nil {
 		return nil, err
 	}
-	failOnStartupError, err := loggerutils.ParseFailOnStartupErrorFlag(ctx)
+
-	if err != nil {
-		return nil, err
-	}
-	bufferLimit, err := parseBufferLimit(ctx.Config["buffer-limit"])
-	if err != nil {
-		return nil, err
-	}
 	extra := ctx.ExtraAttributes(nil)
-	logrus.Debugf("logging driver fluentd configured for container:%s, host:%s, port:%d, tag:%s, extra:%v.", ctx.ContainerID, host, port, tag, extra)
+
-	// logger tries to reconnect 2**32 - 1 times
+	bufferLimit := defaultBufferLimit
-	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
+	if ctx.Config[bufferLimitKey] != "" {
-	log, err := fluent.New(fluent.Config{FluentPort: port, FluentHost: host, RetryWait: 1000, MaxRetry: math.MaxInt32, BufferLimit: bufferLimit})
+		bl64, err := units.RAMInBytes(ctx.Config[bufferLimitKey])
-	if err != nil {
+		if err != nil {
-		if failOnStartupError {
 			return nil, err
 		}
-		logrus.Warnf("fluentd cannot connect to configured endpoint. Ignoring as instructed. Error: %q", err)
+		bufferLimit = int(bl64)
+	}
+
+	retryWait := defaultRetryWait
+	if ctx.Config[retryWaitKey] != "" {
+		rwd, err := time.ParseDuration(ctx.Config[retryWaitKey])
+		if err != nil {
+			return nil, err
+		}
+		retryWait = int(rwd.Seconds() * 1000)
+	}
+
+	maxRetries := defaultMaxRetries
+	if ctx.Config[maxRetriesKey] != "" {
+		mr64, err := strconv.ParseUint(ctx.Config[maxRetriesKey], 10, strconv.IntSize)
+		if err != nil {
+			return nil, err
+		}
+		maxRetries = int(mr64)
+	}
+
+	asyncConnect := false
+	if ctx.Config[asyncConnectKey] != "" {
+		if asyncConnect, err = strconv.ParseBool(ctx.Config[asyncConnectKey]); err != nil {
+			return nil, err
+		}
+	}
+
+	fluentConfig := fluent.Config{
+		FluentPort:   port,
+		FluentHost:   host,
+		BufferLimit:  bufferLimit,
+		RetryWait:    retryWait,
+		MaxRetry:     maxRetries,
+		AsyncConnect: asyncConnect,
+	}
+
+	logrus.WithField("container", ctx.ContainerID).WithField("config", fluentConfig).
+		Debug("logging driver fluentd configured")
+
+	log, err := fluent.New(fluentConfig)
+	if err != nil {
+		return nil, err
 	}
 	return &fluentd{
 		tag:           tag,
@@ -107,13 +158,16 @@ func (f *fluentd) Name() string {
 func ValidateLogOpt(cfg map[string]string) error {
 	for key := range cfg {
 		switch key {
-		case "fluentd-address":
-		case "fluentd-tag":
-		case "tag":
-		case "labels":
 		case "env":
-		case "fail-on-startup-error":
+		case "fluentd-tag":
-		case "buffer-limit":
+		case "labels":
+		case "tag":
+		case addressKey:
+		case bufferLimitKey:
+		case retryWaitKey:
+		case maxRetriesKey:
+		case asyncConnectKey:
+			// Accepted
 		default:
 			return fmt.Errorf("unknown log opt '%s' for fluentd log driver", key)
 		}
@@ -128,7 +182,7 @@ func ValidateLogOpt(cfg map[string]string) error {
 
 func parseAddress(address string) (string, int, error) {
 	if address == "" {
-		return defaultHostName, defaultPort, nil
+		return defaultHost, defaultPort, nil
 	}
 
 	host, port, err := net.SplitHostPort(address)
@@ -145,14 +199,3 @@ func parseAddress(address string) (string, int, error) {
 	}
 	return host, portnum, nil
 }
-
-func parseBufferLimit(bufferLimit string) (int, error) {
-	if bufferLimit == "" {
-		return defaultBufferLimit, nil
-	}
-	limit, err := strconv.Atoi(bufferLimit)
-	if err != nil {
-		return 0, fmt.Errorf("invalid buffer limit %s: %s", bufferLimit, err)
-	}
-	return limit, nil
-}

daemon/logger/loggerutils/log_option_helpers.go

@@ -1,26 +0,0 @@
-package loggerutils
-
-import (
-	"fmt"
-	"strconv"
-
-	"github.com/docker/docker/daemon/logger"
-)
-
-const (
-	defaultFailOnStartupError = true // So that we do not break existing behaviour
-)
-
-// ParseFailOnStartupErrorFlag parses a log driver flag that determines if
-// the driver should ignore possible connection errors during startup
-func ParseFailOnStartupErrorFlag(ctx logger.Context) (bool, error) {
-	failOnStartupError := ctx.Config["fail-on-startup-error"]
-	if failOnStartupError == "" {
-		return defaultFailOnStartupError, nil
-	}
-	failOnStartupErrorFlag, err := strconv.ParseBool(failOnStartupError)
-	if err != nil {
-		return defaultFailOnStartupError, fmt.Errorf("invalid connect error flag %s: %s", failOnStartupError, err)
-	}
-	return failOnStartupErrorFlag, nil
-}

daemon/logger/loggerutils/log_option_helpers_test.go

@@ -1,51 +0,0 @@
-package loggerutils
-
-import (
-	"testing"
-
-	"github.com/docker/docker/daemon/logger"
-)
-
-func TestParseDefaultIgnoreFlag(t *testing.T) {
-	ctx := buildContext(map[string]string{})
-	flag, e := ParseFailOnStartupErrorFlag(ctx)
-	assertFlag(t, e, flag, true)
-}
-
-func TestParseIgnoreFlagWhenFalse(t *testing.T) {
-	ctx := buildContext(map[string]string{"fail-on-startup-error": "false"})
-	flag, e := ParseFailOnStartupErrorFlag(ctx)
-	assertFlag(t, e, flag, false)
-}
-
-func TestParseIgnoreFlagWhenTrue(t *testing.T) {
-	ctx := buildContext(map[string]string{"fail-on-startup-error": "true"})
-	flag, e := ParseFailOnStartupErrorFlag(ctx)
-	assertFlag(t, e, flag, true)
-}
-
-func TestParseIgnoreFlagWithError(t *testing.T) {
-	ctx := buildContext(map[string]string{"fail-on-startup-error": "maybe :)"})
-	flag, e := ParseFailOnStartupErrorFlag(ctx)
-	if e == nil {
-		t.Fatalf("Error should have happened")
-	}
-	assertFlag(t, nil, flag, true)
-}
-
-// Helpers
-
-func buildConfig(cfg map[string]string) logger.Context {
-	return logger.Context{
-		Config: cfg,
-	}
-}
-
-func assertFlag(t *testing.T, e error, flag bool, expected bool) {
-	if e != nil {
-		t.Fatalf("Error parsing ignore connect error flag: %q", e)
-	}
-	if flag != expected {
-		t.Fatalf("Wrong flag: %t, should be %t", flag, expected)
-	}
-}

docs/admin/logging/fluentd.md

@@ -35,8 +35,7 @@ Some options are supported by specifying `--log-opt` as many times as needed:
 
  - `fluentd-address`: specify `host:port` to connect `localhost:24224`
  - `tag`: specify tag for fluentd message, which interpret some markup, ex `{{.ID}}`, `{{.FullID}}` or `{{.Name}}` `docker.{{.ID}}`
- - `fail-on-startup-error`: true/false; Should the logging driver fail container startup in case of connect error during startup. Default: true (backwards compatible)
+
- - `buffer-limit`: Size limit (bytes) for the buffer which is used to buffer messages in case of connection outages. Default: 1M
 
 Configure the default logging driver by passing the
 `--log-driver` option to the Docker daemon:
@@ -55,7 +54,7 @@ connects to this daemon through `localhost:24224` by default. Use the
     docker run --log-driver=fluentd --log-opt fluentd-address=myhost.local:24224
 
 If container cannot connect to the Fluentd daemon, the container stops
-immediately.
+immediately unless the `fluentd-async-connect` option is used.
 
 ## Options
 
@@ -79,20 +78,9 @@ the log tag format.
 
 The `labels` and `env` options each take a comma-separated list of keys. If there is collision between `label` and `env` keys, the value of the `env` takes precedence. Both options add additional fields to the extra attributes of a logging message.
 
-### fail-on-startup-error
+### fluentd-async-connect
 
-By default, if the logging driver cannot connect to the backend it will fail the entire startup of the container. If you wish to ignore potential connect error during container startup supply the `fail-on-startup-error` flag.
+Docker connects to Fluentd in the background. Messages are buffered until the connection is established.
-
-    docker run --log-driver=fluentd --log-opt fail-on-startup-error=false
-
-
-### buffer-limit
-
-When fluent driver loses connection, or cannot connect at container startup, it will buffer the log events locally for re-transmission. Buffer limit option controls how much data will be buffered locally, **per container**. Specified in bytes.
-
-    docker run --log-driver=fluentd --log-opt buffer-limit=5242880
-
-The above would result to use 5M buffer locally. Keep in mind that during possible connection errors all your containers will start buffering locally and thus might result in considerable memory usage.
 
 ## Fluentd daemon management with Docker
 

docs/admin/logging/overview.md

@@ -189,17 +189,20 @@ run slower but compress more. Default value is 1 (BestSpeed).
 You can use the `--log-opt NAME=VALUE` flag to specify these additional Fluentd logging driver options.
 
  - `fluentd-address`: specify `host:port` to connect [localhost:24224]
- - `tag`: specify tag for `fluentd` message,
+ - `tag`: specify tag for `fluentd` message
- - `fail-on-startup-error`: true/false; Should the logging driver fail container startup in case of connect error during startup. Default: true (backwards compatible)
+ - `fluentd-buffer-limit`: specify the maximum size of the fluentd log buffer [8MB]
- - `buffer-limit`: Size limit (bytes) for the buffer which is used to buffer messages in case of connection outages. Default: 1M
+ - `fluentd-retry-wait`: initial delay before a connection retry (after which it increases exponentially) [1000ms]
+ - `fluentd-max-retries`: maximum number of connection retries before abrupt failure of docker [1073741824]
+ - `fluentd-async-connect`: whether to block on initial connection or not [false]
 
 For example, to specify both additional options:
 
 `docker run --log-driver=fluentd --log-opt fluentd-address=localhost:24224 --log-opt tag=docker.{{.Name}}`
 
-If container cannot connect to the Fluentd daemon on the specified address,
+If container cannot connect to the Fluentd daemon on the specified address and
-the container stops immediately. For detailed information on working with this
+`fluentd-async-connect` is not enabled, the container stops immediately.
-logging driver, see [the fluentd logging driver](fluentd.md)
+For detailed information on working with this logging driver,
+see [the fluentd logging driver](fluentd.md)
 
 
 ## Specify Amazon CloudWatch Logs options

hack/vendor.sh

@@ -71,7 +71,7 @@ clone git github.com/golang/protobuf 68415e7123da32b07eab49c96d2c4d6158360e9b
 # gelf logging driver deps
 clone git github.com/Graylog2/go-gelf aab2f594e4585d43468ac57287b0dece9d806883
 
-clone git github.com/fluent/fluent-logger-golang v1.0.0
+clone git github.com/fluent/fluent-logger-golang v1.1.0
 # fluent-logger-golang deps
 clone git github.com/philhofer/fwd 899e4efba8eaa1fea74175308f3fae18ff3319fa
 clone git github.com/tinylib/msgp 75ee40d2601edf122ef667e2a07d600d4c44490c

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

@@ -14,6 +14,8 @@ import (
 
 const (
 	defaultHost                   = "127.0.0.1"
+	defaultNetwork                = "tcp"
+	defaultSocketPath             = ""
 	defaultPort                   = 24224
 	defaultTimeout                = 3 * time.Second
 	defaultBufferLimit            = 8 * 1024 * 1024
@@ -23,13 +25,16 @@ const (
 )
 
 type Config struct {
-	FluentPort  int
+	FluentPort       int
-	FluentHost  string
+	FluentHost       string
-	Timeout     time.Duration
+	FluentNetwork    string
-	BufferLimit int
+	FluentSocketPath string
-	RetryWait   int
+	Timeout          time.Duration
-	MaxRetry    int
+	BufferLimit      int
-	TagPrefix   string
+	RetryWait        int
+	MaxRetry         int
+	TagPrefix        string
+	AsyncConnect     bool
 }
 
 type Fluent struct {
@@ -42,12 +47,18 @@ type Fluent struct {
 
 // New creates a new Logger.
 func New(config Config) (f *Fluent, err error) {
+	if config.FluentNetwork == "" {
+		config.FluentNetwork = defaultNetwork
+	}
 	if config.FluentHost == "" {
 		config.FluentHost = defaultHost
 	}
 	if config.FluentPort == 0 {
 		config.FluentPort = defaultPort
 	}
+	if config.FluentSocketPath == "" {
+		config.FluentSocketPath = defaultSocketPath
+	}
 	if config.Timeout == 0 {
 		config.Timeout = defaultTimeout
 	}
@@ -60,8 +71,13 @@ func New(config Config) (f *Fluent, err error) {
 	if config.MaxRetry == 0 {
 		config.MaxRetry = defaultMaxRetry
 	}
-	f = &Fluent{Config: config, reconnecting: false}
+	if config.AsyncConnect {
-	err = f.connect()
+		f = &Fluent{Config: config, reconnecting: true}
+		f.reconnect()
+	} else {
+		f = &Fluent{Config: config, reconnecting: false}
+		err = f.connect()
+	}
 	return
 }
 
@@ -171,9 +187,9 @@ func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (data
 // Close closes the connection.
 func (f *Fluent) Close() (err error) {
 	if len(f.pending) > 0 {
-		_ = f.send()
+		err = f.send()
 	}
-	err = f.close()
+	f.close()
 	return
 }
 
@@ -194,7 +210,14 @@ func (f *Fluent) close() (err error) {
 
 // connect establishes a new connection using the specified transport.
 func (f *Fluent) connect() (err error) {
-	f.conn, err = net.DialTimeout("tcp", f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort), f.Config.Timeout)
+	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)
+	case "unix":
+		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentSocketPath, f.Config.Timeout)
+	default:
+		err = net.UnknownNetworkError(f.Config.FluentNetwork)
+	}
 	return
 }
 

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

@@ -1,3 +1,3 @@
 package fluent
 
-const Version = "1.0.0"
+const Version = "1.1.0"