Merge pull request #49 from jwolski2/eventtime

Define EventTime for marshalling nanosecond-precision timestamps over-the-wire

Author: tagomoris

fluent/fluent.go

@@ -23,6 +23,9 @@ const (
 	defaultRetryWait              = 500
 	defaultMaxRetry               = 13
 	defaultReconnectWaitIncreRate = 1.5
+	// Default sub-second precision value to false since it is only compatible
+	// with fluentd versions v0.14 and above.
+	defaultSubSecondPrecision = false
 )
 
 type Config struct {
@@ -38,6 +41,10 @@ type Config struct {
 	TagPrefix        string        `json:"tag_prefix"`
 	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"`
 }
 
 type Fluent struct {
@@ -207,6 +214,9 @@ func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (data
 		msg := Message{Tag: tag, Time: timeUnix, Record: message}
 		chunk := &MessageChunk{message: 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)
 	} else {
 		msg := &Message{Tag: tag, Time: timeUnix, Record: message}
 		data, err = msg.MarshalMsg(nil)

fluent/fluent_test.go

@@ -187,6 +187,33 @@ func Test_MarshalAsMsgpack(t *testing.T) {
 	}
 }
 
+func Test_SubSecondPrecision(t *testing.T) {
+	// Setup the test subject
+	fluent := &Fluent{
+		Config: Config{
+			SubSecondPrecision: true,
+		},
+		reconnecting: false,
+	}
+	fluent.conn = &Conn{}
+
+	// Exercise the test subject
+	timestamp := time.Unix(1267867237, 256)
+	encodedData, err := fluent.EncodeData("tag", timestamp, map[string]string{
+		"foo": "bar",
+	})
+
+	// Assert no encoding errors and that the timestamp has been encoded into
+	// the message as expected.
+	if err != nil {
+		t.Error(err)
+	}
+
+	expected := "\x94\xA3tag\xC7\x08\x00K\x92\u001Ee\x00\x00\x01\x00\x81\xA3foo\xA3bar\xC0"
+	actual := string(encodedData)
+	assert.Equal(t, expected, actual)
+}
+
 func Test_MarshalAsJSON(t *testing.T) {
 	f := &Fluent{Config: Config{MarshalAsJSON: true}, reconnecting: false}
 

fluent/proto.go

@@ -2,6 +2,12 @@
 
 package fluent
 
+import (
+	"time"
+
+	"github.com/tinylib/msgp/msgp"
+)
+
 //msgp:tuple Entry
 type Entry struct {
 	Time   int64       `msg:"time"`
@@ -22,3 +28,69 @@ type Message struct {
 	Record interface{} `msg:"record"`
 	Option interface{} `msg:"option"`
 }
+
+//msgp:tuple MessageExt
+type MessageExt struct {
+	Tag    string      `msg:"tag"`
+	Time   EventTime   `msg:"time,extension"`
+	Record interface{} `msg:"record"`
+	Option interface{} `msg:"option"`
+}
+
+// EventTime is an extension to the serialized time value. It builds in support
+// for sub-second (nanosecond) precision in serialized timestamps.
+//
+// You can find the full specification for the msgpack message payload here:
+// https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1.
+//
+// You can find more information on msgpack extension types here:
+// https://github.com/tinylib/msgp/wiki/Using-Extensions.
+type EventTime time.Time
+
+const (
+	extensionType = 0
+	length        = 8
+)
+
+func init() {
+	msgp.RegisterExtension(extensionType, func() msgp.Extension { return new(EventTime) })
+}
+
+func (t *EventTime) ExtensionType() int8 { return extensionType }
+
+func (t *EventTime) Len() int { return length }
+
+func (t *EventTime) MarshalBinaryTo(b []byte) error {
+	// Unwrap to Golang time
+	goTime := time.Time(*t)
+
+	// There's no support for timezones in fluentd's protocol for EventTime.
+	// Convert to UTC.
+	utc := goTime.UTC()
+
+	// Warning! Converting seconds to an int32 is a lossy operation. This code
+	// will hit the "Year 2038" problem.
+	sec := int32(utc.Unix())
+	nsec := utc.Nanosecond()
+
+	// Fill the buffer with 4 bytes for the second component of the timestamp.
+	b[0] = byte(sec >> 24)
+	b[1] = byte(sec >> 16)
+	b[2] = byte(sec >> 8)
+	b[3] = byte(sec)
+
+	// Fill the buffer with 4 bytes for the nanosecond component of the
+	// timestamp.
+	b[4] = byte(nsec >> 24)
+	b[5] = byte(nsec >> 16)
+	b[6] = byte(nsec >> 8)
+	b[7] = byte(nsec)
+
+	return nil
+}
+
+// UnmarshalBinary is not implemented since decoding messages is not supported
+// by this library.
+func (t *EventTime) UnmarshalBinary(b []byte) error {
+	return nil
+}