Browse code
only close LogDriver after LogCopier is done
this prevents the copier from sending messages in the buffer to the closed
driver. If the copied took longer than the timeout to drain the buffer, this
aborts the copier read loop and return back so we can cleanup resources
properly.
Signed-off-by: Daniel Dao <dqminh@cloudflare.com>
Daniel Dao authored on 2016/01/15 22:42:23Showing 3 changed files
| ... | ... |
@@ -369,6 +369,9 @@ func (m *containerMonitor) resetContainer(lock bool) {
|
| 369 | 369 |
select {
|
| 370 | 370 |
case <-time.After(loggerCloseTimeout): |
| 371 | 371 |
logrus.Warnf("Logger didn't exit in time: logs may be truncated")
|
| 372 |
+ container.LogCopier.Close() |
|
| 373 |
+ // always waits for the LogCopier to finished before closing |
|
| 374 |
+ <-exit |
|
| 372 | 375 |
case <-exit: |
| 373 | 376 |
} |
| 374 | 377 |
} |
| ... | ... |
@@ -20,14 +20,16 @@ type Copier struct {
|
| 20 | 20 |
srcs map[string]io.Reader |
| 21 | 21 |
dst Logger |
| 22 | 22 |
copyJobs sync.WaitGroup |
| 23 |
+ closed chan struct{}
|
|
| 23 | 24 |
} |
| 24 | 25 |
|
| 25 | 26 |
// NewCopier creates a new Copier |
| 26 | 27 |
func NewCopier(cid string, srcs map[string]io.Reader, dst Logger) *Copier {
|
| 27 | 28 |
return &Copier{
|
| 28 |
- cid: cid, |
|
| 29 |
- srcs: srcs, |
|
| 30 |
- dst: dst, |
|
| 29 |
+ cid: cid, |
|
| 30 |
+ srcs: srcs, |
|
| 31 |
+ dst: dst, |
|
| 32 |
+ closed: make(chan struct{}),
|
|
| 31 | 33 |
} |
| 32 | 34 |
} |
| 33 | 35 |
|
| ... | ... |
@@ -44,24 +46,28 @@ func (c *Copier) copySrc(name string, src io.Reader) {
|
| 44 | 44 |
reader := bufio.NewReader(src) |
| 45 | 45 |
|
| 46 | 46 |
for {
|
| 47 |
- line, err := reader.ReadBytes('\n')
|
|
| 48 |
- line = bytes.TrimSuffix(line, []byte{'\n'})
|
|
| 47 |
+ select {
|
|
| 48 |
+ case <-c.closed: |
|
| 49 |
+ return |
|
| 50 |
+ default: |
|
| 51 |
+ line, err := reader.ReadBytes('\n')
|
|
| 52 |
+ line = bytes.TrimSuffix(line, []byte{'\n'})
|
|
| 49 | 53 |
|
| 50 |
- // ReadBytes can return full or partial output even when it failed. |
|
| 51 |
- // e.g. it can return a full entry and EOF. |
|
| 52 |
- if err == nil || len(line) > 0 {
|
|
| 53 |
- if logErr := c.dst.Log(&Message{ContainerID: c.cid, Line: line, Source: name, Timestamp: time.Now().UTC()}); logErr != nil {
|
|
| 54 |
- logrus.Errorf("Failed to log msg %q for logger %s: %s", line, c.dst.Name(), logErr)
|
|
| 54 |
+ // ReadBytes can return full or partial output even when it failed. |
|
| 55 |
+ // e.g. it can return a full entry and EOF. |
|
| 56 |
+ if err == nil || len(line) > 0 {
|
|
| 57 |
+ if logErr := c.dst.Log(&Message{ContainerID: c.cid, Line: line, Source: name, Timestamp: time.Now().UTC()}); logErr != nil {
|
|
| 58 |
+ logrus.Errorf("Failed to log msg %q for logger %s: %s", line, c.dst.Name(), logErr)
|
|
| 59 |
+ } |
|
| 55 | 60 |
} |
| 56 |
- } |
|
| 57 | 61 |
|
| 58 |
- if err != nil {
|
|
| 59 |
- if err != io.EOF {
|
|
| 60 |
- logrus.Errorf("Error scanning log stream: %s", err)
|
|
| 62 |
+ if err != nil {
|
|
| 63 |
+ if err != io.EOF {
|
|
| 64 |
+ logrus.Errorf("Error scanning log stream: %s", err)
|
|
| 65 |
+ } |
|
| 66 |
+ return |
|
| 61 | 67 |
} |
| 62 |
- return |
|
| 63 | 68 |
} |
| 64 |
- |
|
| 65 | 69 |
} |
| 66 | 70 |
} |
| 67 | 71 |
|
| ... | ... |
@@ -69,3 +75,12 @@ func (c *Copier) copySrc(name string, src io.Reader) {
|
| 69 | 69 |
func (c *Copier) Wait() {
|
| 70 | 70 |
c.copyJobs.Wait() |
| 71 | 71 |
} |
| 72 |
+ |
|
| 73 |
+// Close closes the copier |
|
| 74 |
+func (c *Copier) Close() {
|
|
| 75 |
+ select {
|
|
| 76 |
+ case <-c.closed: |
|
| 77 |
+ default: |
|
| 78 |
+ close(c.closed) |
|
| 79 |
+ } |
|
| 80 |
+} |
| ... | ... |
@@ -10,9 +10,15 @@ import ( |
| 10 | 10 |
|
| 11 | 11 |
type TestLoggerJSON struct {
|
| 12 | 12 |
*json.Encoder |
| 13 |
+ delay time.Duration |
|
| 13 | 14 |
} |
| 14 | 15 |
|
| 15 |
-func (l *TestLoggerJSON) Log(m *Message) error { return l.Encode(m) }
|
|
| 16 |
+func (l *TestLoggerJSON) Log(m *Message) error {
|
|
| 17 |
+ if l.delay > 0 {
|
|
| 18 |
+ time.Sleep(l.delay) |
|
| 19 |
+ } |
|
| 20 |
+ return l.Encode(m) |
|
| 21 |
+} |
|
| 16 | 22 |
|
| 17 | 23 |
func (l *TestLoggerJSON) Close() error { return nil }
|
| 18 | 24 |
|
| ... | ... |
@@ -94,3 +100,33 @@ func TestCopier(t *testing.T) {
|
| 94 | 94 |
} |
| 95 | 95 |
} |
| 96 | 96 |
} |
| 97 |
+ |
|
| 98 |
+func TestCopierSlow(t *testing.T) {
|
|
| 99 |
+ stdoutLine := "Line that thinks that it is log line from docker stdout" |
|
| 100 |
+ var stdout bytes.Buffer |
|
| 101 |
+ for i := 0; i < 30; i++ {
|
|
| 102 |
+ if _, err := stdout.WriteString(stdoutLine + "\n"); err != nil {
|
|
| 103 |
+ t.Fatal(err) |
|
| 104 |
+ } |
|
| 105 |
+ } |
|
| 106 |
+ |
|
| 107 |
+ var jsonBuf bytes.Buffer |
|
| 108 |
+ //encoder := &encodeCloser{Encoder: json.NewEncoder(&jsonBuf)}
|
|
| 109 |
+ jsonLog := &TestLoggerJSON{Encoder: json.NewEncoder(&jsonBuf), delay: 100 * time.Millisecond}
|
|
| 110 |
+ |
|
| 111 |
+ cid := "a7317399f3f857173c6179d44823594f8294678dea9999662e5c625b5a1c7657" |
|
| 112 |
+ c := NewCopier(cid, map[string]io.Reader{"stdout": &stdout}, jsonLog)
|
|
| 113 |
+ c.Run() |
|
| 114 |
+ wait := make(chan struct{})
|
|
| 115 |
+ go func() {
|
|
| 116 |
+ c.Wait() |
|
| 117 |
+ close(wait) |
|
| 118 |
+ }() |
|
| 119 |
+ <-time.After(150 * time.Millisecond) |
|
| 120 |
+ c.Close() |
|
| 121 |
+ select {
|
|
| 122 |
+ case <-time.After(200 * time.Millisecond): |
|
| 123 |
+ t.Fatalf("failed to exit in time after the copier is closed")
|
|
| 124 |
+ case <-wait: |
|
| 125 |
+ } |
|
| 126 |
+} |