From 85319318eca83a2397aec1c449018f3b2bd15c01 Mon Sep 17 00:00:00 2001 From: Brian Goff Date: Fri, 30 Jun 2017 13:13:32 -0400 Subject: [PATCH] Fix log readers can block writes indefinitely Before this patch, a log reader is able to block all log writes indefinitely (and other operations) by simply opening the log stream and not consuming all the messages. The reason for this is we protect the read stream from corruption by ensuring there are no new writes while the log stream is consumed (and caught up with the live entries). We can get around this issue because log files are append only, so we can limit reads to only the section of the file that was written to when the log stream was first requested. Now logs are only blocked until all files are opened, rather than streamed to the client. Signed-off-by: Brian Goff (cherry picked from commit e2209185ed1c959131d4068ec7fc93e194dc0802) Conflicts: daemon/logger/jsonfilelog/jsonfilelog.go daemon/logger/jsonfilelog/read.go Signed-off-by: Andrew Hsu Signed-off-by: Brian Goff --- daemon/logger/jsonfilelog/jsonfilelog.go | 51 +++++++++++++++--------- daemon/logger/jsonfilelog/read.go | 51 +++++++++++++++--------- 2 files changed, 66 insertions(+), 36 deletions(-) diff --git a/daemon/logger/jsonfilelog/jsonfilelog.go b/daemon/logger/jsonfilelog/jsonfilelog.go index a429a08a4f..fa6f9bf8a2 100644 --- a/daemon/logger/jsonfilelog/jsonfilelog.go +++ b/daemon/logger/jsonfilelog/jsonfilelog.go @@ -7,6 +7,7 @@ import ( "bytes" "encoding/json" "fmt" + "io" "strconv" "sync" @@ -15,6 +16,7 @@ import ( "github.com/docker/docker/daemon/logger/loggerutils" "github.com/docker/docker/pkg/jsonlog" "github.com/docker/go-units" + "github.com/pkg/errors" ) // Name is the name of the file that the jsonlogger logs to. @@ -22,11 +24,13 @@ const Name = "json-file" // JSONFileLogger is Logger implementation for default Docker logging. type JSONFileLogger struct { - buf *bytes.Buffer + extra []byte // json-encoded extra attributes + + mu sync.RWMutex + buf *bytes.Buffer // avoids allocating a new buffer on each call to `Log()` + closed bool writer *loggerutils.RotateFileWriter - mu sync.Mutex readers map[*logger.LogWatcher]struct{} // stores the active log followers - extra []byte // json-encoded extra attributes } func init() { @@ -85,32 +89,43 @@ func New(ctx logger.Context) (logger.Logger, error) { // Log converts logger.Message to jsonlog.JSONLog and serializes it to file. func (l *JSONFileLogger) Log(msg *logger.Message) error { + l.mu.Lock() + err := writeMessageBuf(l.writer, msg, l.extra, l.buf) + l.buf.Reset() + l.mu.Unlock() + return err +} + +func writeMessageBuf(w io.Writer, m *logger.Message, extra json.RawMessage, buf *bytes.Buffer) error { + if err := marshalMessage(m, extra, buf); err != nil { + return err + } + if _, err := w.Write(buf.Bytes()); err != nil { + return errors.Wrap(err, "error writing log entry") + } + return nil +} + +func marshalMessage(msg *logger.Message, extra json.RawMessage, buf *bytes.Buffer) error { timestamp, err := jsonlog.FastTimeMarshalJSON(msg.Timestamp) if err != nil { return err } - l.mu.Lock() - logline := msg.Line + logLine := msg.Line if !msg.Partial { - logline = append(msg.Line, '\n') + logLine = append(msg.Line, '\n') } err = (&jsonlog.JSONLogs{ - Log: logline, + Log: logLine, Stream: msg.Source, Created: timestamp, - RawAttrs: l.extra, - }).MarshalJSONBuf(l.buf) + RawAttrs: extra, + }).MarshalJSONBuf(buf) if err != nil { - l.mu.Unlock() - return err + return errors.Wrap(err, "error writing log message to buffer") } - - l.buf.WriteByte('\n') - _, err = l.writer.Write(l.buf.Bytes()) - l.buf.Reset() - l.mu.Unlock() - - return err + err = buf.WriteByte('\n') + return errors.Wrap(err, "error finalizing log buffer") } // ValidateLogOpt looks for json specific log options max-file & max-size. diff --git a/daemon/logger/jsonfilelog/read.go b/daemon/logger/jsonfilelog/read.go index 30d533fc1f..25cd1f7c37 100644 --- a/daemon/logger/jsonfilelog/read.go +++ b/daemon/logger/jsonfilelog/read.go @@ -3,7 +3,6 @@ package jsonfilelog import ( "bytes" "encoding/json" - "errors" "fmt" "io" "os" @@ -18,6 +17,7 @@ import ( "github.com/docker/docker/pkg/ioutils" "github.com/docker/docker/pkg/jsonlog" "github.com/docker/docker/pkg/tailfile" + "github.com/pkg/errors" ) const maxJSONDecodeRetry = 20000 @@ -48,10 +48,11 @@ func (l *JSONFileLogger) ReadLogs(config logger.ReadConfig) *logger.LogWatcher { func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.ReadConfig) { defer close(logWatcher.Msg) - // lock so the read stream doesn't get corrupted due to rotations or other log data written while we read + // lock so the read stream doesn't get corrupted due to rotations or other log data written while we open these files // This will block writes!!! - l.mu.Lock() + l.mu.RLock() + // TODO it would be nice to move a lot of this reader implementation to the rotate logger object pth := l.writer.LogPath() var files []io.ReadSeeker for i := l.writer.MaxFiles(); i > 1; i-- { @@ -59,25 +60,36 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R if err != nil { if !os.IsNotExist(err) { logWatcher.Err <- err - break + l.mu.RUnlock() + return } continue } defer f.Close() - files = append(files, f) } latestFile, err := os.Open(pth) if err != nil { - logWatcher.Err <- err - l.mu.Unlock() + logWatcher.Err <- errors.Wrap(err, "error opening latest log file") + l.mu.RUnlock() return } defer latestFile.Close() + latestChunk, err := newSectionReader(latestFile) + + // Now we have the reader sectioned, all fd's opened, we can unlock. + // New writes/rotates will not affect seeking through these files + l.mu.RUnlock() + + if err != nil { + logWatcher.Err <- err + return + } + if config.Tail != 0 { - tailer := ioutils.MultiReadSeeker(append(files, latestFile)...) + tailer := ioutils.MultiReadSeeker(append(files, latestChunk)...) tailFile(tailer, logWatcher, config.Tail, config.Since) } @@ -88,29 +100,32 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R } } - if !config.Follow { - if err := latestFile.Close(); err != nil { - logrus.Errorf("Error closing file: %v", err) - } - l.mu.Unlock() + if !config.Follow || l.closed { return } - if config.Tail >= 0 { - latestFile.Seek(0, os.SEEK_END) - } + notifyRotate := l.writer.NotifyRotate() + defer l.writer.NotifyRotateEvict(notifyRotate) + l.mu.Lock() l.readers[logWatcher] = struct{}{} l.mu.Unlock() - notifyRotate := l.writer.NotifyRotate() followLogs(latestFile, logWatcher, notifyRotate, config.Since) l.mu.Lock() delete(l.readers, logWatcher) l.mu.Unlock() +} - l.writer.NotifyRotateEvict(notifyRotate) +func newSectionReader(f *os.File) (*io.SectionReader, error) { + // seek to the end to get the size + // we'll leave this at the end of the file since section reader does not advance the reader + size, err := f.Seek(0, os.SEEK_END) + if err != nil { + return nil, errors.Wrap(err, "error getting current file size") + } + return io.NewSectionReader(f, 0, size), nil } func tailFile(f io.ReadSeeker, logWatcher *logger.LogWatcher, tail int, since time.Time) {