From 42fda5fe7ef5971ea6adec8b095270b351eb2dbf Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Wed, 29 Aug 2018 13:46:46 -0700 Subject: [PATCH 1/7] pkg/filenotify: poller.Add: fix fd leaks on err In case of errors, the file descriptor is never closed. Fix it. Signed-off-by: Kir Kolyshkin (cherry picked from commit 88bcf1573ca2eaffc15da346a1651a3749567554) Upstream-commit: 7be43586af6824c1e55cb502d9d2bab45c9b4505 Component: engine --- components/engine/pkg/filenotify/poller.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/components/engine/pkg/filenotify/poller.go b/components/engine/pkg/filenotify/poller.go index 22f1897034..7a9e1785c3 100644 --- a/components/engine/pkg/filenotify/poller.go +++ b/components/engine/pkg/filenotify/poller.go @@ -54,6 +54,7 @@ func (w *filePoller) Add(name string) error { } fi, err := os.Stat(name) if err != nil { + f.Close() return err } @@ -61,6 +62,7 @@ func (w *filePoller) Add(name string) error { w.watches = make(map[string]chan struct{}) } if _, exists := w.watches[name]; exists { + f.Close() return fmt.Errorf("watch exists") } chClose := make(chan struct{}) From abbd665e30d5a7581c95fecdeb07f6c8124773b8 Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Wed, 29 Aug 2018 13:50:01 -0700 Subject: [PATCH 2/7] pkg/filenotify/poller: close file asap There is no need to wait for up to 200ms in order to close the file descriptor once the chClose is received. This commit might reduce the chances for occasional "The process cannot access the file because it is being used by another process" error on Windows, where an opened file can't be removed. Signed-off-by: Kir Kolyshkin (cherry picked from commit dfbb64ea7d042d5b2bb0c1c2b88e3682b7069b10) Upstream-commit: 3a3bfcbf47e98212abfc9cfed860d9e99fc41cdc Component: engine --- components/engine/pkg/filenotify/poller.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/components/engine/pkg/filenotify/poller.go b/components/engine/pkg/filenotify/poller.go index 7a9e1785c3..8f6c310e1e 100644 --- a/components/engine/pkg/filenotify/poller.go +++ b/components/engine/pkg/filenotify/poller.go @@ -148,12 +148,11 @@ func (w *filePoller) sendErr(e error, chClose <-chan struct{}) error { func (w *filePoller) watch(f *os.File, lastFi os.FileInfo, chClose chan struct{}) { defer f.Close() for { - time.Sleep(watchWaitTime) select { + case <-time.After(watchWaitTime): case <-chClose: logrus.Debugf("watch for %s closed", f.Name()) return - default: } fi, err := os.Stat(f.Name()) From 21c28e45663511d9c2f529f2a39bc0855a9c245e Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Wed, 29 Aug 2018 22:11:42 -0700 Subject: [PATCH 3/7] pkg/filenotify/poller: fix Close() The code in Close() that removes the watches was not working, because it first sets `w.closed = true` and then calls w.close(), which starts with ``` if w.closed { return errPollerClosed } ``` Fix by setting w.closed only after calling w.remove() for all the files being watched. While at it, remove the duplicated `delete(w.watches, name)` code. Signed-off-by: Kir Kolyshkin (cherry picked from commit fffa8958d00860b4e3563327a2cc6836a12d4ba9) Upstream-commit: 4e2dbfa1af48191126b0910b9463bf94d8371886 Component: engine --- components/engine/pkg/filenotify/poller.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/components/engine/pkg/filenotify/poller.go b/components/engine/pkg/filenotify/poller.go index 8f6c310e1e..6161d4ab73 100644 --- a/components/engine/pkg/filenotify/poller.go +++ b/components/engine/pkg/filenotify/poller.go @@ -115,11 +115,10 @@ func (w *filePoller) Close() error { return nil } - w.closed = true for name := range w.watches { w.remove(name) - delete(w.watches, name) } + w.closed = true return nil } From e5cc6e7dc4e923df02151156e6ef9a73757f4fd2 Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Wed, 1 Aug 2018 06:50:45 +0300 Subject: [PATCH 4/7] daemon.ContainerLogs: minor debug logging cleanup This code has many return statements, for some of them the "end logs" or "end stream" message was not printed, giving the impression that this "for" loop never ended. Make sure that "begin logs" is to be followed by "end logs". Signed-off-by: Kir Kolyshkin (cherry picked from commit 2e4c2a6bf9cb47fd07e42f9c043024ed3dbcd04d) Upstream-commit: 2b8bc86679b7153bb4ace063a858637df0f16a2e Component: engine --- components/engine/daemon/logs.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/components/engine/daemon/logs.go b/components/engine/daemon/logs.go index 37ca4caf63..5e5c34992e 100644 --- a/components/engine/daemon/logs.go +++ b/components/engine/daemon/logs.go @@ -118,6 +118,8 @@ func (daemon *Daemon) ContainerLogs(ctx context.Context, containerName string, c defer close(messageChan) lg.Debug("begin logs") + defer lg.Debugf("end logs (%v)", ctx.Err()) + for { select { // i do not believe as the system is currently designed any error @@ -132,14 +134,12 @@ func (daemon *Daemon) ContainerLogs(ctx context.Context, containerName string, c } return case <-ctx.Done(): - lg.Debugf("logs: end stream, ctx is done: %v", ctx.Err()) return case msg, ok := <-logs.Msg: // there is some kind of pool or ring buffer in the logger that // produces these messages, and a possible future optimization // might be to use that pool and reuse message objects if !ok { - lg.Debug("end logs") return } m := msg.AsLogMessage() // just a pointer conversion, does not copy data From a76b67642d591253ba23ff89a796f28a09111db5 Mon Sep 17 00:00:00 2001 From: Brian Goff Date: Mon, 27 Aug 2018 13:53:23 -0700 Subject: [PATCH 5/7] daemon/logger/loggerutils: add TestFollowLogsClose This test case checks that followLogs() exits once the reader is gone. Currently it does not (i.e. this test is supposed to fail) due to #37391. [kolyshkin@: test case Brian Goff, changelog and all bugs are by me] Source: https://gist.github.com/cpuguy83/e538793de18c762608358ee0eaddc197 Signed-off-by: Kir Kolyshkin (cherry picked from commit d37a11bfbab83ab42b1160f116e863daac046192) Upstream-commit: 511741735e0aa2fe68a66d99384c00d187d1a157 Component: engine --- .../daemon/logger/loggerutils/logfile_test.go | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/components/engine/daemon/logger/loggerutils/logfile_test.go b/components/engine/daemon/logger/loggerutils/logfile_test.go index 0e359db3f8..c21aa258e6 100644 --- a/components/engine/daemon/logger/loggerutils/logfile_test.go +++ b/components/engine/daemon/logger/loggerutils/logfile_test.go @@ -4,6 +4,8 @@ import ( "bufio" "context" "io" + "io/ioutil" + "os" "strings" "testing" "time" @@ -74,3 +76,44 @@ func TestTailFiles(t *testing.T) { assert.Assert(t, string(msg.Line) == "Where we're going we don't need roads.", string(msg.Line)) } } + +func TestFollowLogsClose(t *testing.T) { + lw := logger.NewLogWatcher() + + f, err := ioutil.TempFile("", t.Name()) + assert.NilError(t, err) + defer func() { + f.Close() + os.Remove(f.Name()) + }() + + makeDecoder := func(rdr io.Reader) func() (*logger.Message, error) { + return func() (*logger.Message, error) { + return &logger.Message{}, nil + } + } + + followLogsDone := make(chan struct{}) + var since, until time.Time + go func() { + followLogs(f, lw, make(chan interface{}), makeDecoder, since, until) + close(followLogsDone) + }() + + select { + case <-lw.Msg: + case err := <-lw.Err: + assert.NilError(t, err) + case <-followLogsDone: + t.Fatal("follow logs finished unexpectedly") + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for log message") + } + + lw.Close() + select { + case <-followLogsDone: + case <-time.After(20 * time.Second): + t.Fatal("timeout waiting for followLogs() to finish") + } +} From 1a333bfe59625436fa4379085c121979f437a4bc Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Wed, 1 Aug 2018 07:03:55 +0300 Subject: [PATCH 6/7] daemon.ContainerLogs(): fix resource leak on follow When daemon.ContainerLogs() is called with options.follow=true (as in "docker logs --follow"), the "loggerutils.followLogs()" function never returns (even then the logs consumer is gone). As a result, all the resources associated with it (including an opened file descriptor for the log file being read, two FDs for a pipe, and two FDs for inotify watch) are never released. If this is repeated (such as by running "docker logs --follow" and pressing Ctrl-C a few times), this results in DoS caused by either hitting the limit of inotify watches, or the limit of opened files. The only cure is daemon restart. Apparently, what happens is: 1. logs producer (a container) is gone, calling (*LogWatcher).Close() for all its readers (daemon/logger/jsonfilelog/jsonfilelog.go:175). 2. WatchClose() is properly handled by a dedicated goroutine in followLogs(), cancelling the context. 3. Upon receiving the ctx.Done(), the code in followLogs() (daemon/logger/loggerutils/logfile.go#L626-L638) keeps to send messages _synchronously_ (which is OK for now). 4. Logs consumer is gone (Ctrl-C is pressed on a terminal running "docker logs --follow"). Method (*LogWatcher).Close() is properly called (see daemon/logs.go:114). Since it was called before and due to to once.Do(), nothing happens (which is kinda good, as otherwise it will panic on closing a closed channel). 5. A goroutine (see item 3 above) keeps sending log messages synchronously to the logWatcher.Msg channel. Since the channel reader is gone, the channel send operation blocks forever, and resource cleanup set up in defer statements at the beginning of followLogs() never happens. Alas, the fix is somewhat complicated: 1. Distinguish between close from logs producer and logs consumer. To that effect, - yet another channel is added to LogWatcher(); - {Watch,}Close() are renamed to {Watch,}ProducerGone(); - {Watch,}ConsumerGone() are added; *NOTE* that ProducerGone()/WatchProducerGone() pair is ONLY needed in order to stop ConsumerLogs(follow=true) when a container is stopped; otherwise we're not interested in it. In other words, we're only using it in followLogs(). 2. Code that was doing (logWatcher*).Close() is modified to either call ProducerGone() or ConsumerGone(), depending on the context. 3. Code that was waiting for WatchClose() is modified to wait for either ConsumerGone() or ProducerGone(), or both, depending on the context. 4. followLogs() are modified accordingly: - context cancellation is happening on WatchProducerGone(), and once it's received the FileWatcher is closed and waitRead() returns errDone on EOF (i.e. log rotation handling logic is disabled); - due to this, code that was writing synchronously to logWatcher.Msg can be and is removed as the code above it handles this case; - function returns once ConsumerGone is received, freeing all the resources -- this is the bugfix itself. While at it, 1. Let's also remove the ctx usage to simplify the code a bit. It was introduced by commit a69a59ffc7e3d ("Decouple removing the fileWatcher from reading") in order to fix a bug. The bug was actually a deadlock in fsnotify, and the fix was just a workaround. Since then the fsnofify bug has been fixed, and a new fsnotify was vendored in. For more details, please see https://github.com/moby/moby/pull/27782#issuecomment-416794490 2. Since `(*filePoller).Close()` is fixed to remove all the files being watched, there is no need to explicitly call fileWatcher.Remove(name) anymore, so get rid of the extra code. Should fix https://github.com/moby/moby/issues/37391 Signed-off-by: Kir Kolyshkin (cherry picked from commit 916eabd459fe707b5c4a86377d12e2ad1871b353) Upstream-commit: 84a5b528aede5579861201e869870d10fc98c07c Component: engine --- components/engine/daemon/attach.go | 2 +- components/engine/daemon/logger/adapter.go | 16 ++----- .../engine/daemon/logger/adapter_test.go | 2 +- .../engine/daemon/logger/journald/read.go | 4 +- .../daemon/logger/jsonfilelog/jsonfilelog.go | 5 +- .../daemon/logger/jsonfilelog/read_test.go | 3 +- .../engine/daemon/logger/local/local.go | 2 +- components/engine/daemon/logger/logger.go | 47 +++++++++++++------ .../daemon/logger/loggerutils/logfile.go | 41 ++++------------ .../daemon/logger/loggerutils/logfile_test.go | 4 +- components/engine/daemon/logs.go | 4 +- 11 files changed, 58 insertions(+), 72 deletions(-) diff --git a/components/engine/daemon/attach.go b/components/engine/daemon/attach.go index fb14691d24..0e12441ad1 100644 --- a/components/engine/daemon/attach.go +++ b/components/engine/daemon/attach.go @@ -123,7 +123,7 @@ func (daemon *Daemon) containerAttach(c *container.Container, cfg *stream.Attach return logger.ErrReadLogsNotSupported{} } logs := cLog.ReadLogs(logger.ReadConfig{Tail: -1}) - defer logs.Close() + defer logs.ConsumerGone() LogLoop: for { diff --git a/components/engine/daemon/logger/adapter.go b/components/engine/daemon/logger/adapter.go index 95aff9bf3b..d9370352c5 100644 --- a/components/engine/daemon/logger/adapter.go +++ b/components/engine/daemon/logger/adapter.go @@ -93,21 +93,12 @@ func (a *pluginAdapterWithRead) ReadLogs(config ReadConfig) *LogWatcher { dec := logdriver.NewLogEntryDecoder(stream) for { - select { - case <-watcher.WatchClose(): - return - default: - } - var buf logdriver.LogEntry if err := dec.Decode(&buf); err != nil { if err == io.EOF { return } - select { - case watcher.Err <- errors.Wrap(err, "error decoding log message"): - case <-watcher.WatchClose(): - } + watcher.Err <- errors.Wrap(err, "error decoding log message") return } @@ -125,11 +116,10 @@ func (a *pluginAdapterWithRead) ReadLogs(config ReadConfig) *LogWatcher { return } + // send the message unless the consumer is gone select { case watcher.Msg <- msg: - case <-watcher.WatchClose(): - // make sure the message we consumed is sent - watcher.Msg <- msg + case <-watcher.WatchConsumerGone(): return } } diff --git a/components/engine/daemon/logger/adapter_test.go b/components/engine/daemon/logger/adapter_test.go index f47e711c89..d14a48e477 100644 --- a/components/engine/daemon/logger/adapter_test.go +++ b/components/engine/daemon/logger/adapter_test.go @@ -174,7 +174,7 @@ func TestAdapterReadLogs(t *testing.T) { t.Fatal("timeout waiting for message channel to close") } - lw.Close() + lw.ProducerGone() lw = lr.ReadLogs(ReadConfig{Follow: true}) for _, x := range testMsg { diff --git a/components/engine/daemon/logger/journald/read.go b/components/engine/daemon/logger/journald/read.go index d4bcc62d9a..cadb97f4ca 100644 --- a/components/engine/daemon/logger/journald/read.go +++ b/components/engine/daemon/logger/journald/read.go @@ -165,7 +165,7 @@ func (s *journald) Close() error { s.mu.Lock() s.closed = true for reader := range s.readers.readers { - reader.Close() + reader.ProducerGone() } s.mu.Unlock() return nil @@ -299,7 +299,7 @@ func (s *journald) followJournal(logWatcher *logger.LogWatcher, j *C.sd_journal, // Wait until we're told to stop. select { case cursor = <-newCursor: - case <-logWatcher.WatchClose(): + case <-logWatcher.WatchConsumerGone(): // Notify the other goroutine that its work is done. C.close(pfd[1]) cursor = <-newCursor diff --git a/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go b/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go index 3649bdf91c..476e1d834b 100644 --- a/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go +++ b/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go @@ -166,13 +166,14 @@ func ValidateLogOpt(cfg map[string]string) error { return nil } -// Close closes underlying file and signals all readers to stop. +// 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.Close() + r.ProducerGone() delete(l.readers, r) } l.mu.Unlock() diff --git a/components/engine/daemon/logger/jsonfilelog/read_test.go b/components/engine/daemon/logger/jsonfilelog/read_test.go index 6ce4936e0e..cfa8694b19 100644 --- a/components/engine/daemon/logger/jsonfilelog/read_test.go +++ b/components/engine/daemon/logger/jsonfilelog/read_test.go @@ -50,11 +50,10 @@ func BenchmarkJSONFileLoggerReadLogs(b *testing.B) { }() lw := jsonlogger.(*JSONFileLogger).ReadLogs(logger.ReadConfig{Follow: true}) - watchClose := lw.WatchClose() for { select { case <-lw.Msg: - case <-watchClose: + case <-lw.WatchProducerGone(): return case err := <-chError: if err != nil { diff --git a/components/engine/daemon/logger/local/local.go b/components/engine/daemon/logger/local/local.go index 86c55784d4..ba4aa096f7 100644 --- a/components/engine/daemon/logger/local/local.go +++ b/components/engine/daemon/logger/local/local.go @@ -166,7 +166,7 @@ func (d *driver) Close() error { d.closed = true err := d.logfile.Close() for r := range d.readers { - r.Close() + r.ProducerGone() delete(d.readers, r) } d.mu.Unlock() diff --git a/components/engine/daemon/logger/logger.go b/components/engine/daemon/logger/logger.go index 912e855c7f..12e8d0054e 100644 --- a/components/engine/daemon/logger/logger.go +++ b/components/engine/daemon/logger/logger.go @@ -104,33 +104,50 @@ type LogWatcher struct { // For sending log messages to a reader. Msg chan *Message // For sending error messages that occur while while reading logs. - Err chan error - closeOnce sync.Once - closeNotifier chan struct{} + Err chan error + producerOnce sync.Once + producerGone chan struct{} + consumerOnce sync.Once + consumerGone chan struct{} } // NewLogWatcher returns a new LogWatcher. func NewLogWatcher() *LogWatcher { return &LogWatcher{ - Msg: make(chan *Message, logWatcherBufferSize), - Err: make(chan error, 1), - closeNotifier: make(chan struct{}), + Msg: make(chan *Message, logWatcherBufferSize), + Err: make(chan error, 1), + producerGone: make(chan struct{}), + consumerGone: make(chan struct{}), } } -// Close notifies the underlying log reader to stop. -func (w *LogWatcher) Close() { +// ProducerGone notifies the underlying log reader that +// the logs producer (a container) is gone. +func (w *LogWatcher) ProducerGone() { // only close if not already closed - w.closeOnce.Do(func() { - close(w.closeNotifier) + w.producerOnce.Do(func() { + close(w.producerGone) }) } -// WatchClose returns a channel receiver that receives notification -// when the watcher has been closed. This should only be called from -// one goroutine. -func (w *LogWatcher) WatchClose() <-chan struct{} { - return w.closeNotifier +// WatchProducerGone returns a channel receiver that receives notification +// once the logs producer (a container) is gone. +func (w *LogWatcher) WatchProducerGone() <-chan struct{} { + return w.producerGone +} + +// ConsumerGone notifies that the logs consumer is gone. +func (w *LogWatcher) ConsumerGone() { + // only close if not already closed + w.consumerOnce.Do(func() { + close(w.consumerGone) + }) +} + +// WatchConsumerGone returns a channel receiver that receives notification +// when the log watcher consumer is gone. +func (w *LogWatcher) WatchConsumerGone() <-chan struct{} { + return w.consumerGone } // Capability defines the list of capabilities that a driver can implement diff --git a/components/engine/daemon/logger/loggerutils/logfile.go b/components/engine/daemon/logger/loggerutils/logfile.go index 25be44aa49..623f78f5e7 100644 --- a/components/engine/daemon/logger/loggerutils/logfile.go +++ b/components/engine/daemon/logger/loggerutils/logfile.go @@ -488,7 +488,7 @@ func tailFiles(files []SizeReaderAt, watcher *logger.LogWatcher, createDecoder m go func() { select { case <-ctx.Done(): - case <-watcher.WatchClose(): + case <-watcher.WatchConsumerGone(): cancel() } }() @@ -546,22 +546,9 @@ 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() @@ -596,7 +583,9 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int case fsnotify.Rename, fsnotify.Remove: select { case <-notifyRotate: - case <-ctx.Done(): + case <-logWatcher.WatchProducerGone(): + return errDone + case <-logWatcher.WatchConsumerGone(): return errDone } if err := handleRotate(); err != nil { @@ -618,7 +607,9 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int return errRetry } return err - case <-ctx.Done(): + case <-logWatcher.WatchProducerGone(): + return errDone + case <-logWatcher.WatchConsumerGone(): return errDone } } @@ -664,23 +655,11 @@ func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan int if !until.IsZero() && msg.Timestamp.After(until) { return } + // send the message, unless the consumer is gone select { case logWatcher.Msg <- msg: - case <-ctx.Done(): - logWatcher.Msg <- msg - for { - msg, err := decodeLogLine() - if err != nil { - return - } - if !since.IsZero() && msg.Timestamp.Before(since) { - continue - } - if !until.IsZero() && msg.Timestamp.After(until) { - return - } - logWatcher.Msg <- msg - } + case <-logWatcher.WatchConsumerGone(): + return } } } diff --git a/components/engine/daemon/logger/loggerutils/logfile_test.go b/components/engine/daemon/logger/loggerutils/logfile_test.go index c21aa258e6..037592a3f2 100644 --- a/components/engine/daemon/logger/loggerutils/logfile_test.go +++ b/components/engine/daemon/logger/loggerutils/logfile_test.go @@ -77,7 +77,7 @@ func TestTailFiles(t *testing.T) { } } -func TestFollowLogsClose(t *testing.T) { +func TestFollowLogsConsumerGone(t *testing.T) { lw := logger.NewLogWatcher() f, err := ioutil.TempFile("", t.Name()) @@ -110,7 +110,7 @@ func TestFollowLogsClose(t *testing.T) { t.Fatal("timeout waiting for log message") } - lw.Close() + lw.ConsumerGone() select { case <-followLogsDone: case <-time.After(20 * time.Second): diff --git a/components/engine/daemon/logs.go b/components/engine/daemon/logs.go index 5e5c34992e..668a75c778 100644 --- a/components/engine/daemon/logs.go +++ b/components/engine/daemon/logs.go @@ -110,8 +110,8 @@ func (daemon *Daemon) ContainerLogs(ctx context.Context, containerName string, c } }() } - // set up some defers - defer logs.Close() + // signal that the log reader is gone + defer logs.ConsumerGone() // close the messages channel. closing is the only way to signal above // that we're doing with logs (other than context cancel i guess). From a620951919ca95f0f5a67b49b7d8d284a5aadc45 Mon Sep 17 00:00:00 2001 From: Kir Kolyshkin Date: Tue, 28 Aug 2018 19:58:47 -0700 Subject: [PATCH 7/7] TestFollowLogsProducerGone: add This should test that - all the messages produced are delivered (i.e. not lost) - followLogs() exits Loosely based on the test having the same name by Brian Goff, see https://gist.github.com/cpuguy83/e538793de18c762608358ee0eaddc197 Signed-off-by: Kir Kolyshkin (cherry picked from commit f845d76d047760c91dc0c7076aea840291fbdbc5) Upstream-commit: 2a82480df9ad91593d59be4b5283917dbea2da39 Component: engine --- .../daemon/logger/loggerutils/logfile_test.go | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/components/engine/daemon/logger/loggerutils/logfile_test.go b/components/engine/daemon/logger/loggerutils/logfile_test.go index 037592a3f2..e3e63210fc 100644 --- a/components/engine/daemon/logger/loggerutils/logfile_test.go +++ b/components/engine/daemon/logger/loggerutils/logfile_test.go @@ -117,3 +117,87 @@ func TestFollowLogsConsumerGone(t *testing.T) { t.Fatal("timeout waiting for followLogs() to finish") } } + +func TestFollowLogsProducerGone(t *testing.T) { + lw := logger.NewLogWatcher() + + f, err := ioutil.TempFile("", t.Name()) + assert.NilError(t, err) + defer os.Remove(f.Name()) + + var sent, received, closed int + makeDecoder := func(rdr io.Reader) func() (*logger.Message, error) { + return func() (*logger.Message, error) { + if closed == 1 { + closed++ + t.Logf("logDecode() closed after sending %d messages\n", sent) + return nil, io.EOF + } else if closed > 1 { + t.Fatal("logDecode() called after closing!") + return nil, io.EOF + } + sent++ + return &logger.Message{}, nil + } + } + var since, until time.Time + + followLogsDone := make(chan struct{}) + go func() { + followLogs(f, lw, make(chan interface{}), makeDecoder, since, until) + close(followLogsDone) + }() + + // read 1 message + select { + case <-lw.Msg: + received++ + case err := <-lw.Err: + assert.NilError(t, err) + case <-followLogsDone: + t.Fatal("followLogs() finished unexpectedly") + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for log message") + } + + // "stop" the "container" + closed = 1 + lw.ProducerGone() + + // should receive all the messages sent + readDone := make(chan struct{}) + go func() { + defer close(readDone) + for { + select { + case <-lw.Msg: + received++ + if received == sent { + return + } + case err := <-lw.Err: + assert.NilError(t, err) + } + } + }() + select { + case <-readDone: + case <-time.After(30 * time.Second): + t.Fatalf("timeout waiting for log messages to be read (sent: %d, received: %d", sent, received) + } + + t.Logf("messages sent: %d, received: %d", sent, received) + + // followLogs() should be done by now + select { + case <-followLogsDone: + case <-time.After(30 * time.Second): + t.Fatal("timeout waiting for followLogs() to finish") + } + + select { + case <-lw.WatchConsumerGone(): + t.Fatal("consumer should not have exited") + default: + } +}