Add support for RFC3339 time zone offsets in JSON output

libbeat/common/dtfmt/builder.go

@@ -207,7 +207,9 @@ func (b *builder) monthOfYearShortText() {
 	b.appendShortText(ftMonthOfYear)
 }
 
-// TODO: add timezone support
+func (b *builder) timeZoneOffsetText() {
+	b.appendText(ftTimeZoneOffset)
+}
 
 func (b *builder) appendRune(r rune) {
 	b.add(runeLiteral{r})

libbeat/common/dtfmt/ctx.go

@@ -33,16 +33,19 @@ type ctx struct {
 	hour, min, sec int
 	millis         int
 
+	tzOffset int
+
 	buf []byte
 }
 
 type ctxConfig struct {
-	date    bool
-	clock   bool
-	weekday bool
-	yearday bool
-	millis  bool
-	iso     bool
+	date     bool
+	clock    bool
+	weekday  bool
+	yearday  bool
+	millis   bool
+	iso      bool
+	tzOffset bool
 }
 
 func (c *ctx) initTime(config *ctxConfig, t time.Time) {
@@ -67,6 +70,10 @@ func (c *ctx) initTime(config *ctxConfig, t time.Time) {
 	if config.weekday {
 		c.weekday = t.Weekday()
 	}
+
+	if config.tzOffset {
+		_, c.tzOffset = t.Zone()
+	}
 }
 
 func (c *ctxConfig) enableDate() {
@@ -93,6 +100,10 @@ func (c *ctxConfig) enableISO() {
 	c.iso = true
 }
 
+func (c *ctxConfig) enableTimeZoneOffset() {
+	c.tzOffset = true
+}
+
 func isLeap(year int) bool {
 	return year%4 == 0 && (year%100 != 0 || year%400 == 0)
 }

libbeat/common/dtfmt/dtfmt_test.go

@@ -98,6 +98,11 @@ func TestFormat(t *testing.T) {
 		{mkDateTime(2017, 1, 2, 4, 6, 7, 123),
 			"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",
 			"2017-01-02T04:06:07.123Z"},
+
+		// beats timestamp
+		{mkDateTimeWithLocation(2017, 1, 2, 4, 6, 7, 123, time.FixedZone("PST", -8*60*60)),
+			"yyyy-MM-dd'T'HH:mm:ss.SSSz",
+			"2017-01-02T04:06:07.123-08:00"},
 	}
 
 	for i, test := range tests {
@@ -123,5 +128,9 @@ func mkTime(h, m, s, S int) time.Time {
 }
 
 func mkDateTime(y, M, d, h, m, s, S int) time.Time {
-	return time.Date(y, time.Month(M), d, h, m, s, S*1000000, time.UTC)
+	return mkDateTimeWithLocation(y, M, d, h, m, s, S, time.UTC)
+}
+
+func mkDateTimeWithLocation(y, M, d, h, m, s, S int, l *time.Location) time.Time {
+	return time.Date(y, time.Month(M), d, h, m, s, S*1000000, l)
 }

libbeat/common/dtfmt/elems.go

@@ -142,6 +142,8 @@ func (f textField) requires(c *ctxConfig) error {
 		c.enableDate()
 	case ftDayOfWeek:
 		c.enableWeekday()
+	case ftTimeZoneOffset:
+		c.enableTimeZoneOffset()
 	default:
 		return fmt.Errorf("time field %v not supported by text", f.ft)
 	}
@@ -162,6 +164,8 @@ func (f textField) estimateSize() int {
 			return 6
 		}
 		return 9 // max(month) = len(September)
+	case ftTimeZoneOffset:
+		return 6
 	default:
 		return 0
 	}

libbeat/common/dtfmt/fields.go

@@ -43,6 +43,7 @@ const (
 	ftSecondOfMinute
 	ftMillisOfDay
 	ftMillisOfSecond
+	ftTimeZoneOffset
 )
 
 func getIntField(ft fieldType, ctx *ctx, t time.Time) (int, error) {
@@ -125,11 +126,34 @@ func getTextField(ft fieldType, ctx *ctx, t time.Time) (string, error) {
 		return ctx.weekday.String(), nil
 	case ftMonthOfYear:
 		return ctx.month.String(), nil
+	case ftTimeZoneOffset:
+		return tzOffsetString(ctx)
 	default:
 		return "", errors.New("no text field")
 	}
 }
 
+func tzOffsetString(ctx *ctx) (string, error) {
+	buf := make([]byte, 6)
+
+	tzOffsetMinutes := ctx.tzOffset / 60 // convert to minutes
+	if tzOffsetMinutes >= 0 {
+		buf[0] = '+'
+	} else {
+		buf[0] = '-'
+		tzOffsetMinutes = -tzOffsetMinutes
+	}
+
+	tzOffsetHours := tzOffsetMinutes / 60
+	tzOffsetMinutes = tzOffsetMinutes % 60
+	buf[1] = byte(tzOffsetHours/10) + '0'
+	buf[2] = byte(tzOffsetHours%10) + '0'
+	buf[3] = ':'
+	buf[4] = byte(tzOffsetMinutes/10) + '0'
+	buf[5] = byte(tzOffsetMinutes%10) + '0'
+	return string(buf), nil
+}
+
 func getTextFieldShort(ft fieldType, ctx *ctx, t time.Time) (string, error) {
 	switch ft {
 	case ftHalfdayOfDay:

libbeat/common/dtfmt/fmt.go

@@ -211,6 +211,9 @@ func parsePatternTo(b *builder, pattern string) error {
 		case 'S': // fraction of second
 			b.millisOfSecond(tokLen)
 
+		case 'z': // timezone offset
+			b.timeZoneOffsetText()
+
 		case '\'': // literal
 			if tokLen == 1 {
 				b.appendRune(rune(tokText[0]))

libbeat/outputs/codec/common.go

@@ -25,15 +25,34 @@ import (
 	"github.com/elastic/go-structform"
 )
 
+// MakeTimestampEncoder creates encoder function that formats time
+// into RFC3339 representation with UTC timezone in the output.
 func MakeTimestampEncoder() func(*time.Time, structform.ExtVisitor) error {
-	formatter, err := dtfmt.NewFormatter("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+	return MakeUTCOrLocalTimestampEncoder(false)
+}
+
+// MakeUTCOrLocalTimestampEncoder creates encoder function that formats time into RFC3339 representation
+// with UTC or local timezone in the output (based on localTime boolean parameter).
+func MakeUTCOrLocalTimestampEncoder(localTime bool) func(*time.Time, structform.ExtVisitor) error {
+	var dtPattern string
+	if localTime {
+		dtPattern = "yyyy-MM-dd'T'HH:mm:ss.SSSz"
+	} else {
+		dtPattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
+	}
+
+	formatter, err := dtfmt.NewFormatter(dtPattern)
 	if err != nil {
 		panic(err)
 	}
 
 	buf := make([]byte, 0, formatter.EstimateSize())
 	return func(t *time.Time, v structform.ExtVisitor) error {
-		tmp, err := formatter.AppendTo(buf, (*t).UTC())
+		outTime := *t
+		if !localTime {
+			outTime = outTime.UTC()
+		}
+		tmp, err := formatter.AppendTo(buf, outTime)
 		if err != nil {
 			return err
 		}
@@ -43,6 +62,8 @@ func MakeTimestampEncoder() func(*time.Time, structform.ExtVisitor) error {
 	}
 }
 
+// MakeBCTimestampEncoder creates encoder function that formats beats common time
+// into RFC3339 representation with UTC timezone in the output.
 func MakeBCTimestampEncoder() func(*common.Time, structform.ExtVisitor) error {
 	enc := MakeTimestampEncoder()
 	return func(t *common.Time, v structform.ExtVisitor) error {

libbeat/outputs/codec/json/json.go

@@ -41,11 +41,13 @@ type Encoder struct {
 type Config struct {
 	Pretty     bool
 	EscapeHTML bool
+	LocalTime  bool
 }
 
 var defaultConfig = Config{
 	Pretty:     false,
 	EscapeHTML: false,
+	LocalTime:  false,
 }
 
 func init() {
@@ -77,7 +79,7 @@ func (e *Encoder) reset() {
 	// create new encoder with custom time.Time encoding
 	e.folder, err = gotype.NewIterator(visitor,
 		gotype.Folders(
-			codec.MakeTimestampEncoder(),
+			codec.MakeUTCOrLocalTimestampEncoder(e.config.LocalTime),
 			codec.MakeBCTimestampEncoder(),
 		),
 	)

libbeat/outputs/codec/json/json_bench_test.go

@@ -0,0 +1,60 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package json
+
+import (
+	"testing"
+	"time"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+var result []byte
+
+func BenchmarkUTCTime(b *testing.B) {
+	var r []byte
+	codec := New("1.2.3", Config{})
+	fields := common.MapStr{"msg": "message"}
+	var t time.Time
+	var d time.Duration = 1000000000
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		t = t.Add(d)
+		r, _ = codec.Encode("test", &beat.Event{Fields: fields, Timestamp: t})
+	}
+	result = r
+}
+
+func BenchmarkLocalTime(b *testing.B) {
+	var r []byte
+	codec := New("1.2.3", Config{LocalTime: true})
+	fields := common.MapStr{"msg": "message"}
+	var t time.Time
+	var d time.Duration = 1000000000
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		t = t.Add(d)
+		r, _ = codec.Encode("test", &beat.Event{Fields: fields, Timestamp: t})
+	}
+	result = r
+}

libbeat/outputs/codec/json/json_test.go

@@ -19,6 +19,7 @@ package json
 
 import (
 	"testing"
+	"time"
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
@@ -27,6 +28,7 @@ import (
 func TestJsonCodec(t *testing.T) {
 	type testCase struct {
 		config   Config
+		ts       time.Time
 		in       common.MapStr
 		expected string
 	}
@@ -60,14 +62,25 @@ func TestJsonCodec(t *testing.T) {
 			in:       common.MapStr{"msg": "<hello>world</hello>"},
 			expected: `{"@timestamp":"0001-01-01T00:00:00.000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"<hello>world</hello>"}`,
 		},
+		"UTC timezone offset": testCase{
+			config:   Config{LocalTime: true},
+			in:       common.MapStr{"msg": "message"},
+			expected: `{"@timestamp":"0001-01-01T00:00:00.000+00:00","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
+		},
+		"PST timezone offset": testCase{
+			config:   Config{LocalTime: true},
+			ts:       time.Time{}.In(time.FixedZone("PST", -8*60*60)),
+			in:       common.MapStr{"msg": "message"},
+			expected: `{"@timestamp":"0000-12-31T16:00:00.000-08:00","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
+		},
 	}
 
 	for name, test := range cases {
-		cfg, fields, expected := test.config, test.in, test.expected
+		cfg, ts, fields, expected := test.config, test.ts, test.in, test.expected
 
 		t.Run(name, func(t *testing.T) {
 			codec := New("1.2.3", cfg)
-			actual, err := codec.Encode("test", &beat.Event{Fields: fields})
+			actual, err := codec.Encode("test", &beat.Event{Fields: fields, Timestamp: ts})
 
 			if err != nil {
 				t.Errorf("Error during event write %v", err)