LogFile follow without filenotify
- Fixes #30046
- fixes #37646
- fixes #39274

**- What I did**
I fixed issues with log reading on Windows and other platforms.

**- How I did it**
I added an extensive test suite for validating the behavior of `logger.LogReader` implementations. The tests caught bugs in both the `local` driver and the LogFile follow implementation, all of which I have fixed. In order to make the tests repeatable, I had to synchronize the opening of a log watcher so that it consistently [happens before](https://en.wikipedia.org/wiki/Happened-before) the next message is logged. This required some refactoring of log readers and `logger.LogWatcher` so that the logging drivers no longer needed to keep track of active LogWatchers in order to signal when the logging driver is closed.

I modified how LogFile rotates log files, eliminating the need to evict readers or retry renames. Unlike POSIX, there is no direct equivalent to `unlink(3)` on Windows. A file's name is not available for reuse until the file is deleted, and a file is not deleted until all handles to it are closed. (`DeleteFile` only marks a file for delete on close.) If any readers had the oldest log file open when LogFile went to rotate logs and delete that file, renaming the second-oldest log file would fail on Windows. [A file name can be reused immediately by renaming the existing file](https://docs.microsoft.com/en-us/windows/win32/fileio/closing-and-deleting-files#:~:text=The%20DeleteFile%20function%20can%20be%20used%20to%20delete%20a%20file%20on%20close.%20A%20file%20cannot%20be%20deleted%20until%20all%20handles%20to%20it%20are%20closed.%20If%20a%20file%20cannot%20be%20deleted%2C%20its%20name%20cannot%20be%20reused.%20To%20reuse%20a%20file%20name%20immediately%2C%20rename%20the%20existing%20file.) so I have emulated `unlink` on Windows by renaming the file to a random name before marking it for deletion.

I made following when `max-file=1` less likely to miss any log entries. The existing implementation would truncate the log file, immediately erasing all existing log messages. If a follower had fallen behind, any messages between its read position and the rotation would be skipped. Instead of truncating, I changed it so the file is rotated as usual: the existing file is deleted and a new one is created in its place. Any followers which have the file open can continue to read from it uninterrupted, and the OS will free up the disk space as soon as the last follower is done with it.

I rewrote how LogFile follows logs. I replaced filenotify with intra-process communication between the log writer and followers. Compared to the previous attempt to remove filenotify, #19498, this implementation does not regress logging performance. The race condition where followers could read a partially-written log message, which was worked around by retrying the read until it succeeded, is eliminated by having the writer signal the size of the file when the write has completed and so followers know not read past that offset. The event fanout with channels pattern [from Rethinking Classical Concurrency Patterns (pages 102-3)](https://drive.google.com/file/d/1nPdvhB0PutEJzdCq5ms6UI58dp50fcAN/view) is used instead of `pkg/pubsub` as it is lighter weight, consuming no resources unless a follower is waiting for a new message to be written.

I replaced refCounter, used when reading compressed logs, with a new implementation which does not get confused when the compressed log files are renamed, as happens whenever the logs are rotated.

**- How to verify it**
The added unit tests verify that no regressions have been introduced. The new LogFile follow implementation can be verified to not regress performance using `daemon/logger/jsonfilelog.BenchmarkJSONFileLoggerReadLogs`. I have found that benchmarking with a fixed number of iterations (messages logged) yields measurements with significantly less variance than giving the framework a target runtime duration.

In my testing, I have measured slightly _improved_ throughput compared to master. Benchmarked on darwin with `go test ./daemon/logger/jsonfilelog -bench BenchmarkJSONFileLoggerReadLogs -benchtime 1000000x -count=16`:
```
name                       old time/op    new time/op    delta
JSONFileLoggerReadLogs-12    5.49µs ± 7%    5.28µs ±10%  -3.89%  (p=0.034 n=15+16)

name                       old speed      new speed      delta
JSONFileLoggerReadLogs-12  22.1MB/s ± 7%  23.0MB/s ±11%  +4.19%  (p=0.034 n=15+16)
```

Instructions for reproducing, and thus verifying the fix for, the issue following logs on Windows can be found in https://github.com/moby/moby/issues/39274#issuecomment-497966709

**- Description for the changelog**
- Improved reliability of log following on Windows.
- Newlines are no longer dropped at the end of long log lines when using the `local` log driver.

**- A picture of a cute animal (not mandatory but encouraged)**


daemon/logger/loggerutils/follow.go
package loggerutils // import "github.com/docker/docker/daemon/logger/loggerutils"

import (
	"io"
	"os"
	"time"

	"github.com/docker/docker/daemon/logger"
	"github.com/docker/docker/pkg/filenotify"
	"github.com/fsnotify/fsnotify"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

var errRetry = errors.New("retry")
var errDone = errors.New("done")

type follow struct {
	file                      *os.File
	dec                       Decoder
	fileWatcher               filenotify.FileWatcher
	logWatcher                *logger.LogWatcher
	notifyRotate, notifyEvict chan interface{}
	oldSize                   int64
	retries                   int
}

func (fl *follow) handleRotate() error {
	name := fl.file.Name()

	fl.file.Close()
	fl.fileWatcher.Remove(name)

	// retry when the file doesn't exist
	var err error
	for retries := 0; retries <= 5; retries++ {
		f, err := open(name)
		if err == nil || !os.IsNotExist(err) {
			fl.file = f
			break
		}
	}
	if err != nil {
		return err
	}
	if err := fl.fileWatcher.Add(name); err != nil {
		return err
	}
	fl.dec.Reset(fl.file)
	return nil
}

func (fl *follow) handleMustClose(evictErr error) {
	fl.file.Close()
	fl.dec.Close()
	fl.logWatcher.Err <- errors.Wrap(evictErr, "log reader evicted due to errors")
	logrus.WithField("file", fl.file.Name()).Error("Log reader notified that it must re-open log file, some log data may not be streamed to the client.")
}

func (fl *follow) waitRead() error {
	select {
	case e := <-fl.notifyEvict:
		if e != nil {
			err := e.(error)
			fl.handleMustClose(err)
		}
		return errDone
	case e := <-fl.fileWatcher.Events():
		switch e.Op {
		case fsnotify.Write:
			fl.dec.Reset(fl.file)
			return nil
		case fsnotify.Rename, fsnotify.Remove:
			select {
			case <-fl.notifyRotate:
			case <-fl.logWatcher.WatchProducerGone():
				return errDone
			case <-fl.logWatcher.WatchConsumerGone():
				return errDone
			}
			if err := fl.handleRotate(); err != nil {
				return err
			}
			return nil
		}
		return errRetry
	case err := <-fl.fileWatcher.Errors():
		logrus.Debugf("logger got error watching file: %v", err)
		// Something happened, let's try and stay alive and create a new watcher
		if fl.retries <= 5 {
			fl.fileWatcher.Close()
			fl.fileWatcher, err = watchFile(fl.file.Name())
			if err != nil {
				return err
			}
			fl.retries++
			return errRetry
		}
		return err
	case <-fl.logWatcher.WatchProducerGone():
		return errDone
	case <-fl.logWatcher.WatchConsumerGone():
		return errDone
	}
}

func (fl *follow) handleDecodeErr(err error) error {
	if !errors.Is(err, io.EOF) {
		return err
	}

	// Handle special case (#39235): max-file=1 and file was truncated
	st, stErr := fl.file.Stat()
	if stErr == nil {
		size := st.Size()
		defer func() { fl.oldSize = size }()
		if size < fl.oldSize { // truncated
			fl.file.Seek(0, 0)
			fl.dec.Reset(fl.file)
			return nil
		}
	} else {
		logrus.WithError(stErr).Warn("logger: stat error")
	}

	for {
		err := fl.waitRead()
		if err == nil {
			break
		}
		if err == errRetry {
			continue
		}
		return err
	}
	return nil
}

func (fl *follow) mainLoop(since, until time.Time) {
	for {
		select {
		case err := <-fl.notifyEvict:
			if err != nil {
				fl.handleMustClose(err.(error))
			}
			return
		default:
		}
		msg, err := fl.dec.Decode()
		if err != nil {
			if err := fl.handleDecodeErr(err); err != nil {
				if err == errDone {
					return
				}
				// we got an unrecoverable error, so return
				fl.logWatcher.Err <- err
				return
			}
			// ready to try again
			continue
		}

		fl.retries = 0 // reset retries since we've succeeded
		if !since.IsZero() && msg.Timestamp.Before(since) {
			continue
		}
		if !until.IsZero() && msg.Timestamp.After(until) {
			return
		}
		// send the message, unless the consumer is gone
		select {
		case e := <-fl.notifyEvict:
			if e != nil {
				err := e.(error)
				logrus.WithError(err).Debug("Reader evicted while sending log message")
				fl.logWatcher.Err <- err
			}
			return
		case fl.logWatcher.Msg <- msg:
		case <-fl.logWatcher.WatchConsumerGone():
			return
		}
	}
}

func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate, notifyEvict chan interface{}, dec Decoder, since, until time.Time) {
	dec.Reset(f)

	name := f.Name()
	fileWatcher, err := watchFile(name)
	if err != nil {
		logWatcher.Err <- err
		return
	}
	defer func() {
		f.Close()
		dec.Close()
		fileWatcher.Close()
	}()

	fl := &follow{
		file:         f,
		oldSize:      -1,
		logWatcher:   logWatcher,
		fileWatcher:  fileWatcher,
		notifyRotate: notifyRotate,
		notifyEvict:  notifyEvict,
		dec:          dec,
	}
	fl.mainLoop(since, until)
}

package loggerutils // import "github.com/docker/docker/daemon/logger/loggerutils"

import (
	"fmt"
	"io"
	"os"
	"time"

	"github.com/docker/docker/daemon/logger"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

type follow struct {
	LogFile      *LogFile
	Watcher      *logger.LogWatcher
	Decoder      Decoder
	Since, Until time.Time

	log *logrus.Entry
	c   chan logPos
}

// Do follows the log file as it is written, starting from f at read.
func (fl *follow) Do(f *os.File, read logPos) {
	fl.log = logrus.WithFields(logrus.Fields{
		"module": "logger",
		"file":   f.Name(),
	})
	// Optimization: allocate the write-notifications channel only once and
	// reuse it for multiple invocations of nextPos().
	fl.c = make(chan logPos, 1)

	defer func() {
		if err := f.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
			fl.log.WithError(err).Warn("error closing current log file")
		}
	}()

	for {
		wrote, ok := fl.nextPos(read)
		if !ok {
			return
		}

		if wrote.rotation != read.rotation {
			// Flush the current file before moving on to the next.
			if _, err := f.Seek(read.size, io.SeekStart); err != nil {
				fl.Watcher.Err <- err
				return
			}
			if fl.decode(f) {
				return
			}

			// Open the new file, which has the same name as the old
			// file thanks to file rotation. Make no mistake: they
			// are different files, with distinct identities.
			// Atomically capture the wrote position to make
			// absolutely sure that the position corresponds to the
			// file we have opened; more rotations could have
			// occurred since we previously received it.
			if err := f.Close(); err != nil {
				fl.log.WithError(err).Warn("error closing rotated log file")
			}
			var err error
			func() {
				fl.LogFile.fsopMu.RLock()
				st := <-fl.LogFile.read
				defer func() {
					fl.LogFile.read <- st
					fl.LogFile.fsopMu.RUnlock()
				}()
				f, err = open(f.Name())
				wrote = st.pos
			}()
			// We tried to open the file inside a critical section
			// so we shouldn't have been racing the rotation of the
			// file. Any error, even fs.ErrNotFound, is exceptional.
			if err != nil {
				fl.Watcher.Err <- fmt.Errorf("logger: error opening log file for follow after rotation: %w", err)
				return
			}

			if nrot := wrote.rotation - read.rotation; nrot > 1 {
				fl.log.WithField("missed-rotations", nrot).
					Warn("file rotations were missed while following logs; some log messages have been skipped over")
			}

			// Set up our read position to start from the top of the file.
			read.size = 0
		}

		if fl.decode(io.NewSectionReader(f, read.size, wrote.size-read.size)) {
			return
		}
		read = wrote
	}
}

// nextPos waits until the write position of the LogFile being followed has
// advanced from current and returns the new position.
func (fl *follow) nextPos(current logPos) (next logPos, ok bool) {
	var st logReadState
	select {
	case <-fl.Watcher.WatchConsumerGone():
		return current, false
	case st = <-fl.LogFile.read:
	}

	// Have any any logs been written since we last checked?
	if st.pos == current { // Nope.
		// Add ourself to the notify list.
		st.wait = append(st.wait, fl.c)
	} else { // Yes.
		// "Notify" ourself immediately.
		fl.c <- st.pos
	}
	fl.LogFile.read <- st

	select {
	case <-fl.LogFile.closed: // No more logs will be written.
		select { // Have we followed to the end?
		case next = <-fl.c: // No: received a new position.
		default: // Yes.
			return current, false
		}
	case <-fl.Watcher.WatchConsumerGone():
		return current, false
	case next = <-fl.c:
	}
	return next, true
}

// decode decodes log messages from r and sends messages with timestamps between
// Since and Until to the log watcher.
//
// The return value, done, signals whether following should end due to a
// condition encountered during decode.
func (fl *follow) decode(r io.Reader) (done bool) {
	fl.Decoder.Reset(r)
	for {
		msg, err := fl.Decoder.Decode()
		if err != nil {
			if errors.Is(err, io.EOF) {
				return false
			}
			fl.Watcher.Err <- err
			return true
		}

		if !fl.Since.IsZero() && msg.Timestamp.Before(fl.Since) {
			continue
		}
		if !fl.Until.IsZero() && msg.Timestamp.After(fl.Until) {
			return true
		}
		// send the message, unless the consumer is gone
		select {
		case fl.Watcher.Msg <- msg:
		case <-fl.Watcher.WatchConsumerGone():
			return true
		}
	}
}

`done=true` signals that a terminal stopping condition for following logs has been encountered while decoding: unexpected error, watch consumer is gone, or the `docker logs --until` condition has been reached. EOF means that the follower has caught up to the log writer or has reached the end of a rotated log file, neither of which is a reason to stop following.