Merge pull request #39074 from thaJeztah/bump_fluentd

Bump fluent/fluent-logger-golang v1.4.0

daemon/logger/fluentd/fluentd.go

@@ -133,7 +133,7 @@ func New(info logger.Info) (logger.Logger, error) {
 		BufferLimit:        bufferLimit,
 		RetryWait:          retryWait,
 		MaxRetry:           maxRetries,
-		AsyncConnect:       asyncConnect,
+		Async:              asyncConnect,
 		SubSecondPrecision: subSecondPrecision,
 	}
 

vendor.conf

@@ -95,9 +95,9 @@ github.com/golang/protobuf                          aa810b61a9c79d51363740d207bb
 github.com/Graylog2/go-gelf                         4143646226541087117ff2f83334ea48b3201841
 
 # fluent-logger-golang deps
-github.com/fluent/fluent-logger-golang              8bbc2356beaf021b04c9bd5cdc76ea5a7ccb40ec # v1.3.0
+github.com/fluent/fluent-logger-golang              7a6c9dcd7f14c2ed5d8c55c11b894e5455ee311b # v1.4.0
 github.com/philhofer/fwd                            bb6d471dc95d4fe11e432687f8b70ff496cf3136 # v1.0.0
-github.com/tinylib/msgp                             3b556c64540842d4f82967be066a7f7fffc3adad
+github.com/tinylib/msgp                             af6442a0fcf6e2a1b824f70dd0c734f01e817751 # v1.1.0
 
 # fsnotify
 github.com/fsnotify/fsnotify                        1485a34d5d5723fea214f5710708e19a831720e4 # v1.4.7-11-g1485a34

vendor/github.com/fluent/fluent-logger-golang/README.md

@@ -64,6 +64,16 @@ f := fluent.New(fluent.Config{FluentPort: 80, FluentHost: "example.com"})
 Sets the timeout for Write call of logger.Post.
 Since the default is zero value, Write will not time out.
 
+### Async
+
+Enable asynchronous I/O (connect and write) for sending events to Fluentd.
+The default is false.
+
+### RequestAck
+
+Sets whether to request acknowledgment from Fluentd to increase the reliability
+of the connection. The default is false.
+
 ## Tests
 ```
 go test

vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go

@@ -6,12 +6,17 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"os"
 	"reflect"
 	"strconv"
 	"sync"
 	"time"
 
+	"bytes"
+	"encoding/base64"
+	"encoding/binary"
 	"github.com/tinylib/msgp/msgp"
+	"math/rand"
 )
 
 const (
@@ -21,8 +26,9 @@ const (
 	defaultPort                   = 24224
 	defaultTimeout                = 3 * time.Second
 	defaultWriteTimeout           = time.Duration(0) // Write() will not time out
-	defaultBufferLimit            = 8 * 1024 * 1024
+	defaultBufferLimit            = 8 * 1024
 	defaultRetryWait              = 500
+	defaultMaxRetryWait           = 60000
 	defaultMaxRetry               = 13
 	defaultReconnectWaitIncreRate = 1.5
 	// Default sub-second precision value to false since it is only compatible
@@ -40,24 +46,36 @@ type Config struct {
 	BufferLimit      int           `json:"buffer_limit"`
 	RetryWait        int           `json:"retry_wait"`
 	MaxRetry         int           `json:"max_retry"`
+	MaxRetryWait     int           `json:"max_retry_wait"`
 	TagPrefix        string        `json:"tag_prefix"`
-	AsyncConnect     bool          `json:"async_connect"`
-	MarshalAsJSON    bool          `json:"marshal_as_json"`
+	Async            bool          `json:"async"`
+	// Deprecated: Use Async instead
+	AsyncConnect  bool `json:"async_connect"`
+	MarshalAsJSON bool `json:"marshal_as_json"`
 
 	// Sub-second precision timestamps are only possible for those using fluentd
 	// v0.14+ and serializing their messages with msgpack.
 	SubSecondPrecision bool `json:"sub_second_precision"`
+
+	// RequestAck sends the chunk option with a unique ID. The server will
+	// respond with an acknowledgement. This option improves the reliability
+	// of the message transmission.
+	RequestAck bool `json:"request_ack"`
+}
+
+type msgToSend struct {
+	data []byte
+	ack  string
 }
 
 type Fluent struct {
 	Config
 
-	mubuff  sync.Mutex
-	pending []byte
+	pending chan *msgToSend
+	wg      sync.WaitGroup
 
-	muconn       sync.Mutex
-	conn         net.Conn
-	reconnecting bool
+	muconn sync.Mutex
+	conn   net.Conn
 }
 
 // New creates a new Logger.
@@ -89,11 +107,22 @@ func New(config Config) (f *Fluent, err error) {
 	if config.MaxRetry == 0 {
 		config.MaxRetry = defaultMaxRetry
 	}
+	if config.MaxRetryWait == 0 {
+		config.MaxRetryWait = defaultMaxRetryWait
+	}
 	if config.AsyncConnect {
-		f = &Fluent{Config: config, reconnecting: true}
-		go f.reconnect()
+		fmt.Fprintf(os.Stderr, "fluent#New: AsyncConnect is now deprecated, please use Async instead")
+		config.Async = config.Async || config.AsyncConnect
+	}
+	if config.Async {
+		f = &Fluent{
+			Config:  config,
+			pending: make(chan *msgToSend, config.BufferLimit),
+		}
+		f.wg.Add(1)
+		go f.run()
 	} else {
-		f = &Fluent{Config: config, reconnecting: false}
+		f = &Fluent{Config: config}
 		err = f.connect()
 	}
 	return
@@ -173,28 +202,25 @@ func (f *Fluent) PostWithTime(tag string, tm time.Time, message interface{}) err
 }
 
 func (f *Fluent) EncodeAndPostData(tag string, tm time.Time, message interface{}) error {
-	var data []byte
+	var msg *msgToSend
 	var err error
-	if data, err = f.EncodeData(tag, tm, message); err != nil {
+	if msg, err = f.EncodeData(tag, tm, message); err != nil {
 		return fmt.Errorf("fluent#EncodeAndPostData: can't convert '%#v' to msgpack:%v", message, err)
 	}
-	return f.postRawData(data)
+	return f.postRawData(msg)
 }
 
 // Deprecated: Use EncodeAndPostData instead
-func (f *Fluent) PostRawData(data []byte) {
-	f.postRawData(data)
+func (f *Fluent) PostRawData(msg *msgToSend) {
+	f.postRawData(msg)
 }
 
-func (f *Fluent) postRawData(data []byte) error {
-	if err := f.appendBuffer(data); err != nil {
-		return err
-	}
-	if err := f.send(); err != nil {
-		f.close()
-		return err
+func (f *Fluent) postRawData(msg *msgToSend) error {
+	if f.Config.Async {
+		return f.appendBuffer(msg)
 	}
-	return nil
+	// Synchronous write
+	return f.write(msg)
 }
 
 // For sending forward protocol adopted JSON
@@ -207,43 +233,80 @@ type MessageChunk struct {
 // So, it should write JSON marshaler by hand.
 func (chunk *MessageChunk) MarshalJSON() ([]byte, error) {
 	data, err := json.Marshal(chunk.message.Record)
-	return []byte(fmt.Sprintf("[\"%s\",%d,%s,null]", chunk.message.Tag,
-		chunk.message.Time, data)), err
+	if err != nil {
+		return nil, err
+	}
+	option, err := json.Marshal(chunk.message.Option)
+	if err != nil {
+		return nil, err
+	}
+	return []byte(fmt.Sprintf("[\"%s\",%d,%s,%s]", chunk.message.Tag,
+		chunk.message.Time, data, option)), err
 }
 
-func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (data []byte, err error) {
+// getUniqueID returns a base64 encoded unique ID that can be used for chunk/ack
+// mechanism, see
+// https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#option
+func getUniqueID(timeUnix int64) (string, error) {
+	buf := bytes.NewBuffer(nil)
+	enc := base64.NewEncoder(base64.StdEncoding, buf)
+	if err := binary.Write(enc, binary.LittleEndian, timeUnix); err != nil {
+		enc.Close()
+		return "", err
+	}
+	if err := binary.Write(enc, binary.LittleEndian, rand.Uint64()); err != nil {
+		enc.Close()
+		return "", err
+	}
+	// encoder needs to be closed before buf.String(), defer does not work
+	// here
+	enc.Close()
+	return buf.String(), nil
+}
+
+func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (msg *msgToSend, err error) {
+	option := make(map[string]string)
+	msg = &msgToSend{}
 	timeUnix := tm.Unix()
+	if f.Config.RequestAck {
+		var err error
+		msg.ack, err = getUniqueID(timeUnix)
+		if err != nil {
+			return nil, err
+		}
+		option["chunk"] = msg.ack
+	}
 	if f.Config.MarshalAsJSON {
-		msg := Message{Tag: tag, Time: timeUnix, Record: message}
-		chunk := &MessageChunk{message: msg}
-		data, err = json.Marshal(chunk)
+		m := Message{Tag: tag, Time: timeUnix, Record: message, Option: option}
+		chunk := &MessageChunk{message: m}
+		msg.data, err = json.Marshal(chunk)
 	} else if f.Config.SubSecondPrecision {
-		msg := &MessageExt{Tag: tag, Time: EventTime(tm), Record: message}
-		data, err = msg.MarshalMsg(nil)
+		m := &MessageExt{Tag: tag, Time: EventTime(tm), Record: message, Option: option}
+		msg.data, err = m.MarshalMsg(nil)
 	} else {
-		msg := &Message{Tag: tag, Time: timeUnix, Record: message}
-		data, err = msg.MarshalMsg(nil)
+		m := &Message{Tag: tag, Time: timeUnix, Record: message, Option: option}
+		msg.data, err = m.MarshalMsg(nil)
 	}
 	return
 }
 
-// Close closes the connection.
+// Close closes the connection, waiting for pending logs to be sent
 func (f *Fluent) Close() (err error) {
-	if len(f.pending) > 0 {
-		err = f.send()
+	if f.Config.Async {
+		close(f.pending)
+		f.wg.Wait()
 	}
 	f.close()
 	return
 }
 
 // appendBuffer appends data to buffer with lock.
-func (f *Fluent) appendBuffer(data []byte) error {
-	f.mubuff.Lock()
-	defer f.mubuff.Unlock()
-	if len(f.pending)+len(data) > f.Config.BufferLimit {
-		return errors.New(fmt.Sprintf("fluent#appendBuffer: Buffer full, limit %v", f.Config.BufferLimit))
+func (f *Fluent) appendBuffer(msg *msgToSend) error {
+	select {
+	case f.pending <- msg:
+	default:
+		return fmt.Errorf("fluent#appendBuffer: Buffer full, limit %v", f.Config.BufferLimit)
 	}
-	f.pending = append(f.pending, data...)
 	return nil
 }
 
@@ -259,8 +322,6 @@ func (f *Fluent) close() {
 
 // connect establishes a new connection using the specified transport.
 func (f *Fluent) connect() (err error) {
-	f.muconn.Lock()
-	defer f.muconn.Unlock()
 
 	switch f.Config.FluentNetwork {
 	case "tcp":
@@ -270,63 +331,78 @@ func (f *Fluent) connect() (err error) {
 	default:
 		err = net.UnknownNetworkError(f.Config.FluentNetwork)
 	}
+	return err
+}
 
-	if err == nil {
-		f.reconnecting = false
+func (f *Fluent) run() {
+	for {
+		select {
+		case entry, ok := <-f.pending:
+			if !ok {
+				f.wg.Done()
+				return
+			}
+			err := f.write(entry)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "[%s] Unable to send logs to fluentd, reconnecting...\n", time.Now().Format(time.RFC3339))
+			}
+		}
 	}
-	return
 }
 
 func e(x, y float64) int {
 	return int(math.Pow(x, y))
 }
 
-func (f *Fluent) reconnect() {
-	for i := 0; ; i++ {
-		err := f.connect()
-		if err == nil {
-			f.send()
-			return
-		}
-		if i == f.Config.MaxRetry {
-			// TODO: What we can do when connection failed MaxRetry times?
-			panic("fluent#reconnect: failed to reconnect!")
-		}
-		waitTime := f.Config.RetryWait * e(defaultReconnectWaitIncreRate, float64(i-1))
-		time.Sleep(time.Duration(waitTime) * time.Millisecond)
-	}
-}
-
-func (f *Fluent) send() error {
-	f.muconn.Lock()
-	defer f.muconn.Unlock()
-
-	if f.conn == nil {
-		if f.reconnecting == false {
-			f.reconnecting = true
-			go f.reconnect()
+func (f *Fluent) write(msg *msgToSend) error {
+
+	for i := 0; i < f.Config.MaxRetry; i++ {
+
+		// Connect if needed
+		f.muconn.Lock()
+		if f.conn == nil {
+			err := f.connect()
+			if err != nil {
+				f.muconn.Unlock()
+				waitTime := f.Config.RetryWait * e(defaultReconnectWaitIncreRate, float64(i-1))
+				if waitTime > f.Config.MaxRetryWait {
+					waitTime = f.Config.MaxRetryWait
+				}
+				time.Sleep(time.Duration(waitTime) * time.Millisecond)
+				continue
+			}
 		}
-		return errors.New("fluent#send: can't send logs, client is reconnecting")
-	}
-
-	f.mubuff.Lock()
-	defer f.mubuff.Unlock()
+		f.muconn.Unlock()
 
-	var err error
-	if len(f.pending) > 0 {
+		// We're connected, write msg
 		t := f.Config.WriteTimeout
 		if time.Duration(0) < t {
 			f.conn.SetWriteDeadline(time.Now().Add(t))
 		} else {
 			f.conn.SetWriteDeadline(time.Time{})
 		}
-		_, err = f.conn.Write(f.pending)
+		_, err := f.conn.Write(msg.data)
 		if err != nil {
-			f.conn.Close()
-			f.conn = nil
+			f.close()
 		} else {
-			f.pending = f.pending[:0]
+			// Acknowledgment check
+			if msg.ack != "" {
+				resp := &AckResp{}
+				if f.Config.MarshalAsJSON {
+					dec := json.NewDecoder(f.conn)
+					err = dec.Decode(resp)
+				} else {
+					r := msgp.NewReader(f.conn)
+					err = resp.DecodeMsg(r)
+				}
+				if err != nil || resp.Ack != msg.ack {
+					f.close()
+					continue
+				}
+			}
+			return err
 		}
 	}
-	return err
+
+	return fmt.Errorf("fluent#write: failed to reconnect, max retry: %v", f.Config.MaxRetry)
 }

vendor/github.com/fluent/fluent-logger-golang/fluent/proto.go

@@ -16,9 +16,9 @@ type Entry struct {
 
 //msgp:tuple Forward
 type Forward struct {
-	Tag     string      `msg:"tag"`
-	Entries []Entry     `msg:"entries"`
-	Option  interface{} `msg:"option"`
+	Tag     string  `msg:"tag"`
+	Entries []Entry `msg:"entries"`
+	Option  map[string]string
 }
 
 //msgp:tuple Message
@@ -26,7 +26,7 @@ type Message struct {
 	Tag    string      `msg:"tag"`
 	Time   int64       `msg:"time"`
 	Record interface{} `msg:"record"`
-	Option interface{} `msg:"option"`
+	Option map[string]string
 }
 
 //msgp:tuple MessageExt
@@ -34,7 +34,11 @@ type MessageExt struct {
 	Tag    string      `msg:"tag"`
 	Time   EventTime   `msg:"time,extension"`
 	Record interface{} `msg:"record"`
-	Option interface{} `msg:"option"`
+	Option map[string]string
+}
+
+type AckResp struct {
+	Ack string `json:"ack" msg:"ack"`
 }
 
 // EventTime is an extension to the serialized time value. It builds in support

vendor/github.com/fluent/fluent-logger-golang/fluent/proto_gen.go

@@ -1,30 +1,134 @@
 package fluent
 
-// NOTE: THIS FILE WAS PRODUCED BY THE
-// MSGP CODE GENERATION TOOL (github.com/tinylib/msgp)
-// DO NOT EDIT
+// Code generated by github.com/tinylib/msgp DO NOT EDIT.
 
 import (
 	"github.com/tinylib/msgp/msgp"
 )
 
+// DecodeMsg implements msgp.Decodable
+func (z *AckResp) DecodeMsg(dc *msgp.Reader) (err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, err = dc.ReadMapHeader()
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, err = dc.ReadMapKeyPtr()
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "ack":
+			z.Ack, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "Ack")
+				return
+			}
+		default:
+			err = dc.Skip()
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	return
+}
+
+// EncodeMsg implements msgp.Encodable
+func (z AckResp) EncodeMsg(en *msgp.Writer) (err error) {
+	// map header, size 1
+	// write "ack"
+	err = en.Append(0x81, 0xa3, 0x61, 0x63, 0x6b)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.Ack)
+	if err != nil {
+		err = msgp.WrapError(err, "Ack")
+		return
+	}
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z AckResp) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 1
+	// string "ack"
+	o = append(o, 0x81, 0xa3, 0x61, 0x63, 0x6b)
+	o = msgp.AppendString(o, z.Ack)
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *AckResp) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "ack":
+			z.Ack, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Ack")
+				return
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z AckResp) Msgsize() (s int) {
+	s = 1 + 4 + msgp.StringPrefixSize + len(z.Ack)
+	return
+}
+
 // DecodeMsg implements msgp.Decodable
 func (z *Entry) DecodeMsg(dc *msgp.Reader) (err error) {
-	var zxvk uint32
-	zxvk, err = dc.ReadArrayHeader()
+	var zb0001 uint32
+	zb0001, err = dc.ReadArrayHeader()
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zxvk != 2 {
-		err = msgp.ArrayError{Wanted: 2, Got: zxvk}
+	if zb0001 != 2 {
+		err = msgp.ArrayError{Wanted: 2, Got: zb0001}
 		return
 	}
 	z.Time, err = dc.ReadInt64()
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, err = dc.ReadIntf()
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
 	return
@@ -35,14 +139,16 @@ func (z Entry) EncodeMsg(en *msgp.Writer) (err error) {
 	// array header, size 2
 	err = en.Append(0x92)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteInt64(z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	err = en.WriteIntf(z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
 	return
@@ -56,6 +162,7 @@ func (z Entry) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.AppendInt64(o, z.Time)
 	o, err = msgp.AppendIntf(o, z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
 	return
@@ -63,21 +170,24 @@ func (z Entry) MarshalMsg(b []byte) (o []byte, err error) {
 
 // UnmarshalMsg implements msgp.Unmarshaler
 func (z *Entry) UnmarshalMsg(bts []byte) (o []byte, err error) {
-	var zbzg uint32
-	zbzg, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zbzg != 2 {
-		err = msgp.ArrayError{Wanted: 2, Got: zbzg}
+	if zb0001 != 2 {
+		err = msgp.ArrayError{Wanted: 2, Got: zb0001}
 		return
 	}
 	z.Time, bts, err = msgp.ReadInt64Bytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, bts, err = msgp.ReadIntfBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
 	o = bts
@@ -92,52 +202,83 @@ func (z Entry) Msgsize() (s int) {
 
 // DecodeMsg implements msgp.Decodable
 func (z *Forward) DecodeMsg(dc *msgp.Reader) (err error) {
-	var zcmr uint32
-	zcmr, err = dc.ReadArrayHeader()
+	var zb0001 uint32
+	zb0001, err = dc.ReadArrayHeader()
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zcmr != 3 {
-		err = msgp.ArrayError{Wanted: 3, Got: zcmr}
+	if zb0001 != 3 {
+		err = msgp.ArrayError{Wanted: 3, Got: zb0001}
 		return
 	}
 	z.Tag, err = dc.ReadString()
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
-	var zajw uint32
-	zajw, err = dc.ReadArrayHeader()
+	var zb0002 uint32
+	zb0002, err = dc.ReadArrayHeader()
 	if err != nil {
+		err = msgp.WrapError(err, "Entries")
 		return
 	}
-	if cap(z.Entries) >= int(zajw) {
-		z.Entries = (z.Entries)[:zajw]
+	if cap(z.Entries) >= int(zb0002) {
+		z.Entries = (z.Entries)[:zb0002]
 	} else {
-		z.Entries = make([]Entry, zajw)
+		z.Entries = make([]Entry, zb0002)
 	}
-	for zbai := range z.Entries {
-		var zwht uint32
-		zwht, err = dc.ReadArrayHeader()
+	for za0001 := range z.Entries {
+		var zb0003 uint32
+		zb0003, err = dc.ReadArrayHeader()
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001)
 			return
 		}
-		if zwht != 2 {
-			err = msgp.ArrayError{Wanted: 2, Got: zwht}
+		if zb0003 != 2 {
+			err = msgp.ArrayError{Wanted: 2, Got: zb0003}
 			return
 		}
-		z.Entries[zbai].Time, err = dc.ReadInt64()
+		z.Entries[za0001].Time, err = dc.ReadInt64()
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Time")
 			return
 		}
-		z.Entries[zbai].Record, err = dc.ReadIntf()
+		z.Entries[za0001].Record, err = dc.ReadIntf()
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Record")
 			return
 		}
 	}
-	z.Option, err = dc.ReadIntf()
+	var zb0004 uint32
+	zb0004, err = dc.ReadMapHeader()
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0004)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0004 > 0 {
+		zb0004--
+		var za0002 string
+		var za0003 string
+		za0002, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0003, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0002)
+			return
+		}
+		z.Option[za0002] = za0003
+	}
 	return
 }
 
@@ -146,35 +287,52 @@ func (z *Forward) EncodeMsg(en *msgp.Writer) (err error) {
 	// array header, size 3
 	err = en.Append(0x93)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteString(z.Tag)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	err = en.WriteArrayHeader(uint32(len(z.Entries)))
 	if err != nil {
+		err = msgp.WrapError(err, "Entries")
 		return
 	}
-	for zbai := range z.Entries {
+	for za0001 := range z.Entries {
 		// array header, size 2
 		err = en.Append(0x92)
 		if err != nil {
-			return err
+			return
 		}
-		err = en.WriteInt64(z.Entries[zbai].Time)
+		err = en.WriteInt64(z.Entries[za0001].Time)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Time")
 			return
 		}
-		err = en.WriteIntf(z.Entries[zbai].Record)
+		err = en.WriteIntf(z.Entries[za0001].Record)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Record")
 			return
 		}
 	}
-	err = en.WriteIntf(z.Option)
+	err = en.WriteMapHeader(uint32(len(z.Option)))
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	for za0002, za0003 := range z.Option {
+		err = en.WriteString(za0002)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		err = en.WriteString(za0003)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0002)
+			return
+		}
+	}
 	return
 }
 
@@ -185,70 +343,103 @@ func (z *Forward) MarshalMsg(b []byte) (o []byte, err error) {
 	o = append(o, 0x93)
 	o = msgp.AppendString(o, z.Tag)
 	o = msgp.AppendArrayHeader(o, uint32(len(z.Entries)))
-	for zbai := range z.Entries {
+	for za0001 := range z.Entries {
 		// array header, size 2
 		o = append(o, 0x92)
-		o = msgp.AppendInt64(o, z.Entries[zbai].Time)
-		o, err = msgp.AppendIntf(o, z.Entries[zbai].Record)
+		o = msgp.AppendInt64(o, z.Entries[za0001].Time)
+		o, err = msgp.AppendIntf(o, z.Entries[za0001].Record)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Record")
 			return
 		}
 	}
-	o, err = msgp.AppendIntf(o, z.Option)
-	if err != nil {
-		return
+	o = msgp.AppendMapHeader(o, uint32(len(z.Option)))
+	for za0002, za0003 := range z.Option {
+		o = msgp.AppendString(o, za0002)
+		o = msgp.AppendString(o, za0003)
 	}
 	return
 }
 
 // UnmarshalMsg implements msgp.Unmarshaler
 func (z *Forward) UnmarshalMsg(bts []byte) (o []byte, err error) {
-	var zhct uint32
-	zhct, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zhct != 3 {
-		err = msgp.ArrayError{Wanted: 3, Got: zhct}
+	if zb0001 != 3 {
+		err = msgp.ArrayError{Wanted: 3, Got: zb0001}
 		return
 	}
 	z.Tag, bts, err = msgp.ReadStringBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
-	var zcua uint32
-	zcua, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	var zb0002 uint32
+	zb0002, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Entries")
 		return
 	}
-	if cap(z.Entries) >= int(zcua) {
-		z.Entries = (z.Entries)[:zcua]
+	if cap(z.Entries) >= int(zb0002) {
+		z.Entries = (z.Entries)[:zb0002]
 	} else {
-		z.Entries = make([]Entry, zcua)
+		z.Entries = make([]Entry, zb0002)
 	}
-	for zbai := range z.Entries {
-		var zxhx uint32
-		zxhx, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	for za0001 := range z.Entries {
+		var zb0003 uint32
+		zb0003, bts, err = msgp.ReadArrayHeaderBytes(bts)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001)
 			return
 		}
-		if zxhx != 2 {
-			err = msgp.ArrayError{Wanted: 2, Got: zxhx}
+		if zb0003 != 2 {
+			err = msgp.ArrayError{Wanted: 2, Got: zb0003}
 			return
 		}
-		z.Entries[zbai].Time, bts, err = msgp.ReadInt64Bytes(bts)
+		z.Entries[za0001].Time, bts, err = msgp.ReadInt64Bytes(bts)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Time")
 			return
 		}
-		z.Entries[zbai].Record, bts, err = msgp.ReadIntfBytes(bts)
+		z.Entries[za0001].Record, bts, err = msgp.ReadIntfBytes(bts)
 		if err != nil {
+			err = msgp.WrapError(err, "Entries", za0001, "Record")
 			return
 		}
 	}
-	z.Option, bts, err = msgp.ReadIntfBytes(bts)
+	var zb0004 uint32
+	zb0004, bts, err = msgp.ReadMapHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0004)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0004 > 0 {
+		var za0002 string
+		var za0003 string
+		zb0004--
+		za0002, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0003, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0002)
+			return
+		}
+		z.Option[za0002] = za0003
+	}
 	o = bts
 	return
 }
@@ -256,40 +447,75 @@ func (z *Forward) UnmarshalMsg(bts []byte) (o []byte, err error) {
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *Forward) Msgsize() (s int) {
 	s = 1 + msgp.StringPrefixSize + len(z.Tag) + msgp.ArrayHeaderSize
-	for zbai := range z.Entries {
-		s += 1 + msgp.Int64Size + msgp.GuessSize(z.Entries[zbai].Record)
+	for za0001 := range z.Entries {
+		s += 1 + msgp.Int64Size + msgp.GuessSize(z.Entries[za0001].Record)
+	}
+	s += msgp.MapHeaderSize
+	if z.Option != nil {
+		for za0002, za0003 := range z.Option {
+			_ = za0003
+			s += msgp.StringPrefixSize + len(za0002) + msgp.StringPrefixSize + len(za0003)
+		}
 	}
-	s += msgp.GuessSize(z.Option)
 	return
 }
 
 // DecodeMsg implements msgp.Decodable
 func (z *Message) DecodeMsg(dc *msgp.Reader) (err error) {
-	var zlqf uint32
-	zlqf, err = dc.ReadArrayHeader()
+	var zb0001 uint32
+	zb0001, err = dc.ReadArrayHeader()
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zlqf != 4 {
-		err = msgp.ArrayError{Wanted: 4, Got: zlqf}
+	if zb0001 != 4 {
+		err = msgp.ArrayError{Wanted: 4, Got: zb0001}
 		return
 	}
 	z.Tag, err = dc.ReadString()
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	z.Time, err = dc.ReadInt64()
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, err = dc.ReadIntf()
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	z.Option, err = dc.ReadIntf()
+	var zb0002 uint32
+	zb0002, err = dc.ReadMapHeader()
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0002)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0002 > 0 {
+		zb0002--
+		var za0001 string
+		var za0002 string
+		za0001, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0002, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+		z.Option[za0001] = za0002
+	}
 	return
 }
 
@@ -298,24 +524,40 @@ func (z *Message) EncodeMsg(en *msgp.Writer) (err error) {
 	// array header, size 4
 	err = en.Append(0x94)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteString(z.Tag)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	err = en.WriteInt64(z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	err = en.WriteIntf(z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	err = en.WriteIntf(z.Option)
+	err = en.WriteMapHeader(uint32(len(z.Option)))
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	for za0001, za0002 := range z.Option {
+		err = en.WriteString(za0001)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		err = en.WriteString(za0002)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+	}
 	return
 }
 
@@ -328,79 +570,145 @@ func (z *Message) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.AppendInt64(o, z.Time)
 	o, err = msgp.AppendIntf(o, z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	o, err = msgp.AppendIntf(o, z.Option)
-	if err != nil {
-		return
+	o = msgp.AppendMapHeader(o, uint32(len(z.Option)))
+	for za0001, za0002 := range z.Option {
+		o = msgp.AppendString(o, za0001)
+		o = msgp.AppendString(o, za0002)
 	}
 	return
 }
 
 // UnmarshalMsg implements msgp.Unmarshaler
 func (z *Message) UnmarshalMsg(bts []byte) (o []byte, err error) {
-	var zdaf uint32
-	zdaf, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zdaf != 4 {
-		err = msgp.ArrayError{Wanted: 4, Got: zdaf}
+	if zb0001 != 4 {
+		err = msgp.ArrayError{Wanted: 4, Got: zb0001}
 		return
 	}
 	z.Tag, bts, err = msgp.ReadStringBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	z.Time, bts, err = msgp.ReadInt64Bytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, bts, err = msgp.ReadIntfBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	z.Option, bts, err = msgp.ReadIntfBytes(bts)
+	var zb0002 uint32
+	zb0002, bts, err = msgp.ReadMapHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0002)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0002 > 0 {
+		var za0001 string
+		var za0002 string
+		zb0002--
+		za0001, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0002, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+		z.Option[za0001] = za0002
+	}
 	o = bts
 	return
 }
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *Message) Msgsize() (s int) {
-	s = 1 + msgp.StringPrefixSize + len(z.Tag) + msgp.Int64Size + msgp.GuessSize(z.Record) + msgp.GuessSize(z.Option)
+	s = 1 + msgp.StringPrefixSize + len(z.Tag) + msgp.Int64Size + msgp.GuessSize(z.Record) + msgp.MapHeaderSize
+	if z.Option != nil {
+		for za0001, za0002 := range z.Option {
+			_ = za0002
+			s += msgp.StringPrefixSize + len(za0001) + msgp.StringPrefixSize + len(za0002)
+		}
+	}
 	return
 }
 
 // DecodeMsg implements msgp.Decodable
 func (z *MessageExt) DecodeMsg(dc *msgp.Reader) (err error) {
-	var zpks uint32
-	zpks, err = dc.ReadArrayHeader()
+	var zb0001 uint32
+	zb0001, err = dc.ReadArrayHeader()
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zpks != 4 {
-		err = msgp.ArrayError{Wanted: 4, Got: zpks}
+	if zb0001 != 4 {
+		err = msgp.ArrayError{Wanted: 4, Got: zb0001}
 		return
 	}
 	z.Tag, err = dc.ReadString()
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	err = dc.ReadExtension(&z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, err = dc.ReadIntf()
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	z.Option, err = dc.ReadIntf()
+	var zb0002 uint32
+	zb0002, err = dc.ReadMapHeader()
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0002)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0002 > 0 {
+		zb0002--
+		var za0001 string
+		var za0002 string
+		za0001, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0002, err = dc.ReadString()
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+		z.Option[za0001] = za0002
+	}
 	return
 }
 
@@ -409,24 +717,40 @@ func (z *MessageExt) EncodeMsg(en *msgp.Writer) (err error) {
 	// array header, size 4
 	err = en.Append(0x94)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteString(z.Tag)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	err = en.WriteExtension(&z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	err = en.WriteIntf(z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	err = en.WriteIntf(z.Option)
+	err = en.WriteMapHeader(uint32(len(z.Option)))
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	for za0001, za0002 := range z.Option {
+		err = en.WriteString(za0001)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		err = en.WriteString(za0002)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+	}
 	return
 }
 
@@ -438,52 +762,90 @@ func (z *MessageExt) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.AppendString(o, z.Tag)
 	o, err = msgp.AppendExtension(o, &z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	o, err = msgp.AppendIntf(o, z.Record)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	o, err = msgp.AppendIntf(o, z.Option)
-	if err != nil {
-		return
+	o = msgp.AppendMapHeader(o, uint32(len(z.Option)))
+	for za0001, za0002 := range z.Option {
+		o = msgp.AppendString(o, za0001)
+		o = msgp.AppendString(o, za0002)
 	}
 	return
 }
 
 // UnmarshalMsg implements msgp.Unmarshaler
 func (z *MessageExt) UnmarshalMsg(bts []byte) (o []byte, err error) {
-	var zjfb uint32
-	zjfb, bts, err = msgp.ReadArrayHeaderBytes(bts)
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	if zjfb != 4 {
-		err = msgp.ArrayError{Wanted: 4, Got: zjfb}
+	if zb0001 != 4 {
+		err = msgp.ArrayError{Wanted: 4, Got: zb0001}
 		return
 	}
 	z.Tag, bts, err = msgp.ReadStringBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Tag")
 		return
 	}
 	bts, err = msgp.ReadExtensionBytes(bts, &z.Time)
 	if err != nil {
+		err = msgp.WrapError(err, "Time")
 		return
 	}
 	z.Record, bts, err = msgp.ReadIntfBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Record")
 		return
 	}
-	z.Option, bts, err = msgp.ReadIntfBytes(bts)
+	var zb0002 uint32
+	zb0002, bts, err = msgp.ReadMapHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err, "Option")
 		return
 	}
+	if z.Option == nil {
+		z.Option = make(map[string]string, zb0002)
+	} else if len(z.Option) > 0 {
+		for key := range z.Option {
+			delete(z.Option, key)
+		}
+	}
+	for zb0002 > 0 {
+		var za0001 string
+		var za0002 string
+		zb0002--
+		za0001, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option")
+			return
+		}
+		za0002, bts, err = msgp.ReadStringBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err, "Option", za0001)
+			return
+		}
+		z.Option[za0001] = za0002
+	}
 	o = bts
 	return
 }
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *MessageExt) Msgsize() (s int) {
-	s = 1 + msgp.StringPrefixSize + len(z.Tag) + msgp.ExtensionPrefixSize + z.Time.Len() + msgp.GuessSize(z.Record) + msgp.GuessSize(z.Option)
+	s = 1 + msgp.StringPrefixSize + len(z.Tag) + msgp.ExtensionPrefixSize + z.Time.Len() + msgp.GuessSize(z.Record) + msgp.MapHeaderSize
+	if z.Option != nil {
+		for za0001, za0002 := range z.Option {
+			_ = za0002
+			s += msgp.StringPrefixSize + len(za0001) + msgp.StringPrefixSize + len(za0002)
+		}
+	}
 	return
 }

vendor/github.com/fluent/fluent-logger-golang/fluent/test_message_gen.go

@@ -1,8 +1,6 @@
 package fluent
 
-// NOTE: THIS FILE WAS PRODUCED BY THE
-// MSGP CODE GENERATION TOOL (github.com/tinylib/msgp)
-// DO NOT EDIT
+// Code generated by github.com/tinylib/msgp DO NOT EDIT.
 
 import (
 	"github.com/tinylib/msgp/msgp"
@@ -12,31 +10,36 @@ import (
 func (z *TestMessage) DecodeMsg(dc *msgp.Reader) (err error) {
 	var field []byte
 	_ = field
-	var zxvk uint32
-	zxvk, err = dc.ReadMapHeader()
+	var zb0001 uint32
+	zb0001, err = dc.ReadMapHeader()
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	for zxvk > 0 {
-		zxvk--
+	for zb0001 > 0 {
+		zb0001--
 		field, err = dc.ReadMapKeyPtr()
 		if err != nil {
+			err = msgp.WrapError(err)
 			return
 		}
 		switch msgp.UnsafeString(field) {
 		case "foo":
 			z.Foo, err = dc.ReadString()
 			if err != nil {
+				err = msgp.WrapError(err, "Foo")
 				return
 			}
 		case "hoge":
 			z.Hoge, err = dc.ReadString()
 			if err != nil {
+				err = msgp.WrapError(err, "Hoge")
 				return
 			}
 		default:
 			err = dc.Skip()
 			if err != nil {
+				err = msgp.WrapError(err)
 				return
 			}
 		}
@@ -50,19 +53,21 @@ func (z TestMessage) EncodeMsg(en *msgp.Writer) (err error) {
 	// write "foo"
 	err = en.Append(0x82, 0xa3, 0x66, 0x6f, 0x6f)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteString(z.Foo)
 	if err != nil {
+		err = msgp.WrapError(err, "Foo")
 		return
 	}
 	// write "hoge"
 	err = en.Append(0xa4, 0x68, 0x6f, 0x67, 0x65)
 	if err != nil {
-		return err
+		return
 	}
 	err = en.WriteString(z.Hoge)
 	if err != nil {
+		err = msgp.WrapError(err, "Hoge")
 		return
 	}
 	return
@@ -85,31 +90,36 @@ func (z TestMessage) MarshalMsg(b []byte) (o []byte, err error) {
 func (z *TestMessage) UnmarshalMsg(bts []byte) (o []byte, err error) {
 	var field []byte
 	_ = field
-	var zbzg uint32
-	zbzg, bts, err = msgp.ReadMapHeaderBytes(bts)
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
 	if err != nil {
+		err = msgp.WrapError(err)
 		return
 	}
-	for zbzg > 0 {
-		zbzg--
+	for zb0001 > 0 {
+		zb0001--
 		field, bts, err = msgp.ReadMapKeyZC(bts)
 		if err != nil {
+			err = msgp.WrapError(err)
 			return
 		}
 		switch msgp.UnsafeString(field) {
 		case "foo":
 			z.Foo, bts, err = msgp.ReadStringBytes(bts)
 			if err != nil {
+				err = msgp.WrapError(err, "Foo")
 				return
 			}
 		case "hoge":
 			z.Hoge, bts, err = msgp.ReadStringBytes(bts)
 			if err != nil {
+				err = msgp.WrapError(err, "Hoge")
 				return
 			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
+				err = msgp.WrapError(err)
 				return
 			}
 		}

vendor/github.com/fluent/fluent-logger-golang/fluent/version.go

@@ -1,3 +1,3 @@
 package fluent
 
-const Version = "1.3.0"
+const Version = "1.4.0"

vendor/github.com/tinylib/msgp/msgp/edit.go

@@ -198,6 +198,7 @@ func resizeMap(raw []byte, delta int64) []byte {
 		if cap(raw)-len(raw) >= 2 {
 			raw = raw[0 : len(raw)+2]
 			copy(raw[5:], raw[3:])
+			raw[0] = mmap32
 			big.PutUint32(raw[1:], uint32(sz+delta))
 			return raw
 		}

vendor/github.com/tinylib/msgp/msgp/errors.go

@@ -5,6 +5,8 @@ import (
 	"reflect"
 )
 
+const resumableDefault = false
+
 var (
 	// ErrShortBytes is returned when the
 	// slice being decoded is too short to
@@ -26,84 +28,240 @@ type Error interface {
 	// Resumable returns whether
 	// or not the error means that
 	// the stream of data is malformed
-	// and  the information is unrecoverable.
+	// and the information is unrecoverable.
 	Resumable() bool
 }
 
+// contextError allows msgp Error instances to be enhanced with additional
+// context about their origin.
+type contextError interface {
+	Error
+
+	// withContext must not modify the error instance - it must clone and
+	// return a new error with the context added.
+	withContext(ctx string) error
+}
+
+// Cause returns the underlying cause of an error that has been wrapped
+// with additional context.
+func Cause(e error) error {
+	out := e
+	if e, ok := e.(errWrapped); ok && e.cause != nil {
+		out = e.cause
+	}
+	return out
+}
+
+// Resumable returns whether or not the error means that the stream of data is
+// malformed and the information is unrecoverable.
+func Resumable(e error) bool {
+	if e, ok := e.(Error); ok {
+		return e.Resumable()
+	}
+	return resumableDefault
+}
+
+// WrapError wraps an error with additional context that allows the part of the
+// serialized type that caused the problem to be identified. Underlying errors
+// can be retrieved using Cause()
+//
+// The input error is not modified - a new error should be returned.
+//
+// ErrShortBytes is not wrapped with any context due to backward compatibility
+// issues with the public API.
+//
+func WrapError(err error, ctx ...interface{}) error {
+	switch e := err.(type) {
+	case errShort:
+		return e
+	case contextError:
+		return e.withContext(ctxString(ctx))
+	default:
+		return errWrapped{cause: err, ctx: ctxString(ctx)}
+	}
+}
+
+// ctxString converts the incoming interface{} slice into a single string.
+func ctxString(ctx []interface{}) string {
+	out := ""
+	for idx, cv := range ctx {
+		if idx > 0 {
+			out += "/"
+		}
+		out += fmt.Sprintf("%v", cv)
+	}
+	return out
+}
+
+func addCtx(ctx, add string) string {
+	if ctx != "" {
+		return add + "/" + ctx
+	} else {
+		return add
+	}
+}
+
+// errWrapped allows arbitrary errors passed to WrapError to be enhanced with
+// context and unwrapped with Cause()
+type errWrapped struct {
+	cause error
+	ctx   string
+}
+
+func (e errWrapped) Error() string {
+	if e.ctx != "" {
+		return fmt.Sprintf("%s at %s", e.cause, e.ctx)
+	} else {
+		return e.cause.Error()
+	}
+}
+
+func (e errWrapped) Resumable() bool {
+	if e, ok := e.cause.(Error); ok {
+		return e.Resumable()
+	}
+	return resumableDefault
+}
+
 type errShort struct{}
 
 func (e errShort) Error() string   { return "msgp: too few bytes left to read object" }
 func (e errShort) Resumable() bool { return false }
 
-type errFatal struct{}
+type errFatal struct {
+	ctx string
+}
+
+func (f errFatal) Error() string {
+	out := "msgp: fatal decoding error (unreachable code)"
+	if f.ctx != "" {
+		out += " at " + f.ctx
+	}
+	return out
+}
 
-func (f errFatal) Error() string   { return "msgp: fatal decoding error (unreachable code)" }
 func (f errFatal) Resumable() bool { return false }
 
+func (f errFatal) withContext(ctx string) error { f.ctx = addCtx(f.ctx, ctx); return f }
+
 // ArrayError is an error returned
 // when decoding a fix-sized array
 // of the wrong size
 type ArrayError struct {
 	Wanted uint32
 	Got    uint32
+	ctx    string
 }
 
 // Error implements the error interface
 func (a ArrayError) Error() string {
-	return fmt.Sprintf("msgp: wanted array of size %d; got %d", a.Wanted, a.Got)
+	out := fmt.Sprintf("msgp: wanted array of size %d; got %d", a.Wanted, a.Got)
+	if a.ctx != "" {
+		out += " at " + a.ctx
+	}
+	return out
 }
 
 // Resumable is always 'true' for ArrayErrors
 func (a ArrayError) Resumable() bool { return true }
 
+func (a ArrayError) withContext(ctx string) error { a.ctx = addCtx(a.ctx, ctx); return a }
+
 // IntOverflow is returned when a call
 // would downcast an integer to a type
 // with too few bits to hold its value.
 type IntOverflow struct {
 	Value         int64 // the value of the integer
 	FailedBitsize int   // the bit size that the int64 could not fit into
+	ctx           string
 }
 
 // Error implements the error interface
 func (i IntOverflow) Error() string {
-	return fmt.Sprintf("msgp: %d overflows int%d", i.Value, i.FailedBitsize)
+	str := fmt.Sprintf("msgp: %d overflows int%d", i.Value, i.FailedBitsize)
+	if i.ctx != "" {
+		str += " at " + i.ctx
+	}
+	return str
 }
 
 // Resumable is always 'true' for overflows
 func (i IntOverflow) Resumable() bool { return true }
 
+func (i IntOverflow) withContext(ctx string) error { i.ctx = addCtx(i.ctx, ctx); return i }
+
 // UintOverflow is returned when a call
 // would downcast an unsigned integer to a type
 // with too few bits to hold its value
 type UintOverflow struct {
 	Value         uint64 // value of the uint
 	FailedBitsize int    // the bit size that couldn't fit the value
+	ctx           string
 }
 
 // Error implements the error interface
 func (u UintOverflow) Error() string {
-	return fmt.Sprintf("msgp: %d overflows uint%d", u.Value, u.FailedBitsize)
+	str := fmt.Sprintf("msgp: %d overflows uint%d", u.Value, u.FailedBitsize)
+	if u.ctx != "" {
+		str += " at " + u.ctx
+	}
+	return str
 }
 
 // Resumable is always 'true' for overflows
 func (u UintOverflow) Resumable() bool { return true }
 
+func (u UintOverflow) withContext(ctx string) error { u.ctx = addCtx(u.ctx, ctx); return u }
+
+// UintBelowZero is returned when a call
+// would cast a signed integer below zero
+// to an unsigned integer.
+type UintBelowZero struct {
+	Value int64 // value of the incoming int
+	ctx   string
+}
+
+// Error implements the error interface
+func (u UintBelowZero) Error() string {
+	str := fmt.Sprintf("msgp: attempted to cast int %d to unsigned", u.Value)
+	if u.ctx != "" {
+		str += " at " + u.ctx
+	}
+	return str
+}
+
+// Resumable is always 'true' for overflows
+func (u UintBelowZero) Resumable() bool { return true }
+
+func (u UintBelowZero) withContext(ctx string) error {
+	u.ctx = ctx
+	return u
+}
+
 // A TypeError is returned when a particular
 // decoding method is unsuitable for decoding
 // a particular MessagePack value.
 type TypeError struct {
 	Method  Type // Type expected by method
 	Encoded Type // Type actually encoded
+
+	ctx string
 }
 
 // Error implements the error interface
 func (t TypeError) Error() string {
-	return fmt.Sprintf("msgp: attempted to decode type %q with method for %q", t.Encoded, t.Method)
+	out := fmt.Sprintf("msgp: attempted to decode type %q with method for %q", t.Encoded, t.Method)
+	if t.ctx != "" {
+		out += " at " + t.ctx
+	}
+	return out
 }
 
 // Resumable returns 'true' for TypeErrors
 func (t TypeError) Resumable() bool { return true }
 
+func (t TypeError) withContext(ctx string) error { t.ctx = addCtx(t.ctx, ctx); return t }
+
 // returns either InvalidPrefixError or
 // TypeError depending on whether or not
 // the prefix is recognized
@@ -133,10 +291,24 @@ func (i InvalidPrefixError) Resumable() bool { return false }
 // to a function that takes `interface{}`.
 type ErrUnsupportedType struct {
 	T reflect.Type
+
+	ctx string
 }
 
 // Error implements error
-func (e *ErrUnsupportedType) Error() string { return fmt.Sprintf("msgp: type %q not supported", e.T) }
+func (e *ErrUnsupportedType) Error() string {
+	out := fmt.Sprintf("msgp: type %q not supported", e.T)
+	if e.ctx != "" {
+		out += " at " + e.ctx
+	}
+	return out
+}
 
 // Resumable returns 'true' for ErrUnsupportedType
 func (e *ErrUnsupportedType) Resumable() bool { return true }
+
+func (e *ErrUnsupportedType) withContext(ctx string) error {
+	o := *e
+	o.ctx = addCtx(o.ctx, ctx)
+	return &o
+}

vendor/github.com/tinylib/msgp/msgp/extension.go

@@ -445,26 +445,27 @@ func AppendExtension(b []byte, e Extension) ([]byte, error) {
 		o[n] = mfixext16
 		o[n+1] = byte(e.ExtensionType())
 		n += 2
-	}
-	switch {
-	case l < math.MaxUint8:
-		o, n = ensure(b, l+3)
-		o[n] = mext8
-		o[n+1] = byte(uint8(l))
-		o[n+2] = byte(e.ExtensionType())
-		n += 3
-	case l < math.MaxUint16:
-		o, n = ensure(b, l+4)
-		o[n] = mext16
-		big.PutUint16(o[n+1:], uint16(l))
-		o[n+3] = byte(e.ExtensionType())
-		n += 4
 	default:
-		o, n = ensure(b, l+6)
-		o[n] = mext32
-		big.PutUint32(o[n+1:], uint32(l))
-		o[n+5] = byte(e.ExtensionType())
-		n += 6
+		switch {
+		case l < math.MaxUint8:
+			o, n = ensure(b, l+3)
+			o[n] = mext8
+			o[n+1] = byte(uint8(l))
+			o[n+2] = byte(e.ExtensionType())
+			n += 3
+		case l < math.MaxUint16:
+			o, n = ensure(b, l+4)
+			o[n] = mext16
+			big.PutUint16(o[n+1:], uint16(l))
+			o[n+3] = byte(e.ExtensionType())
+			n += 4
+		default:
+			o, n = ensure(b, l+6)
+			o[n] = mext32
+			big.PutUint32(o[n+1:], uint32(l))
+			o[n+5] = byte(e.ExtensionType())
+			n += 6
+		}
 	}
 	return o, e.MarshalBinaryTo(o[n:])
 }

vendor/github.com/tinylib/msgp/msgp/appengine.go → vendor/github.com/tinylib/msgp/msgp/purego.go

@@ -1,4 +1,4 @@
-// +build appengine
+// +build purego appengine
 
 package msgp
 

vendor/github.com/tinylib/msgp/msgp/read.go

@@ -583,6 +583,14 @@ func (m *Reader) ReadInt64() (i int64, err error) {
 		i = int64(getMint8(p))
 		return
 
+	case muint8:
+		p, err = m.R.Next(2)
+		if err != nil {
+			return
+		}
+		i = int64(getMuint8(p))
+		return
+
 	case mint16:
 		p, err = m.R.Next(3)
 		if err != nil {
@@ -591,6 +599,14 @@ func (m *Reader) ReadInt64() (i int64, err error) {
 		i = int64(getMint16(p))
 		return
 
+	case muint16:
+		p, err = m.R.Next(3)
+		if err != nil {
+			return
+		}
+		i = int64(getMuint16(p))
+		return
+
 	case mint32:
 		p, err = m.R.Next(5)
 		if err != nil {
@@ -599,6 +615,14 @@ func (m *Reader) ReadInt64() (i int64, err error) {
 		i = int64(getMint32(p))
 		return
 
+	case muint32:
+		p, err = m.R.Next(5)
+		if err != nil {
+			return
+		}
+		i = int64(getMuint32(p))
+		return
+
 	case mint64:
 		p, err = m.R.Next(9)
 		if err != nil {
@@ -607,6 +631,19 @@ func (m *Reader) ReadInt64() (i int64, err error) {
 		i = getMint64(p)
 		return
 
+	case muint64:
+		p, err = m.R.Next(9)
+		if err != nil {
+			return
+		}
+		u := getMuint64(p)
+		if u > math.MaxInt64 {
+			err = UintOverflow{Value: u, FailedBitsize: 64}
+			return
+		}
+		i = int64(u)
+		return
+
 	default:
 		err = badPrefix(IntType, lead)
 		return
@@ -678,6 +715,19 @@ func (m *Reader) ReadUint64() (u uint64, err error) {
 		return
 	}
 	switch lead {
+	case mint8:
+		p, err = m.R.Next(2)
+		if err != nil {
+			return
+		}
+		v := int64(getMint8(p))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		return
+
 	case muint8:
 		p, err = m.R.Next(2)
 		if err != nil {
@@ -686,6 +736,19 @@ func (m *Reader) ReadUint64() (u uint64, err error) {
 		u = uint64(getMuint8(p))
 		return
 
+	case mint16:
+		p, err = m.R.Next(3)
+		if err != nil {
+			return
+		}
+		v := int64(getMint16(p))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		return
+
 	case muint16:
 		p, err = m.R.Next(3)
 		if err != nil {
@@ -694,6 +757,19 @@ func (m *Reader) ReadUint64() (u uint64, err error) {
 		u = uint64(getMuint16(p))
 		return
 
+	case mint32:
+		p, err = m.R.Next(5)
+		if err != nil {
+			return
+		}
+		v := int64(getMint32(p))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		return
+
 	case muint32:
 		p, err = m.R.Next(5)
 		if err != nil {
@@ -702,6 +778,19 @@ func (m *Reader) ReadUint64() (u uint64, err error) {
 		u = uint64(getMuint32(p))
 		return
 
+	case mint64:
+		p, err = m.R.Next(9)
+		if err != nil {
+			return
+		}
+		v := int64(getMint64(p))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		return
+
 	case muint64:
 		p, err = m.R.Next(9)
 		if err != nil {
@@ -711,7 +800,11 @@ func (m *Reader) ReadUint64() (u uint64, err error) {
 		return
 
 	default:
-		err = badPrefix(UintType, lead)
+		if isnfixint(lead) {
+			err = UintBelowZero{Value: int64(rnfixint(lead))}
+		} else {
+			err = badPrefix(UintType, lead)
+		}
 		return
 
 	}

vendor/github.com/tinylib/msgp/msgp/read_bytes.go

@@ -79,6 +79,9 @@ func (r *Raw) UnmarshalMsg(b []byte) ([]byte, error) {
 		return b, err
 	}
 	rlen := l - len(out)
+	if IsNil(b[:rlen]) {
+		rlen = 0
+	}
 	if cap(*r) < rlen {
 		*r = make(Raw, rlen)
 	} else {
@@ -104,7 +107,11 @@ func (r Raw) EncodeMsg(w *Writer) error {
 // next object on the wire.
 func (r *Raw) DecodeMsg(f *Reader) error {
 	*r = (*r)[:0]
-	return appendNext(f, (*[]byte)(r))
+	err := appendNext(f, (*[]byte)(r))
+	if IsNil(*r) {
+		*r = (*r)[:0]
+	}
+	return err
 }
 
 // Msgsize implements msgp.Sizer
@@ -368,6 +375,15 @@ func ReadInt64Bytes(b []byte) (i int64, o []byte, err error) {
 		o = b[2:]
 		return
 
+	case muint8:
+		if l < 2 {
+			err = ErrShortBytes
+			return
+		}
+		i = int64(getMuint8(b))
+		o = b[2:]
+		return
+
 	case mint16:
 		if l < 3 {
 			err = ErrShortBytes
@@ -377,6 +393,15 @@ func ReadInt64Bytes(b []byte) (i int64, o []byte, err error) {
 		o = b[3:]
 		return
 
+	case muint16:
+		if l < 3 {
+			err = ErrShortBytes
+			return
+		}
+		i = int64(getMuint16(b))
+		o = b[3:]
+		return
+
 	case mint32:
 		if l < 5 {
 			err = ErrShortBytes
@@ -386,12 +411,35 @@ func ReadInt64Bytes(b []byte) (i int64, o []byte, err error) {
 		o = b[5:]
 		return
 
+	case muint32:
+		if l < 5 {
+			err = ErrShortBytes
+			return
+		}
+		i = int64(getMuint32(b))
+		o = b[5:]
+		return
+
 	case mint64:
 		if l < 9 {
 			err = ErrShortBytes
 			return
 		}
-		i = getMint64(b)
+		i = int64(getMint64(b))
+		o = b[9:]
+		return
+
+	case muint64:
+		if l < 9 {
+			err = ErrShortBytes
+			return
+		}
+		u := getMuint64(b)
+		if u > math.MaxInt64 {
+			err = UintOverflow{Value: u, FailedBitsize: 64}
+			return
+		}
+		i = int64(u)
 		o = b[9:]
 		return
 
@@ -477,6 +525,20 @@ func ReadUint64Bytes(b []byte) (u uint64, o []byte, err error) {
 	}
 
 	switch lead {
+	case mint8:
+		if l < 2 {
+			err = ErrShortBytes
+			return
+		}
+		v := int64(getMint8(b))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		o = b[2:]
+		return
+
 	case muint8:
 		if l < 2 {
 			err = ErrShortBytes
@@ -486,6 +548,20 @@ func ReadUint64Bytes(b []byte) (u uint64, o []byte, err error) {
 		o = b[2:]
 		return
 
+	case mint16:
+		if l < 3 {
+			err = ErrShortBytes
+			return
+		}
+		v := int64(getMint16(b))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		o = b[3:]
+		return
+
 	case muint16:
 		if l < 3 {
 			err = ErrShortBytes
@@ -495,6 +571,20 @@ func ReadUint64Bytes(b []byte) (u uint64, o []byte, err error) {
 		o = b[3:]
 		return
 
+	case mint32:
+		if l < 5 {
+			err = ErrShortBytes
+			return
+		}
+		v := int64(getMint32(b))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		o = b[5:]
+		return
+
 	case muint32:
 		if l < 5 {
 			err = ErrShortBytes
@@ -504,6 +594,20 @@ func ReadUint64Bytes(b []byte) (u uint64, o []byte, err error) {
 		o = b[5:]
 		return
 
+	case mint64:
+		if l < 9 {
+			err = ErrShortBytes
+			return
+		}
+		v := int64(getMint64(b))
+		if v < 0 {
+			err = UintBelowZero{Value: v}
+			return
+		}
+		u = uint64(v)
+		o = b[9:]
+		return
+
 	case muint64:
 		if l < 9 {
 			err = ErrShortBytes
@@ -514,7 +618,11 @@ func ReadUint64Bytes(b []byte) (u uint64, o []byte, err error) {
 		return
 
 	default:
-		err = badPrefix(UintType, lead)
+		if isnfixint(lead) {
+			err = UintBelowZero{Value: int64(rnfixint(lead))}
+		} else {
+			err = badPrefix(UintType, lead)
+		}
 		return
 	}
 }

vendor/github.com/tinylib/msgp/msgp/unsafe.go

@@ -1,4 +1,4 @@
-// +build !appengine
+// +build !purego,!appengine
 
 package msgp
 

vendor/github.com/tinylib/msgp/msgp/write.go

@@ -685,7 +685,7 @@ func (mw *Writer) WriteIntf(v interface{}) error {
 	case reflect.Map:
 		return mw.writeMap(val)
 	}
-	return &ErrUnsupportedType{val.Type()}
+	return &ErrUnsupportedType{T: val.Type()}
 }
 
 func (mw *Writer) writeMap(v reflect.Value) (err error) {