Merge pull request #27782 from tombooth/27779-decouple-watcher-close

Decouple removing the fileWatcher from reading

daemon/logger/jsonfilelog/read.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"time"
 
+	"golang.org/x/net/context"
 	"gopkg.in/fsnotify.v1"
 
 	"github.com/Sirupsen/logrus"
@@ -113,7 +114,8 @@ func (l *JSONFileLogger) readLogs(logWatcher *logger.LogWatcher, config logger.R
 }
 
 func tailFile(f io.ReadSeeker, logWatcher *logger.LogWatcher, tail int, since time.Time) {
-	var rdr io.Reader = f
+	var rdr io.Reader
+	rdr = f
 	if tail > 0 {
 		ls, err := tailfile.TailFile(f, tail)
 		if err != nil {
@@ -171,9 +173,22 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int
 	}
 	defer func() {
 		f.Close()
+		fileWatcher.Remove(name)
 		fileWatcher.Close()
 	}()
 
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-logWatcher.WatchClose():
+			fileWatcher.Remove(name)
+			cancel()
+		case <-ctx.Done():
+			return
+		}
+	}()
+
 	var retries int
 	handleRotate := func() error {
 		f.Close()
@@ -208,8 +223,7 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int
 			case fsnotify.Rename, fsnotify.Remove:
 				select {
 				case <-notifyRotate:
-				case <-logWatcher.WatchClose():
-					fileWatcher.Remove(name)
+				case <-ctx.Done():
 					return errDone
 				}
 				if err := handleRotate(); err != nil {
@@ -231,8 +245,7 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int
 				return errRetry
 			}
 			return err
-		case <-logWatcher.WatchClose():
-			fileWatcher.Remove(name)
+		case <-ctx.Done():
 			return errDone
 		}
 	}
@@ -289,7 +302,7 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int
 		}
 		select {
 		case logWatcher.Msg <- msg:
-		case <-logWatcher.WatchClose():
+		case <-ctx.Done():
 			logWatcher.Msg <- msg
 			for {
 				msg, err := decodeLogLine(dec, l)