moby · thaJeztah · May 19, 2022 · Feb 25, 2022 · Mar 2, 2022 · Mar 2, 2022
diff --git a/daemon/logger/adapter_test.go b/daemon/logger/adapter_test.go
@@ -174,7 +174,7 @@ func TestAdapterReadLogs(t *testing.T) {
 		t.Fatal("timeout waiting for message channel to close")

 	}
-	lw.ProducerGone()
+	lw.ConsumerGone()

 	lw = lr.ReadLogs(ReadConfig{Follow: true})
 	for _, x := range testMsg {

diff --git a/daemon/logger/journald/journald.go b/daemon/logger/journald/journald.go
@@ -8,7 +8,6 @@ package journald // import "github.com/docker/docker/daemon/logger/journald"
 import (
 	"fmt"
 	"strconv"
-	"sync"
 	"unicode"

 	"github.com/coreos/go-systemd/v22/journal"
@@ -19,9 +18,9 @@ import (
 const name = "journald"

 type journald struct {
-	mu      sync.Mutex        //nolint:structcheck,unused
-	vars    map[string]string // additional variables and values to send to the journal along with the log message
-	readers map[*logger.LogWatcher]struct{}
+	vars map[string]string // additional variables and values to send to the journal along with the log message
+
+	closed chan struct{}
 }

 func init() {
@@ -81,7 +80,7 @@ func New(info logger.Info) (logger.Logger, error) {
 	for k, v := range extraAttrs {
 		vars[k] = v
 	}
-	return &journald{vars: vars, readers: make(map[*logger.LogWatcher]struct{})}, nil
+	return &journald{vars: vars, closed: make(chan struct{})}, nil
 }

 // We don't actually accept any options, but we have to supply a callback for
@@ -128,3 +127,8 @@ func (s *journald) Log(msg *logger.Message) error {
 func (s *journald) Name() string {
 	return name
 }
+
+func (s *journald) Close() error {
+	close(s.closed)
+	return nil
+}
diff --git a/daemon/logger/journald/read.go b/daemon/logger/journald/read.go
@@ -116,16 +116,6 @@ import (
 	"github.com/sirupsen/logrus"
 )

-func (s *journald) Close() error {
-	s.mu.Lock()
-	for r := range s.readers {
-		r.ProducerGone()
-		delete(s.readers, r)
-	}
-	s.mu.Unlock()
-	return nil
-}
-
 // CErr converts error code returned from a sd_journal_* function
 // (which returns -errno) to a string
 func CErr(ret C.int) string {
@@ -233,22 +223,20 @@ drain:
 }

 func (s *journald) followJournal(logWatcher *logger.LogWatcher, j *C.sd_journal, cursor *C.char, untilUnixMicro uint64) *C.char {
-	s.mu.Lock()
-	s.readers[logWatcher] = struct{}{}
-	s.mu.Unlock()
+	defer close(logWatcher.Msg)

 	waitTimeout := C.uint64_t(250000) // 0.25s

 	for {
 		status := C.sd_journal_wait(j, waitTimeout)
 		if status < 0 {
 			logWatcher.Err <- errors.New("error waiting for journal: " + CErr(status))
-			goto cleanup
+			break
 		}
 		select {
 		case <-logWatcher.WatchConsumerGone():
-			goto cleanup // won't be able to write anything anymore
-		case <-logWatcher.WatchProducerGone():
+			break // won't be able to write anything anymore
+		case <-s.closed:
 			// container is gone, drain journal
 		default:
 			// container is still alive
@@ -264,11 +252,6 @@ func (s *journald) followJournal(logWatcher *logger.LogWatcher, j *C.sd_journal,
 		}
 	}

-cleanup:
-	s.mu.Lock()
-	delete(s.readers, logWatcher)
-	s.mu.Unlock()
-	close(logWatcher.Msg)
 	return cursor
 }


diff --git a/daemon/logger/journald/read_unsupported.go b/daemon/logger/journald/read_unsupported.go
diff --git a/daemon/logger/jsonfilelog/jsonfilelog.go b/daemon/logger/jsonfilelog/jsonfilelog.go
@@ -8,7 +8,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"strconv"
-	"sync"

 	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/docker/daemon/logger/jsonfilelog/jsonlog"
@@ -22,11 +21,8 @@ const Name = "json-file"

 // JSONFileLogger is Logger implementation for default Docker logging.
 type JSONFileLogger struct {
-	mu      sync.Mutex
-	closed  bool
-	writer  *loggerutils.LogFile
-	readers map[*logger.LogWatcher]struct{} // stores the active log followers
-	tag     string                          // tag values requested by the user to log
+	writer *loggerutils.LogFile
+	tag    string // tag values requested by the user to log
 }

 func init() {
@@ -115,18 +111,14 @@ func New(info logger.Info) (logger.Logger, error) {
 	}

 	return &JSONFileLogger{
-		writer:  writer,
-		readers: make(map[*logger.LogWatcher]struct{}),
-		tag:     tag,
+		writer: writer,
+		tag:    tag,
 	}, nil
 }

 // Log converts logger.Message to jsonlog.JSONLog and serializes it to file.
 func (l *JSONFileLogger) Log(msg *logger.Message) error {
-	l.mu.Lock()
-	err := l.writer.WriteLogEntry(msg)
-	l.mu.Unlock()
-	return err
+	return l.writer.WriteLogEntry(msg)
 }

 func marshalMessage(msg *logger.Message, extra json.RawMessage, buf *bytes.Buffer) error {
@@ -169,15 +161,7 @@ func ValidateLogOpt(cfg map[string]string) error {
 // Close closes underlying file and signals all the readers
 // that the logs producer is gone.
 func (l *JSONFileLogger) Close() error {
-	l.mu.Lock()
-	l.closed = true
-	err := l.writer.Close()
-	for r := range l.readers {
-		r.ProducerGone()
-		delete(l.readers, r)
-	}
-	l.mu.Unlock()
-	return err
+	return l.writer.Close()
 }

 // Name returns name of this logger.

diff --git a/daemon/logger/jsonfilelog/read.go b/daemon/logger/jsonfilelog/read.go
@@ -10,32 +10,12 @@ import (
 	"github.com/docker/docker/daemon/logger/jsonfilelog/jsonlog"
 	"github.com/docker/docker/daemon/logger/loggerutils"
 	"github.com/docker/docker/pkg/tailfile"
-	"github.com/sirupsen/logrus"
 )

-const maxJSONDecodeRetry = 20000
-
 // ReadLogs implements the logger's LogReader interface for the logs
 // created by this driver.
 func (l *JSONFileLogger) ReadLogs(config logger.ReadConfig) *logger.LogWatcher {
-	logWatcher := logger.NewLogWatcher()
-
-	go l.readLogs(logWatcher, config)
-	return logWatcher
-}
-
-func (l *JSONFileLogger) readLogs(watcher *logger.LogWatcher, config logger.ReadConfig) {
-	defer close(watcher.Msg)
-
-	l.mu.Lock()
-	l.readers[watcher] = struct{}{}
-	l.mu.Unlock()
-
-	l.writer.ReadLogs(config, watcher)
-
-	l.mu.Lock()
-	delete(l.readers, watcher)
-	l.mu.Unlock()
+	return l.writer.ReadLogs(config)
 }

 func decodeLogLine(dec *json.Decoder, l *jsonlog.JSONLog) (*logger.Message, error) {
@@ -61,10 +41,9 @@ func decodeLogLine(dec *json.Decoder, l *jsonlog.JSONLog) (*logger.Message, erro
 }

 type decoder struct {
-	rdr      io.Reader
-	dec      *json.Decoder
-	jl       *jsonlog.JSONLog
-	maxRetry int
+	rdr io.Reader
+	dec *json.Decoder
+	jl  *jsonlog.JSONLog
 }

 func (d *decoder) Reset(rdr io.Reader) {
@@ -88,74 +67,7 @@ func (d *decoder) Decode() (msg *logger.Message, err error) {
 	if d.jl == nil {
 		d.jl = &jsonlog.JSONLog{}
 	}
-	if d.maxRetry == 0 {
-		// We aren't using maxJSONDecodeRetry directly so we can give a custom value for testing.
-		d.maxRetry = maxJSONDecodeRetry
-	}
-	for retries := 0; retries < d.maxRetry; retries++ {
-		msg, err = decodeLogLine(d.dec, d.jl)
-		if err == nil || err == io.EOF {
-			break
-		}
-
-		logrus.WithError(err).WithField("retries", retries).Warn("got error while decoding json")
-		// try again, could be due to a an incomplete json object as we read
-		if _, ok := err.(*json.SyntaxError); ok {
-			d.dec = json.NewDecoder(d.rdr)
-			continue
-		}
-
-		// 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 {
-			d.rdr = combineReaders(d.dec.Buffered(), d.rdr)
-			d.dec = json.NewDecoder(d.rdr)
-			continue
-		}
-	}
-	return msg, err
-}
-
-func combineReaders(pre, rdr io.Reader) io.Reader {
-	return &combinedReader{pre: pre, rdr: rdr}
-}
-
-// combinedReader is a reader which is like `io.MultiReader` where except it does not cache a full EOF.
-// Once `io.MultiReader` returns EOF, it is always EOF.
-//
-// For this usecase we have an underlying reader which is a file which may reach EOF but have more data written to it later.
-// As such, io.MultiReader does not work for us.
-type combinedReader struct {
-	pre io.Reader
-	rdr io.Reader
-}
-
-func (r *combinedReader) Read(p []byte) (int, error) {
-	var read int
-	if r.pre != nil {
-		n, err := r.pre.Read(p)
-		if err != nil {
-			if err != io.EOF {
-				return n, err
-			}
-			r.pre = nil
-		}
-		read = n
-	}
-
-	if read < len(p) {
-		n, err := r.rdr.Read(p[read:])
-		if n > 0 {
-			read += n
-		}
-		if err != nil {
-			return read, err
-		}
-	}
-
-	return read, nil
+	return decodeLogLine(d.dec, d.jl)
 }

 // decodeFunc is used to create a decoder for the log file reader