Merge pull request #10568 from LK4D4/logging_drivers

Logging drivers

daemon/config.go

@@ -7,6 +7,7 @@ import (
 	"github.com/docker/docker/opts"
 	flag "github.com/docker/docker/pkg/mflag"
 	"github.com/docker/docker/pkg/ulimit"
+	"github.com/docker/docker/runconfig"
 )
 
 const (
@@ -47,6 +48,7 @@ type Config struct {
 	TrustKeyPath                string
 	Labels                      []string
 	Ulimits                     map[string]*ulimit.Ulimit
+	LogConfig                   runconfig.LogConfig
 }
 
 // InstallFlags adds command-line options to the top-level flag parser for
@@ -81,6 +83,7 @@ func (config *Config) InstallFlags() {
 	opts.LabelListVar(&config.Labels, []string{"-label"}, "Set key=value labels to the daemon")
 	config.Ulimits = make(map[string]*ulimit.Ulimit)
 	opts.UlimitMapVar(config.Ulimits, []string{"-default-ulimit"}, "Set default ulimits for containers")
+	flag.StringVar(&config.LogConfig.Type, []string{"-log-driver"}, "json-file", "Containers logging driver(json-file/none)")
 }
 
 func getDefaultNetworkMtu() int {

daemon/container.go

@@ -21,6 +21,8 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/docker/daemon/execdriver"
+	"github.com/docker/docker/daemon/logger"
+	"github.com/docker/docker/daemon/logger/jsonfilelog"
 	"github.com/docker/docker/engine"
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/links"
@@ -98,9 +100,11 @@ type Container struct {
 	VolumesRW  map[string]bool
 	hostConfig *runconfig.HostConfig
 
-	activeLinks        map[string]*links.Link
-	monitor            *containerMonitor
-	execCommands       *execStore
+	activeLinks  map[string]*links.Link
+	monitor      *containerMonitor
+	execCommands *execStore
+	// logDriver for closing
+	logDriver          logger.Logger
 	AppliedVolumesFrom map[string]struct{}
 }
 
@@ -1355,21 +1359,36 @@ func (container *Container) setupWorkingDirectory() error {
 	return nil
 }
 
-func (container *Container) startLoggingToDisk() error {
-	// Setup logging of stdout and stderr to disk
-	logPath, err := container.logPath("json")
-	if err != nil {
-		return err
+func (container *Container) startLogging() error {
+	cfg := container.hostConfig.LogConfig
+	if cfg.Type == "" {
+		cfg = container.daemon.defaultLogConfig
 	}
-	container.LogPath = logPath
+	var l logger.Logger
+	switch cfg.Type {
+	case "json-file":
+		pth, err := container.logPath("json")
+		if err != nil {
+			return err
+		}
 
-	if err := container.daemon.LogToDisk(container.stdout, container.LogPath, "stdout"); err != nil {
-		return err
+		dl, err := jsonfilelog.New(pth)
+		if err != nil {
+			return err
+		}
+		l = dl
+	case "none":
+		return nil
+	default:
+		return fmt.Errorf("Unknown logging driver: %s", cfg.Type)
 	}
 
-	if err := container.daemon.LogToDisk(container.stderr, container.LogPath, "stderr"); err != nil {
+	if copier, err := logger.NewCopier(container.ID, map[string]io.Reader{"stdout": container.StdoutPipe(), "stderr": container.StderrPipe()}, l); err != nil {
 		return err
+	} else {
+		copier.Run()
 	}
+	container.logDriver = l
 
 	return nil
 }
@@ -1470,3 +1489,12 @@ func (container *Container) getNetworkedContainer() (*Container, error) {
 func (container *Container) Stats() (*execdriver.ResourceStats, error) {
 	return container.daemon.Stats(container)
 }
+
+func (c *Container) LogDriverType() string {
+	c.Lock()
+	defer c.Unlock()
+	if c.hostConfig.LogConfig.Type == "" {
+		return c.daemon.defaultLogConfig.Type
+	}
+	return c.hostConfig.LogConfig.Type
+}

daemon/daemon.go

@@ -89,23 +89,24 @@ func (c *contStore) List() []*Container {
 }
 
 type Daemon struct {
-	ID             string
-	repository     string
-	sysInitPath    string
-	containers     *contStore
-	execCommands   *execStore
-	graph          *graph.Graph
-	repositories   *graph.TagStore
-	idIndex        *truncindex.TruncIndex
-	sysInfo        *sysinfo.SysInfo
-	volumes        *volumes.Repository
-	eng            *engine.Engine
-	config         *Config
-	containerGraph *graphdb.Database
-	driver         graphdriver.Driver
-	execDriver     execdriver.Driver
-	trustStore     *trust.TrustStore
-	statsCollector *statsCollector
+	ID               string
+	repository       string
+	sysInitPath      string
+	containers       *contStore
+	execCommands     *execStore
+	graph            *graph.Graph
+	repositories     *graph.TagStore
+	idIndex          *truncindex.TruncIndex
+	sysInfo          *sysinfo.SysInfo
+	volumes          *volumes.Repository
+	eng              *engine.Engine
+	config           *Config
+	containerGraph   *graphdb.Database
+	driver           graphdriver.Driver
+	execDriver       execdriver.Driver
+	trustStore       *trust.TrustStore
+	statsCollector   *statsCollector
+	defaultLogConfig runconfig.LogConfig
 }
 
 // Install installs daemon capabilities to eng.
@@ -991,23 +992,24 @@ func NewDaemonFromDirectory(config *Config, eng *engine.Engine) (*Daemon, error)
 	}
 
 	daemon := &Daemon{
-		ID:             trustKey.PublicKey().KeyID(),
-		repository:     daemonRepo,
-		containers:     &contStore{s: make(map[string]*Container)},
-		execCommands:   newExecStore(),
-		graph:          g,
-		repositories:   repositories,
-		idIndex:        truncindex.NewTruncIndex([]string{}),
-		sysInfo:        sysInfo,
-		volumes:        volumes,
-		config:         config,
-		containerGraph: graph,
-		driver:         driver,
-		sysInitPath:    sysInitPath,
-		execDriver:     ed,
-		eng:            eng,
-		trustStore:     t,
-		statsCollector: newStatsCollector(1 * time.Second),
+		ID:               trustKey.PublicKey().KeyID(),
+		repository:       daemonRepo,
+		containers:       &contStore{s: make(map[string]*Container)},
+		execCommands:     newExecStore(),
+		graph:            g,
+		repositories:     repositories,
+		idIndex:          truncindex.NewTruncIndex([]string{}),
+		sysInfo:          sysInfo,
+		volumes:          volumes,
+		config:           config,
+		containerGraph:   graph,
+		driver:           driver,
+		sysInitPath:      sysInitPath,
+		execDriver:       ed,
+		eng:              eng,
+		trustStore:       t,
+		statsCollector:   newStatsCollector(1 * time.Second),
+		defaultLogConfig: config.LogConfig,
 	}
 
 	// Setup shutdown handlers

daemon/inspect.go

@@ -62,6 +62,14 @@ func (daemon *Daemon) ContainerInspect(job *engine.Job) engine.Status {
 			container.hostConfig.Links = append(container.hostConfig.Links, fmt.Sprintf("%s:%s", child.Name, linkAlias))
 		}
 	}
+	// we need this trick to preserve empty log driver, so
+	// container will use daemon defaults even if daemon change them
+	if container.hostConfig.LogConfig.Type == "" {
+		container.hostConfig.LogConfig = daemon.defaultLogConfig
+		defer func() {
+			container.hostConfig.LogConfig = runconfig.LogConfig{}
+		}()
+	}
 
 	out.SetJson("HostConfig", container.hostConfig)
 

daemon/logger/copier.go

@@ -0,0 +1,48 @@
+package logger
+
+import (
+	"bufio"
+	"io"
+	"time"
+
+	"github.com/Sirupsen/logrus"
+)
+
+// Copier can copy logs from specified sources to Logger and attach
+// ContainerID and Timestamp.
+// Writes are concurrent, so you need implement some sync in your logger
+type Copier struct {
+	// cid is container id for which we copying logs
+	cid string
+	// srcs is map of name -> reader pairs, for example "stdout", "stderr"
+	srcs map[string]io.Reader
+	dst  Logger
+}
+
+// NewCopier creates new Copier
+func NewCopier(cid string, srcs map[string]io.Reader, dst Logger) (*Copier, error) {
+	return &Copier{
+		cid:  cid,
+		srcs: srcs,
+		dst:  dst,
+	}, nil
+}
+
+// Run starts logs copying
+func (c *Copier) Run() {
+	for src, w := range c.srcs {
+		go c.copySrc(src, w)
+	}
+}
+
+func (c *Copier) copySrc(name string, src io.Reader) {
+	scanner := bufio.NewScanner(src)
+	for scanner.Scan() {
+		if err := c.dst.Log(&Message{ContainerID: c.cid, Line: scanner.Bytes(), Source: name, Timestamp: time.Now().UTC()}); err != nil {
+			logrus.Errorf("Failed to log msg %q for logger %s: %s", scanner.Bytes(), c.dst.Name(), err)
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		logrus.Errorf("Error scanning log stream: %s", err)
+	}
+}

daemon/logger/copier_test.go

@@ -0,0 +1,100 @@
+package logger
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+)
+
+type TestLoggerJSON struct {
+	*json.Encoder
+}
+
+func (l *TestLoggerJSON) Log(m *Message) error {
+	return l.Encode(m)
+}
+
+func (l *TestLoggerJSON) Close() error {
+	return nil
+}
+
+func (l *TestLoggerJSON) Name() string {
+	return "json"
+}
+
+type TestLoggerText struct {
+	*bytes.Buffer
+}
+
+func (l *TestLoggerText) Log(m *Message) error {
+	_, err := l.WriteString(m.ContainerID + " " + m.Source + " " + string(m.Line) + "\n")
+	return err
+}
+
+func (l *TestLoggerText) Close() error {
+	return nil
+}
+
+func (l *TestLoggerText) Name() string {
+	return "text"
+}
+
+func TestCopier(t *testing.T) {
+	stdoutLine := "Line that thinks that it is log line from docker stdout"
+	stderrLine := "Line that thinks that it is log line from docker stderr"
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	for i := 0; i < 30; i++ {
+		if _, err := stdout.WriteString(stdoutLine + "\n"); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := stderr.WriteString(stderrLine + "\n"); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	var jsonBuf bytes.Buffer
+
+	jsonLog := &TestLoggerJSON{Encoder: json.NewEncoder(&jsonBuf)}
+
+	cid := "a7317399f3f857173c6179d44823594f8294678dea9999662e5c625b5a1c7657"
+	c, err := NewCopier(cid,
+		map[string]io.Reader{
+			"stdout": &stdout,
+			"stderr": &stderr,
+		},
+		jsonLog)
+	if err != nil {
+		t.Fatal(err)
+	}
+	c.Run()
+	time.Sleep(100 * time.Millisecond)
+	dec := json.NewDecoder(&jsonBuf)
+	for {
+		var msg Message
+		if err := dec.Decode(&msg); err != nil {
+			if err == io.EOF {
+				break
+			}
+			t.Fatal(err)
+		}
+		if msg.Source != "stdout" && msg.Source != "stderr" {
+			t.Fatalf("Wrong Source: %q, should be %q or %q", msg.Source, "stdout", "stderr")
+		}
+		if msg.ContainerID != cid {
+			t.Fatalf("Wrong ContainerID: %q, expected %q", msg.ContainerID, cid)
+		}
+		if msg.Source == "stdout" {
+			if string(msg.Line) != stdoutLine {
+				t.Fatalf("Wrong Line: %q, expected %q", msg.Line, stdoutLine)
+			}
+		}
+		if msg.Source == "stderr" {
+			if string(msg.Line) != stderrLine {
+				t.Fatalf("Wrong Line: %q, expected %q", msg.Line, stderrLine)
+			}
+		}
+	}
+}

daemon/logger/jsonfilelog/jsonfilelog.go

@@ -0,0 +1,49 @@
+package jsonfilelog
+
+import (
+	"bytes"
+	"os"
+
+	"github.com/docker/docker/daemon/logger"
+	"github.com/docker/docker/pkg/jsonlog"
+)
+
+// JSONFileLogger is Logger implementation for default docker logging:
+// JSON objects to file
+type JSONFileLogger struct {
+	buf *bytes.Buffer
+	f   *os.File // store for closing
+}
+
+// New creates new JSONFileLogger which writes to filename
+func New(filename string) (logger.Logger, error) {
+	log, err := os.OpenFile(filename, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0600)
+	if err != nil {
+		return nil, err
+	}
+	return &JSONFileLogger{
+		f:   log,
+		buf: bytes.NewBuffer(nil),
+	}, nil
+}
+
+// Log converts logger.Message to jsonlog.JSONLog and serializes it to file
+func (l *JSONFileLogger) Log(msg *logger.Message) error {
+	err := (&jsonlog.JSONLog{Log: string(msg.Line) + "\n", Stream: msg.Source, Created: msg.Timestamp}).MarshalJSONBuf(l.buf)
+	if err != nil {
+		return err
+	}
+	l.buf.WriteByte('\n')
+	_, err = l.buf.WriteTo(l.f)
+	return err
+}
+
+// Close closes underlying file
+func (l *JSONFileLogger) Close() error {
+	return l.f.Close()
+}
+
+// Name returns name of this logger
+func (l *JSONFileLogger) Name() string {
+	return "JSONFile"
+}

daemon/logger/jsonfilelog/jsonfilelog_test.go

@@ -0,0 +1,78 @@
+package jsonfilelog
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/docker/docker/daemon/logger"
+	"github.com/docker/docker/pkg/jsonlog"
+)
+
+func TestJSONFileLogger(t *testing.T) {
+	tmp, err := ioutil.TempDir("", "docker-logger-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+	filename := filepath.Join(tmp, "container.log")
+	l, err := New(filename)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+	cid := "a7317399f3f857173c6179d44823594f8294678dea9999662e5c625b5a1c7657"
+	if err := l.Log(&logger.Message{ContainerID: cid, Line: []byte("line1"), Source: "src1"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := l.Log(&logger.Message{ContainerID: cid, Line: []byte("line2"), Source: "src2"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := l.Log(&logger.Message{ContainerID: cid, Line: []byte("line3"), Source: "src3"}); err != nil {
+		t.Fatal(err)
+	}
+	res, err := ioutil.ReadFile(filename)
+	if err != nil {
+		t.Fatal(err)
+	}
+	expected := `{"log":"line1\n","stream":"src1","time":"0001-01-01T00:00:00Z"}
+{"log":"line2\n","stream":"src2","time":"0001-01-01T00:00:00Z"}
+{"log":"line3\n","stream":"src3","time":"0001-01-01T00:00:00Z"}
+`
+
+	if string(res) != expected {
+		t.Fatalf("Wrong log content: %q, expected %q", res, expected)
+	}
+}
+
+func BenchmarkJSONFileLogger(b *testing.B) {
+	tmp, err := ioutil.TempDir("", "docker-logger-")
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+	filename := filepath.Join(tmp, "container.log")
+	l, err := New(filename)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer l.Close()
+	cid := "a7317399f3f857173c6179d44823594f8294678dea9999662e5c625b5a1c7657"
+	testLine := "Line that thinks that it is log line from docker\n"
+	msg := &logger.Message{ContainerID: cid, Line: []byte(testLine), Source: "stderr", Timestamp: time.Now().UTC()}
+	jsonlog, err := (&jsonlog.JSONLog{Log: string(msg.Line) + "\n", Stream: msg.Source, Created: msg.Timestamp}).MarshalJSON()
+	if err != nil {
+		b.Fatal(err)
+	}
+	b.SetBytes(int64(len(jsonlog)+1) * 30)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 30; j++ {
+			if err := l.Log(msg); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+}

daemon/logger/logger.go

@@ -0,0 +1,18 @@
+package logger
+
+import "time"
+
+// Message is datastructure that represents record from some container
+type Message struct {
+	ContainerID string
+	Line        []byte
+	Source      string
+	Timestamp   time.Time
+}
+
+// Logger is interface for docker logging drivers
+type Logger interface {
+	Log(*Message) error
+	Name() string
+	Close() error
+}

daemon/logs.go

@@ -44,6 +44,9 @@ func (daemon *Daemon) ContainerLogs(job *engine.Job) engine.Status {
 	if err != nil {
 		return job.Error(err)
 	}
+	if container.LogDriverType() != "json-file" {
+		return job.Errorf("\"logs\" endpoint is supported only for \"json-file\" logging driver")
+	}
 	cLog, err := container.ReadLog("json")
 	if err != nil && os.IsNotExist(err) {
 		// Legacy logs

daemon/monitor.go

@@ -123,7 +123,7 @@ func (m *containerMonitor) Start() error {
 	for {
 		m.container.RestartCount++
 
-		if err := m.container.startLoggingToDisk(); err != nil {
+		if err := m.container.startLogging(); err != nil {
 			m.resetContainer(false)
 
 			return err
@@ -302,6 +302,11 @@ func (m *containerMonitor) resetContainer(lock bool) {
 		container.stdin, container.stdinPipe = io.Pipe()
 	}
 
+	if container.logDriver != nil {
+		container.logDriver.Close()
+		container.logDriver = nil
+	}
+
 	c := container.command.ProcessConfig.Cmd
 
 	container.command.ProcessConfig.Cmd = exec.Cmd{