Merge pull request #18041 from jnummelin/feature/fluent-logger-ignore-connect-error-flag

Added flag to ignore fluentd connect error on container start
Upstream-commit: 7a016578dbb5c9b8a43b43e1ce96785fa26baf56
Component: engine

components/engine/daemon/logger/fluentd/fluentd.go

@@ -24,10 +24,12 @@ type fluentd struct {
 }
 
 const (
-	name             = "fluentd"
-	defaultHostName  = "localhost"
-	defaultPort      = 24224
-	defaultTagPrefix = "docker"
+	name                             = "fluentd"
+	defaultHostName                  = "localhost"
+	defaultPort                      = 24224
+	defaultTagPrefix                 = "docker"
+	defaultIgnoreConnectErrorOnStart = false           // So that we do not break existing behaviour
+	defaultBufferLimit               = 1 * 1024 * 1024 // 1M buffer by default
 )
 
 func init() {
@@ -52,13 +54,24 @@ 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
 	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
-	log, err := fluent.New(fluent.Config{FluentPort: port, FluentHost: host, RetryWait: 1000, MaxRetry: math.MaxInt32})
+	log, err := fluent.New(fluent.Config{FluentPort: port, FluentHost: host, RetryWait: 1000, MaxRetry: math.MaxInt32, BufferLimit: bufferLimit})
 	if err != nil {
-		return nil, err
+		if failOnStartupError {
+			return nil, err
+		}
+		logrus.Warnf("fluentd cannot connect to configured endpoint. Ignoring as instructed. Error: %q", err)
 	}
 	return &fluentd{
 		tag:           tag,
@@ -101,6 +114,8 @@ func ValidateLogOpt(cfg map[string]string) error {
 		case "tag":
 		case "labels":
 		case "env":
+		case "fail-on-startup-error":
+		case "buffer-limit":
 		default:
 			return fmt.Errorf("unknown log opt '%s' for fluentd log driver", key)
 		}
@@ -132,3 +147,14 @@ 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
+}

components/engine/daemon/logger/loggerutils/log_option_helpers.go

@@ -0,0 +1,26 @@
+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
+}

components/engine/daemon/logger/loggerutils/log_option_helpers_test.go

@@ -0,0 +1,51 @@
+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)
+	}
+}