Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Akihiro Suda <suda.akihiro@lab.ntt.co.jp>
- Loading branch information
1 parent
287b0a6
commit 55ec631
Showing
17 changed files
with
372 additions
and
83 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -402,6 +402,7 @@ __docker_complete_log_drivers() { | |
| journald | ||
| json-file | ||
| none | ||
| rawfifo | ||
| splunk | ||
| syslog | ||
| " -- "$cur" ) ) | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,83 +1,23 @@ | ||
| package logger | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "fmt" | ||
| "io" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/Sirupsen/logrus" | ||
| ) | ||
|
|
||
| // Copier can copy logs from specified sources to Logger and attach | ||
| // ContainerID and Timestamp. | ||
| // Writes are concurrent, so you need implement some sync in your logger | ||
| type Copier struct { | ||
| // srcs is map of name -> reader pairs, for example "stdout", "stderr" | ||
| srcs map[string]io.Reader | ||
| dst Logger | ||
| copyJobs sync.WaitGroup | ||
| closed chan struct{} | ||
| // Copier can copy logs from specified sources to Logger | ||
| type Copier interface { | ||
| Run() | ||
| Wait() | ||
| Close() | ||
| } | ||
|
|
||
| // NewCopier creates a new Copier | ||
| func NewCopier(srcs map[string]io.Reader, dst Logger) *Copier { | ||
| return &Copier{ | ||
| srcs: srcs, | ||
| dst: dst, | ||
| closed: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| // Run starts logs copying | ||
| func (c *Copier) Run() { | ||
| for src, w := range c.srcs { | ||
| c.copyJobs.Add(1) | ||
| go c.copySrc(src, w) | ||
| } | ||
| } | ||
|
|
||
| func (c *Copier) copySrc(name string, src io.Reader) { | ||
| defer c.copyJobs.Done() | ||
| reader := bufio.NewReader(src) | ||
|
|
||
| for { | ||
| select { | ||
| case <-c.closed: | ||
| return | ||
| default: | ||
| line, err := reader.ReadBytes('\n') | ||
| line = bytes.TrimSuffix(line, []byte{'\n'}) | ||
|
|
||
| // ReadBytes can return full or partial output even when it failed. | ||
| // e.g. it can return a full entry and EOF. | ||
| if err == nil || len(line) > 0 { | ||
| if logErr := c.dst.Log(&Message{Line: line, Source: name, Timestamp: time.Now().UTC()}); logErr != nil { | ||
| logrus.Errorf("Failed to log msg %q for logger %s: %s", line, c.dst.Name(), logErr) | ||
| } | ||
| } | ||
|
|
||
| if err != nil { | ||
| if err != io.EOF { | ||
| logrus.Errorf("Error scanning log stream: %s", err) | ||
| } | ||
| return | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Wait waits until all copying is done | ||
| func (c *Copier) Wait() { | ||
| c.copyJobs.Wait() | ||
| } | ||
|
|
||
| // Close closes the copier | ||
| func (c *Copier) Close() { | ||
| select { | ||
| case <-c.closed: | ||
| default: | ||
| close(c.closed) | ||
| func NewCopier(srcs map[string]io.Reader, dst Logger) (Copier, error) { | ||
| if ml, ok := dst.(MessageLogger); ok { | ||
| return NewMessageCopier(srcs, ml), nil | ||
| } else if rl, ok := dst.(RawLogger); ok { | ||
| return NewRawCopier(srcs, rl), nil | ||
| } | ||
| return nil, fmt.Errorf("strange logger %s", dst.Name()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package logger | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "io" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/Sirupsen/logrus" | ||
| ) | ||
|
|
||
| // MessageCopier can copy log messagess from specified sources to Logger and attach Timestamp. | ||
| // Writes are concurrent, so you need implement some sync in your logger | ||
| type MessageCopier struct { | ||
| // srcs is map of name -> reader pairs, for example "stdout", "stderr" | ||
| srcs map[string]io.Reader | ||
| dst MessageLogger | ||
| copyJobs sync.WaitGroup | ||
| closed chan struct{} | ||
| } | ||
|
|
||
| // NewMessageCopier creates a new MessageCopier | ||
| func NewMessageCopier(srcs map[string]io.Reader, dst MessageLogger) *MessageCopier { | ||
| return &MessageCopier{ | ||
| srcs: srcs, | ||
| dst: dst, | ||
| closed: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| // Run starts logs copying | ||
| func (c *MessageCopier) Run() { | ||
| for src, w := range c.srcs { | ||
| c.copyJobs.Add(1) | ||
| go c.copySrc(src, w) | ||
| } | ||
| } | ||
|
|
||
| func (c *MessageCopier) copySrc(name string, src io.Reader) { | ||
| defer c.copyJobs.Done() | ||
| reader := bufio.NewReader(src) | ||
|
|
||
| for { | ||
| select { | ||
| case <-c.closed: | ||
| return | ||
| default: | ||
| line, err := reader.ReadBytes('\n') | ||
| line = bytes.TrimSuffix(line, []byte{'\n'}) | ||
|
|
||
| // ReadBytes can return full or partial output even when it failed. | ||
| // e.g. it can return a full entry and EOF. | ||
| if err == nil || len(line) > 0 { | ||
| if logErr := c.dst.Log(&Message{Line: line, Source: name, Timestamp: time.Now().UTC()}); logErr != nil { | ||
| logrus.Errorf("Failed to log msg %q for logger %s: %s", line, c.dst.Name(), logErr) | ||
| } | ||
| } | ||
|
|
||
| if err != nil { | ||
| if err != io.EOF { | ||
| logrus.Errorf("Error scanning log stream: %s", err) | ||
| } | ||
| return | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Wait waits until all copying is done | ||
| func (c *MessageCopier) Wait() { | ||
| c.copyJobs.Wait() | ||
| } | ||
|
|
||
| // Close closes the copier | ||
| func (c *MessageCopier) Close() { | ||
| select { | ||
| case <-c.closed: | ||
| default: | ||
| close(c.closed) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package logger | ||
|
|
||
| import ( | ||
| "io" | ||
| "sync" | ||
|
|
||
| "github.com/Sirupsen/logrus" | ||
| ) | ||
|
|
||
| // RawCopier can copy logs from specified sources | ||
| type RawCopier struct { | ||
| // srcs is map of name -> reader pairs, for example "stdout", "stderr" | ||
| srcs map[string]io.Reader | ||
| dst RawLogger | ||
| copyJobs sync.WaitGroup | ||
| closed chan struct{} | ||
| } | ||
|
|
||
| // NewRawCopier creates a new RawCopier | ||
| func NewRawCopier(srcs map[string]io.Reader, dst RawLogger) *RawCopier { | ||
| return &RawCopier{ | ||
| srcs: srcs, | ||
| dst: dst, | ||
| closed: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| // Run starts logs copying | ||
| func (c *RawCopier) Run() { | ||
| for src, w := range c.srcs { | ||
| c.copyJobs.Add(1) | ||
| go c.copySrc(src, w) | ||
| } | ||
| } | ||
|
|
||
| func (c *RawCopier) copySrc(name string, src io.Reader) { | ||
| defer c.copyJobs.Done() | ||
| w, err := c.dst.RawWriter(name) | ||
| if err != nil { | ||
| logrus.Errorf("error while opening RawWriter for %s: %v", name, err) | ||
| return | ||
| } | ||
| for { | ||
| select { | ||
| case <-c.closed: | ||
| if err = w.Close(); err != nil { | ||
| logrus.Errorf("error while closing RawWriter for %s: %v", name, err) | ||
| } | ||
| return | ||
| default: | ||
| // use io.CopyN rather than io.Copy, so that we can catch <-c.closed | ||
| bufsz := int64(64 * 1024) | ||
| if _, err := io.CopyN(w, src, bufsz); err != nil { | ||
| logrus.Errorf("stream copy error: %s: %v", name, err) | ||
| return | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Wait waits until all copying is done | ||
| func (c *RawCopier) Wait() { | ||
| c.copyJobs.Wait() | ||
| } | ||
|
|
||
| // Close closes the copier | ||
| func (c *RawCopier) Close() { | ||
| select { | ||
| case <-c.closed: | ||
| default: | ||
| close(c.closed) | ||
| } | ||
| } |
Oops, something went wrong.