Merge pull request #19831 from cloudflare/optimize-gelf

GELF logger: Add gelf-compression-type and gelf-compression-level
Upstream-commit: 0f59b0b12c37b5a3b4ad1c8a0cb91c93ff4228a9
Component: engine

components/engine/vendor/src/github.com/Graylog2/go-gelf/gelf/reader.go

@@ -66,7 +66,6 @@ func (r *Reader) ReadMessage() (*Message, error) {
 	var (
 		err        error
 		n, length  int
-		buf        bytes.Buffer
 		cid, ocid  []byte
 		seq, total uint8
 		cHead      []byte
@@ -122,19 +121,18 @@ func (r *Reader) ReadMessage() (*Message, error) {
 		// zlib is slightly more complicated, but correct
 		cReader, err = zlib.NewReader(bytes.NewReader(cBuf))
 	} else {
-		return nil, fmt.Errorf("unknown magic: %x %v", cHead, cHead)
+		// compliance with https://github.com/Graylog2/graylog2-server
+		// treating all messages as uncompressed if  they are not gzip, zlib or
+		// chunked
+		cReader = bytes.NewReader(cBuf)
 	}
 
 	if err != nil {
 		return nil, fmt.Errorf("NewReader: %s", err)
 	}
 
-	if _, err = io.Copy(&buf, cReader); err != nil {
-		return nil, fmt.Errorf("io.Copy: %s", err)
-	}
-
 	msg := new(Message)
-	if err := json.Unmarshal(buf.Bytes(), &msg); err != nil {
+	if err := json.NewDecoder(cReader).Decode(&msg); err != nil {
 		return nil, fmt.Errorf("json.Unmarshal: %s", err)
 	}
 

components/engine/vendor/src/github.com/Graylog2/go-gelf/gelf/writer.go

@@ -41,6 +41,7 @@ type CompressType int
 const (
 	CompressGzip CompressType = iota
 	CompressZlib
+	CompressNone
 )
 
 // Message represents the contents of the GELF message.  It is gzipped
@@ -49,15 +50,14 @@ type Message struct {
 	Version  string                 `json:"version"`
 	Host     string                 `json:"host"`
 	Short    string                 `json:"short_message"`
-	Full     string                 `json:"full_message"`
+	Full     string                 `json:"full_message,omitempty"`
 	TimeUnix float64                `json:"timestamp"`
-	Level    int32                  `json:"level"`
-	Facility string                 `json:"facility"`
+	Level    int32                  `json:"level,omitempty"`
+	Facility string                 `json:"facility,omitempty"`
 	Extra    map[string]interface{} `json:"-"`
+	RawExtra json.RawMessage        `json:"-"`
 }
 
-type innerMessage Message //against circular (Un)MarshalJSON
-
 // Used to control GELF chunking.  Should be less than (MTU - len(UDP
 // header)).
 //
@@ -76,14 +76,14 @@ var (
 
 // Syslog severity levels
 const (
-  LOG_EMERG   = int32(0)
-  LOG_ALERT   = int32(1)
-  LOG_CRIT    = int32(2)
-  LOG_ERR     = int32(3)
-  LOG_WARNING = int32(4)
-  LOG_NOTICE  = int32(5)
-  LOG_INFO    = int32(6)
-  LOG_DEBUG   = int32(7)
+	LOG_EMERG   = int32(0)
+	LOG_ALERT   = int32(1)
+	LOG_CRIT    = int32(2)
+	LOG_ERR     = int32(3)
+	LOG_WARNING = int32(4)
+	LOG_NOTICE  = int32(5)
+	LOG_INFO    = int32(6)
+	LOG_DEBUG   = int32(7)
 )
 
 // numChunks returns the number of GELF chunks necessary to transmit
@@ -176,40 +176,70 @@ func (w *Writer) writeChunked(zBytes []byte) (err error) {
 	return nil
 }
 
+// 1k bytes buffer by default
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		return bytes.NewBuffer(make([]byte, 0, 1024))
+	},
+}
+
+func newBuffer() *bytes.Buffer {
+	b := bufPool.Get().(*bytes.Buffer)
+	if b != nil {
+		b.Reset()
+		return b
+	}
+	return bytes.NewBuffer(nil)
+}
+
 // WriteMessage sends the specified message to the GELF server
 // specified in the call to New().  It assumes all the fields are
 // filled out appropriately.  In general, clients will want to use
 // Write, rather than WriteMessage.
 func (w *Writer) WriteMessage(m *Message) (err error) {
-	mBytes, err := json.Marshal(m)
-	if err != nil {
-		return
+	mBuf := newBuffer()
+	defer bufPool.Put(mBuf)
+	if err = m.MarshalJSONBuf(mBuf); err != nil {
+		return err
 	}
+	mBytes := mBuf.Bytes()
+
+	var (
+		zBuf   *bytes.Buffer
+		zBytes []byte
+	)
 
-	var zBuf bytes.Buffer
 	var zw io.WriteCloser
 	switch w.CompressionType {
 	case CompressGzip:
-		zw, err = gzip.NewWriterLevel(&zBuf, w.CompressionLevel)
+		zBuf = newBuffer()
+		defer bufPool.Put(zBuf)
+		zw, err = gzip.NewWriterLevel(zBuf, w.CompressionLevel)
 	case CompressZlib:
-		zw, err = zlib.NewWriterLevel(&zBuf, w.CompressionLevel)
+		zBuf = newBuffer()
+		defer bufPool.Put(zBuf)
+		zw, err = zlib.NewWriterLevel(zBuf, w.CompressionLevel)
+	case CompressNone:
+		zBytes = mBytes
 	default:
 		panic(fmt.Sprintf("unknown compression type %d",
 			w.CompressionType))
 	}
-	if err != nil {
-		return
+	if zw != nil {
+		if err != nil {
+			return
+		}
+		if _, err = zw.Write(mBytes); err != nil {
+			zw.Close()
+			return
+		}
+		zw.Close()
+		zBytes = zBuf.Bytes()
 	}
-	if _, err = zw.Write(mBytes); err != nil {
-		return
-	}
-	zw.Close()
 
-	zBytes := zBuf.Bytes()
 	if numChunks(zBytes) > 1 {
 		return w.writeChunked(zBytes)
 	}
-
 	n, err := w.conn.Write(zBytes)
 	if err != nil {
 		return
@@ -222,8 +252,8 @@ func (w *Writer) WriteMessage(m *Message) (err error) {
 }
 
 // Close connection and interrupt blocked Read or Write operations
-func (w *Writer) Close() (error) {
-  return w.conn.Close()
+func (w *Writer) Close() error {
+	return w.conn.Close()
 }
 
 /*
@@ -315,28 +345,43 @@ func (w *Writer) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
-func (m *Message) MarshalJSON() ([]byte, error) {
-	var err error
-	var b, eb []byte
-
-	extra := m.Extra
-	b, err = json.Marshal((*innerMessage)(m))
-	m.Extra = extra
+func (m *Message) MarshalJSONBuf(buf *bytes.Buffer) error {
+	b, err := json.Marshal(m)
 	if err != nil {
-		return nil, err
+		return err
+	}
+	// write up until the final }
+	if _, err = buf.Write(b[:len(b)-1]); err != nil {
+		return err
+	}
+	if len(m.Extra) > 0 {
+		eb, err := json.Marshal(m.Extra)
+		if err != nil {
+			return err
+		}
+		// merge serialized message + serialized extra map
+		if err = buf.WriteByte(','); err != nil {
+			return err
+		}
+		// write serialized extra bytes, without enclosing quotes
+		if _, err = buf.Write(eb[1 : len(eb)-1]); err != nil {
+			return err
+		}
 	}
 
-	if len(extra) == 0 {
-		return b, nil
+	if len(m.RawExtra) > 0 {
+		if err := buf.WriteByte(','); err != nil {
+			return err
+		}
+
+		// write serialized extra bytes, without enclosing quotes
+		if _, err = buf.Write(m.RawExtra[1 : len(m.RawExtra)-1]); err != nil {
+			return err
+		}
 	}
 
-	if eb, err = json.Marshal(extra); err != nil {
-		return nil, err
-	}
-
-	// merge serialized message + serialized extra map
-	b[len(b)-1] = ','
-	return append(b, eb[1:len(eb)]...), nil
+	// write final closing quotes
+	return buf.WriteByte('}')
 }
 
 func (m *Message) UnmarshalJSON(data []byte) error {