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 <cpuguy83@gmail.com>
(cherry picked from commit e2209185ed1c959131d4068ec7fc93e194dc0802)
Conflicts:
daemon/logger/jsonfilelog/jsonfilelog.go
daemon/logger/jsonfilelog/read.go
Signed-off-by: Andrew Hsu <andrewhsu@docker.com>
Signed-off-by: Brian Goff <cpuguy83@gmail.com>
diff --git a/daemon/logger/jsonfilelog/jsonfilelog.go b/daemon/logger/jsonfilelog/jsonfilelog.go
index a429a08..fa6f9bf 100644
--- a/daemon/logger/jsonfilelog/jsonfilelog.go
+++ b/daemon/logger/jsonfilelog/jsonfilelog.go
@@ -7,6 +7,7 @@
"bytes"
"encoding/json"
"fmt"
+ "io"
"strconv"
"sync"
@@ -15,6 +16,7 @@
"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 @@
// 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 @@
// 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 30d533f..25cd1f7 100644
--- a/daemon/logger/jsonfilelog/read.go
+++ b/daemon/logger/jsonfilelog/read.go
@@ -3,7 +3,6 @@
import (
"bytes"
"encoding/json"
- "errors"
"fmt"
"io"
"os"
@@ -18,6 +17,7 @@
"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(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 @@
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 @@
}
}
- 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) {
