Implement optional ring buffer for container logs

This allows the user to set a logging mode to "blocking" (default), or
"non-blocking", which uses the ring buffer as a proxy to the real log
driver.

This allows a container to never be blocked on stdio at the cost of
dropping log messages.

Introduces 2 new log-opts that works for all drivers, `log-mode` and
`log-size`. `log-mode` takes a  value of "blocking", or "non-blocking"
I chose not to implement this as a bool since it is difficult to
determine if the mode was set to false vs just not set... especially
difficult when merging the default daemon config with the container config.
`log-size` takes a size string, e.g. `2MB`, which sets the max size
of the ring buffer. When the max size is reached, it will start
dropping log messages.

```
BenchmarkRingLoggerThroughputNoReceiver-8           	2000000000	        36.2 ns/op	 856.35 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputWithReceiverDelay0-8   	300000000	       156 ns/op	 198.48 MB/s	      32 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay1-8        	2000000000	        36.1 ns/op	 857.80 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay10-8       	1000000000	        36.2 ns/op	 856.53 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay50-8       	2000000000	        34.7 ns/op	 894.65 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay100-8      	2000000000	        35.1 ns/op	 883.91 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay300-8      	1000000000	        35.9 ns/op	 863.90 MB/s	       0 B/op	       0 allocs/op
BenchmarkRingLoggerThroughputConsumeDelay500-8      	2000000000	        35.8 ns/op	 866.88 MB/s	       0 B/op	       0 allocs/op
```

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

Author: cpuguy83

api/types/container/host_config.go

@@ -229,6 +229,17 @@ func (rp *RestartPolicy) IsSame(tp *RestartPolicy) bool {
 	return rp.Name == tp.Name && rp.MaximumRetryCount == tp.MaximumRetryCount
 }
 
+// LogMode is a type to define the available modes for logging
+// These modes affect how logs are handled when log messages start piling up.
+type LogMode string
+
+// Available logging modes
+const (
+	LogModeUnset            = ""
+	LogModeBlocking LogMode = "blocking"
+	LogModeNonBlock LogMode = "non-blocking"
+)
+
 // LogConfig represents the logging configuration of the container.
 type LogConfig struct {
 	Type   string

container/container.go

@@ -37,6 +37,7 @@ import (
 	runconfigopts "github.com/docker/docker/runconfig/opts"
 	"github.com/docker/docker/volume"
 	"github.com/docker/go-connections/nat"
+	"github.com/docker/go-units"
 	"github.com/docker/libnetwork"
 	"github.com/docker/libnetwork/netlabel"
 	"github.com/docker/libnetwork/options"
@@ -310,7 +311,7 @@ func (container *Container) CheckpointDir() string {
 // StartLogger starts a new logger driver for the container.
 func (container *Container) StartLogger() (logger.Logger, error) {
 	cfg := container.HostConfig.LogConfig
-	c, err := logger.GetLogDriver(cfg.Type)
+	initDriver, err := logger.GetLogDriver(cfg.Type)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get logging factory: %v", err)
 	}
@@ -335,7 +336,23 @@ func (container *Container) StartLogger() (logger.Logger, error) {
 			return nil, err
 		}
 	}
-	return c(info)
+
+	l, err := initDriver(info)
+	if err != nil {
+		return nil, err
+	}
+
+	if containertypes.LogMode(cfg.Config["mode"]) == containertypes.LogModeNonBlock {
+		bufferSize := int64(-1)
+		if s, exists := cfg.Config["max-buffer-size"]; exists {
+			bufferSize, err = units.RAMInBytes(s)
+			if err != nil {
+				return nil, err
+			}
+		}
+		l = logger.NewRingLogger(l, info, bufferSize)
+	}
+	return l, nil
 }
 
 // GetProcessLabel returns the process label for the container.

daemon/logger/awslogs/cloudwatchlogs.go

@@ -176,7 +176,7 @@ func (l *logStream) Log(msg *logger.Message) error {
 	defer l.lock.RUnlock()
 	if !l.closed {
 		// buffer up the data, making sure to copy the Line data
-		l.messages <- logger.CopyMessage(msg)
+		l.messages <- msg
 	}
 	return nil
 }

daemon/logger/copier.go

@@ -47,7 +47,6 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 	buf := make([]byte, bufSize)
 	n := 0
 	eof := false
-	msg := &Message{Source: name}
 
 	for {
 		select {
@@ -78,13 +77,16 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 			// Break up the data that we've buffered up into lines, and log each in turn.
 			p := 0
 			for q := bytes.Index(buf[p:n], []byte{'\n'}); q >= 0; q = bytes.Index(buf[p:n], []byte{'\n'}) {
-				msg.Line = buf[p : p+q]
-				msg.Timestamp = time.Now().UTC()
-				msg.Partial = false
 				select {
 				case <-c.closed:
 					return
 				default:
+					msg := &Message{
+						Source:    name,
+						Timestamp: time.Now().UTC(),
+					}
+					msg.Line = append(msg.Line, buf[p:p+q]...)
+
 					if logErr := c.dst.Log(msg); logErr != nil {
 						logrus.Errorf("Failed to log msg %q for logger %s: %s", msg.Line, c.dst.Name(), logErr)
 					}
@@ -96,9 +98,14 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 			// noting that it's a partial log line.
 			if eof || (p == 0 && n == len(buf)) {
 				if p < n {
-					msg.Line = buf[p:n]
-					msg.Timestamp = time.Now().UTC()
+					msg := &Message{
+						Source:    name,
+						Timestamp: time.Now().UTC(),
+						Partial:   true,
+					}
+					msg.Line = append(msg.Line, buf[p:n]...)
 					msg.Partial = true
+
 					if logErr := c.dst.Log(msg); logErr != nil {
 						logrus.Errorf("Failed to log msg %q for logger %s: %s", msg.Line, c.dst.Name(), logErr)
 					}

daemon/logger/factory.go

@@ -3,6 +3,10 @@ package logger
 import (
 	"fmt"
 	"sync"
+
+	containertypes "github.com/docker/docker/api/types/container"
+	units "github.com/docker/go-units"
+	"github.com/pkg/errors"
 )
 
 // Creator builds a logging driver instance with given context.
@@ -85,20 +89,47 @@ func GetLogDriver(name string) (Creator, error) {
 	return factory.get(name)
 }
 
+var builtInLogOpts = map[string]bool{
+	"mode":            true,
+	"max-buffer-size": true,
+}
+
 // ValidateLogOpts checks the options for the given log driver. The
 // options supported are specific to the LogDriver implementation.
 func ValidateLogOpts(name string, cfg map[string]string) error {
 	if name == "none" {
 		return nil
 	}
 
+	switch containertypes.LogMode(cfg["mode"]) {
+	case containertypes.LogModeBlocking, containertypes.LogModeNonBlock, containertypes.LogModeUnset:
+	default:
+		return fmt.Errorf("logger: logging mode not supported: %s", cfg["mode"])
+	}
+
+	if s, ok := cfg["max-buffer-size"]; ok {
+		if containertypes.LogMode(cfg["mode"]) != containertypes.LogModeNonBlock {
+			return fmt.Errorf("logger: max-buffer-size option is only supported with 'mode=%s'", containertypes.LogModeNonBlock)
+		}
+		if _, err := units.RAMInBytes(s); err != nil {
+			return errors.Wrap(err, "error parsing option max-buffer-size")
+		}
+	}
+
 	if !factory.driverRegistered(name) {
 		return fmt.Errorf("logger: no log driver named '%s' is registered", name)
 	}
 
+	filteredOpts := make(map[string]string, len(builtInLogOpts))
+	for k, v := range cfg {
+		if !builtInLogOpts[k] {
+			filteredOpts[k] = v
+		}
+	}
+
 	validator := factory.getLogOptValidator(name)
 	if validator != nil {
-		return validator(cfg)
+		return validator(filteredOpts)
 	}
 	return nil
 }

daemon/logger/logger.go

@@ -37,24 +37,6 @@ type Message struct {
 	Partial   bool
 }
 
-// CopyMessage creates a copy of the passed-in Message which will remain
-// unchanged if the original is changed.  Log drivers which buffer Messages
-// rather than dispatching them during their Log() method should use this
-// function to obtain a Message whose Line member's contents won't change.
-func CopyMessage(msg *Message) *Message {
-	m := new(Message)
-	m.Line = make([]byte, len(msg.Line))
-	copy(m.Line, msg.Line)
-	m.Source = msg.Source
-	m.Timestamp = msg.Timestamp
-	m.Partial = msg.Partial
-	m.Attrs = make(LogAttributes)
-	for k, v := range msg.Attrs {
-		m.Attrs[k] = v
-	}
-	return m
-}
-
 // LogAttributes is used to hold the extra attributes available in the log message
 // Primarily used for converting the map type to string and sorting.
 type LogAttributes map[string]string