copier.go 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. package logger
  2. import (
  3. "bufio"
  4. "io"
  5. "time"
  6. "github.com/Sirupsen/logrus"
  7. )
  8. // Copier can copy logs from specified sources to Logger and attach
  9. // ContainerID and Timestamp.
  10. // Writes are concurrent, so you need implement some sync in your logger
  11. type Copier struct {
  12. // cid is container id for which we copying logs
  13. cid string
  14. // srcs is map of name -> reader pairs, for example "stdout", "stderr"
  15. srcs map[string]io.Reader
  16. dst Logger
  17. }
  18. // NewCopier creates new Copier
  19. func NewCopier(cid string, srcs map[string]io.Reader, dst Logger) (*Copier, error) {
  20. return &Copier{
  21. cid: cid,
  22. srcs: srcs,
  23. dst: dst,
  24. }, nil
  25. }
  26. // Run starts logs copying
  27. func (c *Copier) Run() {
  28. for src, w := range c.srcs {
  29. go c.copySrc(src, w)
  30. }
  31. }
  32. func (c *Copier) copySrc(name string, src io.Reader) {
  33. scanner := bufio.NewScanner(src)
  34. for scanner.Scan() {
  35. if err := c.dst.Log(&Message{ContainerID: c.cid, Line: scanner.Bytes(), Source: name, Timestamp: time.Now().UTC()}); err != nil {
  36. logrus.Errorf("Failed to log msg %q for logger %s: %s", scanner.Bytes(), c.dst.Name(), err)
  37. }
  38. }
  39. if err := scanner.Err(); err != nil {
  40. logrus.Errorf("Error scanning log stream: %s", err)
  41. }
  42. }