Use sync.Pool for logger Messages

This reduces allocs and bytes used per log entry significantly as well
as some improvement to time per log operation.

Each log driver, however, must put messages back in the pool once they
are finished with the message.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

Author: cpuguy83

daemon/logger/awslogs/cloudwatchlogs.go

@@ -175,7 +175,6 @@ func (l *logStream) Log(msg *logger.Message) error {
 	l.lock.RLock()
 	defer l.lock.RUnlock()
 	if !l.closed {
-		// buffer up the data, making sure to copy the Line data
 		l.messages <- msg
 	}
 	return nil
@@ -277,6 +276,7 @@ func (l *logStream) collectBatch() {
 				})
 				bytes += (lineBytes + perEventBytes)
 			}
+			logger.PutMessage(msg)
 		}
 	}
 }

daemon/logger/copier.go

@@ -76,15 +76,14 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 			}
 			// Break up the data that we've buffered up into lines, and log each in turn.
 			p := 0
-			for q := bytes.Index(buf[p:n], []byte{'\n'}); q >= 0; q = bytes.Index(buf[p:n], []byte{'\n'}) {
+			for q := bytes.IndexByte(buf[p:n], '\n'); q >= 0; q = bytes.IndexByte(buf[p:n], '\n') {
 				select {
 				case <-c.closed:
 					return
 				default:
-					msg := &Message{
-						Source:    name,
-						Timestamp: time.Now().UTC(),
-					}
+					msg := NewMessage()
+					msg.Source = name
+					msg.Timestamp = time.Now().UTC()
 					msg.Line = append(msg.Line, buf[p:p+q]...)
 
 					if logErr := c.dst.Log(msg); logErr != nil {
@@ -98,11 +97,9 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 			// noting that it's a partial log line.
 			if eof || (p == 0 && n == len(buf)) {
 				if p < n {
-					msg := &Message{
-						Source:    name,
-						Timestamp: time.Now().UTC(),
-						Partial:   true,
-					}
+					msg := NewMessage()
+					msg.Source = name
+					msg.Timestamp = time.Now().UTC()
 					msg.Line = append(msg.Line, buf[p:n]...)
 					msg.Partial = true
 

daemon/logger/copier_test.go

@@ -208,7 +208,7 @@ func TestCopierSlow(t *testing.T) {
 type BenchmarkLoggerDummy struct {
 }
 
-func (l *BenchmarkLoggerDummy) Log(m *Message) error { return nil }
+func (l *BenchmarkLoggerDummy) Log(m *Message) error { PutMessage(m); return nil }
 
 func (l *BenchmarkLoggerDummy) Close() error { return nil }
 

daemon/logger/etwlogs/etwlogs_windows.go

@@ -76,7 +76,9 @@ func (etwLogger *etwLogs) Log(msg *logger.Message) error {
 		logrus.Error(errorMessage)
 		return errors.New(errorMessage)
 	}
-	return callEventWriteString(createLogMessage(etwLogger, msg))
+	m := createLogMessage(etwLogger, msg)
+	logger.PutMessage(msg)
+	return callEventWriteString(m)
 }
 
 // Close closes the logger by unregistering the ETW provider.

daemon/logger/fluentd/fluentd.go

@@ -151,9 +151,12 @@ func (f *fluentd) Log(msg *logger.Message) error {
 	for k, v := range f.extra {
 		data[k] = v
 	}
+
+	ts := msg.Timestamp
+	logger.PutMessage(msg)
 	// fluent-logger-golang buffers logs from failures and disconnections,
 	// and these are transferred again automatically.
-	return f.writer.PostWithTime(f.tag, msg.Timestamp, data)
+	return f.writer.PostWithTime(f.tag, ts, data)
 }
 
 func (f *fluentd) Close() error {

daemon/logger/gcplogs/gcplogging.go

@@ -185,12 +185,16 @@ func ValidateLogOpts(cfg map[string]string) error {
 }
 
 func (l *gcplogs) Log(m *logger.Message) error {
+	data := string(m.Line)
+	ts := m.Timestamp
+	logger.PutMessage(m)
+
 	l.logger.Log(logging.Entry{
-		Timestamp: m.Timestamp,
+		Timestamp: ts,
 		Payload: &dockerLogEntry{
 			Instance:  l.instance,
 			Container: l.container,
-			Data:      string(m.Line),
+			Data:      data,
 		},
 	})
 	return nil

daemon/logger/gelf/gelf.go

@@ -133,6 +133,7 @@ func (s *gelfLogger) Log(msg *logger.Message) error {
 		Level:    level,
 		RawExtra: s.rawExtra,
 	}
+	logger.PutMessage(msg)
 
 	if err := s.writer.WriteMessage(&m); err != nil {
 		return fmt.Errorf("gelf: cannot send GELF message: %v", err)

daemon/logger/journald/journald.go

@@ -105,10 +105,14 @@ func (s *journald) Log(msg *logger.Message) error {
 	if msg.Partial {
 		vars["CONTAINER_PARTIAL_MESSAGE"] = "true"
 	}
+
+	line := string(msg.Line)
+	logger.PutMessage(msg)
+
 	if msg.Source == "stderr" {
-		return journal.Send(string(msg.Line), journal.PriErr, vars)
+		return journal.Send(line, journal.PriErr, vars)
 	}
-	return journal.Send(string(msg.Line), journal.PriInfo, vars)
+	return journal.Send(line, journal.PriInfo, vars)
 }
 
 func (s *journald) Name() string {

daemon/logger/jsonfilelog/jsonfilelog.go

@@ -100,6 +100,7 @@ func (l *JSONFileLogger) Log(msg *logger.Message) error {
 		Created:  timestamp,
 		RawAttrs: l.extra,
 	}).MarshalJSONBuf(l.buf)
+	logger.PutMessage(msg)
 	if err != nil {
 		l.mu.Unlock()
 		return err

daemon/logger/logentries/logentries.go

@@ -61,7 +61,9 @@ func (f *logentries) Log(msg *logger.Message) error {
 	for k, v := range f.extra {
 		data[k] = v
 	}
-	f.writer.Println(f.tag, msg.Timestamp, data)
+	ts := msg.Timestamp
+	logger.PutMessage(msg)
+	f.writer.Println(f.tag, ts, data)
 	return nil
 }