upgrade go-gelf vendor

Signed-off-by: Daniel Dao <dqminh@cloudflare.com>
This commit is contained in:
Daniel Dao 2016-01-21 12:28:25 +00:00
parent 9ad946eded
commit ae6ee5e4af
3 changed files with 97 additions and 54 deletions

View File

@ -68,7 +68,7 @@ clone git github.com/syndtr/gocapability 2c00daeb6c3b45114c80ac44119e7b8801fdd85
clone git github.com/golang/protobuf f7137ae6b19afbfd61a94b746fda3b3fe0491874 clone git github.com/golang/protobuf f7137ae6b19afbfd61a94b746fda3b3fe0491874
# gelf logging driver deps # gelf logging driver deps
clone git github.com/Graylog2/go-gelf 6c62a85f1d47a67f2a5144c0e745b325889a8120 clone git github.com/Graylog2/go-gelf aab2f594e4585d43468ac57287b0dece9d806883
clone git github.com/fluent/fluent-logger-golang v1.0.0 clone git github.com/fluent/fluent-logger-golang v1.0.0
# fluent-logger-golang deps # fluent-logger-golang deps

View File

@ -66,7 +66,6 @@ func (r *Reader) ReadMessage() (*Message, error) {
var ( var (
err error err error
n, length int n, length int
buf bytes.Buffer
cid, ocid []byte cid, ocid []byte
seq, total uint8 seq, total uint8
cHead []byte cHead []byte
@ -122,19 +121,18 @@ func (r *Reader) ReadMessage() (*Message, error) {
// zlib is slightly more complicated, but correct // zlib is slightly more complicated, but correct
cReader, err = zlib.NewReader(bytes.NewReader(cBuf)) cReader, err = zlib.NewReader(bytes.NewReader(cBuf))
} else { } 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 { if err != nil {
return nil, fmt.Errorf("NewReader: %s", err) 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) 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) return nil, fmt.Errorf("json.Unmarshal: %s", err)
} }

View File

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