From 84e60a7e10278e3acd2b783d0e6955dc5198b57c Mon Sep 17 00:00:00 2001 From: Brian Goff Date: Mon, 11 Jul 2016 16:31:42 -0400 Subject: [PATCH] Fix issues with tailing rotated jsonlog file Fixes a race where the log reader would get events for both an actual rotation as we from fsnotify (`fsnotify.Rename`). This issue becomes extremely apparent when rotations are fast, for example: ``` $ docker run -d --name test --log-opt max-size=1 --log-opt max-file=2 busybox sh -c 'while true; do echo hello; usleep 100000; done' ``` With this change the log reader for jsonlogs can handle rotations that happen as above. Instead of listening for both fs events AND rotation events simultaneously, potentially meaning we see 2 rotations for only a single rotation due to channel buffering, only listen for fs events (like `Rename`) and then wait to be notified about rotation by the logger. This makes sure that we don't see 2 rotations for 1, and that we don't start trying to read until the logger is actually ready for us to. Signed-off-by: Brian Goff --- daemon/logger/jsonfilelog/read.go | 196 ++++++++++++++++++++---------- 1 file changed, 130 insertions(+), 66 deletions(-) diff --git a/daemon/logger/jsonfilelog/read.go b/daemon/logger/jsonfilelog/read.go index bea83ddf14..a46d177be5 100644 --- a/daemon/logger/jsonfilelog/read.go +++ b/daemon/logger/jsonfilelog/read.go @@ -3,11 +3,14 @@ package jsonfilelog import ( "bytes" "encoding/json" + "errors" "fmt" "io" "os" "time" + "gopkg.in/fsnotify.v1" + "github.com/Sirupsen/logrus" "github.com/docker/docker/daemon/logger" "github.com/docker/docker/pkg/filenotify" @@ -44,6 +47,10 @@ 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 + // This will block writes!!! + l.mu.Lock() + pth := l.writer.LogPath() var files []io.ReadSeeker for i := l.writer.MaxFiles(); i > 1; i-- { @@ -61,6 +68,7 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R latestFile, err := os.Open(pth) if err != nil { logWatcher.Err <- err + l.mu.Unlock() return } @@ -80,6 +88,7 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R if err := latestFile.Close(); err != nil { logrus.Errorf("Error closing file: %v", err) } + l.mu.Unlock() return } @@ -87,7 +96,6 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R latestFile.Seek(0, os.SEEK_END) } - l.mu.Lock() l.readers[logWatcher] = struct{}{} l.mu.Unlock() @@ -128,19 +136,11 @@ func tailFile(f io.ReadSeeker, logWatcher *logger.LogWatcher, tail int, since ti } } -func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) { - dec := json.NewDecoder(f) - l := &jsonlog.JSONLog{} - +func watchFile(name string) (filenotify.FileWatcher, error) { fileWatcher, err := filenotify.New() if err != nil { - logWatcher.Err <- err + return nil, err } - defer func() { - f.Close() - fileWatcher.Close() - }() - name := f.Name() if err := fileWatcher.Add(name); err != nil { logrus.WithField("logger", "json-file").Warnf("falling back to file poller due to error: %v", err) @@ -148,72 +148,136 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int fileWatcher = filenotify.NewPollingWatcher() if err := fileWatcher.Add(name); err != nil { + fileWatcher.Close() logrus.Debugf("error watching log file for modifications: %v", err) - logWatcher.Err <- err - return + return nil, err + } + } + return fileWatcher, nil +} + +func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) { + dec := json.NewDecoder(f) + l := &jsonlog.JSONLog{} + + name := f.Name() + fileWatcher, err := watchFile(name) + if err != nil { + logWatcher.Err <- err + return + } + defer func() { + f.Close() + fileWatcher.Close() + }() + + var retries int + handleRotate := func() error { + f.Close() + fileWatcher.Remove(name) + + // retry when the file doesn't exist + for retries := 0; retries <= 5; retries++ { + f, err = os.Open(name) + if err == nil || !os.IsNotExist(err) { + break + } + } + if err != nil { + return err + } + if err := fileWatcher.Add(name); err != nil { + return err + } + dec = json.NewDecoder(f) + return nil + } + + errRetry := errors.New("retry") + errDone := errors.New("done") + waitRead := func() error { + select { + case e := <-fileWatcher.Events(): + switch e.Op { + case fsnotify.Write: + dec = json.NewDecoder(f) + return nil + case fsnotify.Rename, fsnotify.Remove: + select { + case <-notifyRotate: + case <-logWatcher.WatchClose(): + fileWatcher.Remove(name) + return errDone + } + if err := handleRotate(); err != nil { + return err + } + return nil + } + return errRetry + case err := <-fileWatcher.Errors(): + logrus.Debug("logger got error watching file: %v", err) + // Something happened, let's try and stay alive and create a new watcher + if retries <= 5 { + fileWatcher.Close() + fileWatcher, err = watchFile(name) + if err != nil { + return err + } + retries++ + return errRetry + } + return err + case <-logWatcher.WatchClose(): + fileWatcher.Remove(name) + return errDone } } - var retries int + handleDecodeErr := func(err error) error { + if err == io.EOF { + for err := waitRead(); err != nil; { + if err == errRetry { + // retry the waitRead + continue + } + return err + } + return nil + } + // try again because this shouldn't happen + if _, ok := err.(*json.SyntaxError); ok && retries <= maxJSONDecodeRetry { + dec = json.NewDecoder(f) + retries++ + return nil + } + // io.ErrUnexpectedEOF is returned from json.Decoder when there is + // remaining data in the parser's buffer while an io.EOF occurs. + // If the json logger writes a partial json log entry to the disk + // while at the same time the decoder tries to decode it, the race condition happens. + if err == io.ErrUnexpectedEOF && retries <= maxJSONDecodeRetry { + reader := io.MultiReader(dec.Buffered(), f) + dec = json.NewDecoder(reader) + retries++ + return nil + } + return err + } + + // main loop for { msg, err := decodeLogLine(dec, l) if err != nil { - if err != io.EOF { - // try again because this shouldn't happen - if _, ok := err.(*json.SyntaxError); ok && retries <= maxJSONDecodeRetry { - dec = json.NewDecoder(f) - retries++ - continue + if err := handleDecodeErr(err); err != nil { + if err == errDone { + return } - - // io.ErrUnexpectedEOF is returned from json.Decoder when there is - // remaining data in the parser's buffer while an io.EOF occurs. - // If the json logger writes a partial json log entry to the disk - // while at the same time the decoder tries to decode it, the race condition happens. - if err == io.ErrUnexpectedEOF && retries <= maxJSONDecodeRetry { - reader := io.MultiReader(dec.Buffered(), f) - dec = json.NewDecoder(reader) - retries++ - continue - } - - return - } - - select { - case <-fileWatcher.Events(): - dec = json.NewDecoder(f) - continue - case <-fileWatcher.Errors(): + // we got an unrecoverable error, so return logWatcher.Err <- err return - case <-logWatcher.WatchClose(): - fileWatcher.Remove(name) - return - case <-notifyRotate: - f.Close() - fileWatcher.Remove(name) - - // retry when the file doesn't exist - for retries := 0; retries <= 5; retries++ { - f, err = os.Open(name) - if err == nil || !os.IsNotExist(err) { - break - } - } - - if err = fileWatcher.Add(name); err != nil { - logWatcher.Err <- err - return - } - if err != nil { - logWatcher.Err <- err - return - } - - dec = json.NewDecoder(f) - continue } + // ready to try again + continue } retries = 0 // reset retries since we've succeeded