[WIP] Minor improvements and test fixes for timestamp nanosecond support

libbeat/common/datetime.go

@@ -21,37 +21,38 @@ import (
  "encoding/binary"
  "encoding/json"
  "errors"
- "fmt"
  "hash"
  "time"
+
+ "github.com/elastic/beats/libbeat/common/dtfmt"
 )
 
 const (
  // TsLayout is the seconds layout to be used in the timestamp marshaling/unmarshaling everywhere.
  // The timezone must always be UTC.
- TsLayout = "2006-01-02T15:04:05"
+ TsLayout = "2006-01-02T15:04:05.000Z"
+
+ tsLayoutMillis = "2006-01-02T15:04:05.000Z"
+ tsLayoutMicros = "2006-01-02T15:04:05.000000Z"
+ tsLayoutNanos = "2006-01-02T15:04:05.000000000Z"
 )
 
 // Time is an abstraction for the time.Time type
 type Time time.Time
 
-func (t Time) generateTsLayout() string {
- nanoTime := time.Time(t).UTC().UnixNano()
- trailZero := "000000000"
- for i := 0; i < 2; i++ {
- if nanoTime%1000 != 0 {
- break
- }
- trailZero = trailZero[:len(trailZero)-3]
- nanoTime = nanoTime / 1000
- }
- return fmt.Sprintf("%s.%sZ", TsLayout, trailZero)
+var defaultTimeFormatter = dtfmt.MustNewFormatter("yyyy-MM-dd'T'HH:mm:ss.fffffffff'Z'")
+
+var defaultParseFormats = []string{
+ tsLayoutMillis,
+ tsLayoutMicros,
+ tsLayoutNanos,
 }
 
 // MarshalJSON implements json.Marshaler interface.
 // The time is a quoted string in the JsTsLayout format.
 func (t Time) MarshalJSON() ([]byte, error) {
- return json.Marshal(time.Time(t).UTC().Format(t.generateTsLayout()))
+ str, _ := defaultTimeFormatter.Format(time.Time(t).UTC())
+ return json.Marshal(str)
 }
 
 // UnmarshalJSON implements js.Unmarshaler interface.
@@ -72,26 +73,22 @@ func (t Time) Hash32(h hash.Hash32) error {
 
 // ParseTime parses a time in the NanoTsLayout format first, then use millisTsLayout format
 func ParseTime(timespec string) (Time, error) {
- var (
- t time.Time
- err error
- tsLayout string
- trailZero string
- )
-
- for i := 0; i < 3; i++ {
- trailZero += "000"
- tsLayout = fmt.Sprintf("%s.%sZ", TsLayout, trailZero)
- t, err = time.Parse(tsLayout, timespec)
+ var err error
+ var t time.Time
+
+ for _, layout := range defaultParseFormats {
+ t, err = time.Parse(layout, timespec)
  if err == nil {
  break
  }
  }
+
  return Time(t), err
 }
 
 func (t Time) String() string {
- return time.Time(t).Format(t.generateTsLayout())
+ str, _ := defaultTimeFormatter.Format(time.Time(t))
+ return str
 }
 
 // MustParseTime is a convenience equivalent of the ParseTime function

libbeat/common/dtfmt/fmt.go

@@ -55,6 +55,16 @@ func releaseCtx(c *ctx) {
  ctxPool.Put(c)
 }
 
+// MustNewFormatter creates a new time formatter based on the provided pattern.
+// The functions panics if the pattern is invalid
+func MustNewFormatter(pattern string) *Formatter {
+ f, err := NewFormatter(pattern)
+ if err != nil {
+ panic(err)
+ }
+ return f
+}
+
 // NewFormatter creates a new time formatter based on provided pattern.
 // If pattern is invalid an error is returned.
 func NewFormatter(pattern string) (*Formatter, error) {

libbeat/outputs/codec/json/json_test.go

@@ -37,13 +37,13 @@ func TestJsonCodec(t *testing.T) {
  "default json": testCase{
  config: defaultConfig,
  in: common.MapStr{"msg": "message"},
- expected: `{"@timestamp":"0001-01-01T00:00:00.000000000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
+ expected: `{"@timestamp":"0001-01-01T00:00:00.000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
  },
  "pretty enabled": testCase{
  config: Config{Pretty: true},
  in: common.MapStr{"msg": "message"},
  expected: `{
- "@timestamp": "0001-01-01T00:00:00.000000000Z",
+ "@timestamp": "0001-01-01T00:00:00.000Z",
  "@metadata": {
  "beat": "test",
  "type": "_doc",
@@ -55,23 +55,23 @@ func TestJsonCodec(t *testing.T) {
  "html escaping enabled": testCase{
  config: Config{EscapeHTML: true},
  in: common.MapStr{"msg": "<hello>world</hello>"},
- expected: `{"@timestamp":"0001-01-01T00:00:00.000000000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"\u003chello\u003eworld\u003c/hello\u003e"}`,
+ expected: `{"@timestamp":"0001-01-01T00:00:00.000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"\u003chello\u003eworld\u003c/hello\u003e"}`,
  },
  "html escaping disabled": testCase{
  config: Config{EscapeHTML: false},
  in: common.MapStr{"msg": "<hello>world</hello>"},
- expected: `{"@timestamp":"0001-01-01T00:00:00.000000000Z","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"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.000000000+00:00","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"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.000000000-08:00","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
+ expected: `{"@timestamp":"0000-12-31T16:00:00.000-08:00","@metadata":{"beat":"test","type":"_doc","version":"1.2.3"},"msg":"message"}`,
  },
  }
 

libbeat/outputs/console/console_test.go

@@ -85,7 +85,7 @@ func TestConsoleOutput(t *testing.T) {
  []beat.Event{
  {Fields: event("field", "value")},
  },
- "{\"@timestamp\":\"0001-01-01T00:00:00.000000000Z\",\"@metadata\":{\"beat\":\"test\",\"type\":\"_doc\",\"version\":\"1.2.3\"},\"field\":\"value\"}\n",
+ "{\"@timestamp\":\"0001-01-01T00:00:00.000Z\",\"@metadata\":{\"beat\":\"test\",\"type\":\"_doc\",\"version\":\"1.2.3\"},\"field\":\"value\"}\n",
  },
  {
  "single json event (pretty=true)",
@@ -96,7 +96,7 @@ func TestConsoleOutput(t *testing.T) {
  []beat.Event{
  {Fields: event("field", "value")},
  },
- "{\n \"@timestamp\": \"0001-01-01T00:00:00.000000000Z\",\n \"@metadata\": {\n \"beat\": \"test\",\n \"type\": \"_doc\",\n \"version\": \"1.2.3\"\n },\n \"field\": \"value\"\n}\n",
+ "{\n \"@timestamp\": \"0001-01-01T00:00:00.000Z\",\n \"@metadata\": {\n \"beat\": \"test\",\n \"type\": \"_doc\",\n \"version\": \"1.2.3\"\n },\n \"field\": \"value\"\n}\n",
  },
  // TODO: enable test after update fmtstr support to beat.Event
  {

libbeat/outputs/logstash/logstash_integration_test.go

@@ -536,7 +536,24 @@ func testLogstashElasticOutputPluginBulkCompatibleMessage(t *testing.T, name str
 func checkEvent(t *testing.T, ls, es map[string]interface{}) {
  lsEvent := ls["_source"].(map[string]interface{})
  esEvent := es["_source"].(map[string]interface{})
- commonFields := []string{"@timestamp", "host", "type", "message"}
+
+ mustParseTs := func(spec string) time.Time {
+ ts, err := common.ParseTime(spec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return time.Time(ts)
+ }
+
+ // XXX: Logstash only support millsecond precisions. We assume that is still the case
+ // and round esTimestamp to millseconds as well
+ lsTimestamp := mustParseTs(lsEvent["@timestamp"].(string))
+ esTimestamp := mustParseTs(esEvent["@timestamp"].(string))
+ nanos := time.Duration(esTimestamp.Nanosecond())
+ esTimestamp = esTimestamp.Add(-(nanos % time.Millisecond))
+ assert.Equal(t, lsTimestamp, esTimestamp)
+
+ commonFields := []string{"host", "type", "message"}
  for _, field := range commonFields {
  assert.NotNil(t, lsEvent[field])
  assert.NotNil(t, esEvent[field])

packetbeat/tests/system/test_0032_dns.py

@@ -19,15 +19,16 @@ def test_A(self):
  objs = self.read_output()
  assert len(objs) == 1
  o = objs[0]
+ print(o)
 
  assert o["type"] == "dns"
  assert o["network.type"] == "ipv4"
  assert o["network.protocol"] == "dns"
  assert o["network.transport"] == "udp"
  assert o["network.bytes"] == 312
  assert "network.community_id" in o
- assert o["event.start"] == "2015-08-27T08:00:55.638Z"
- assert o["event.end"] == "2015-08-27T08:00:55.700Z"
+ assert o["event.start"] == "2015-08-27T08:00:55.638957Z"
+ assert o["event.end"] == "2015-08-27T08:00:55.700739Z"
  assert o["event.duration"] == 61782000
  assert o["client.ip"] == "192.168.238.68"
  assert o["source.ip"] == "192.168.238.68"

packetbeat/tests/system/test_0050_icmp.py

@@ -10,10 +10,15 @@ def test_2_pings(self):
 
  assert len(objs) == 2
  assert all([o["icmp.version"] == 4 for o in objs])
- assert objs[0]["@timestamp"] == "2015-10-19T21:47:49.900Z"
+
+ print(objs[0])
+ assert objs[0]["@timestamp"] == "2015-10-19T21:47:49.900657Z"
  assert objs[0]["event.duration"] == 12152000
- assert objs[1]["@timestamp"] == "2015-10-19T21:47:49.924Z"
+
+ print(objs[1])
+ assert objs[1]["@timestamp"] == "2015-10-19T21:47:49.924909Z"
  assert objs[1]["event.duration"] == 11935000
+
  self.assert_common_fields(objs)
  self.assert_common_icmp4_fields(objs[0])
  self.assert_common_icmp4_fields(objs[1])
@@ -24,8 +29,9 @@ def test_icmp4_ping(self):
  objs = self.read_output()
 
  assert len(objs) == 1
+ print(objs[0])
  assert objs[0]["icmp.version"] == 4
- assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.817Z"
+ assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.817185Z"
  assert objs[0]["event.duration"] == 20130000
  self.assert_common_fields(objs)
  self.assert_common_icmp4_fields(objs[0])
@@ -36,8 +42,9 @@ def test_icmp4_ping_over_vlan(self):
  objs = self.read_output()
 
  assert len(objs) == 1
+ print(objs[0])
  assert objs[0]["icmp.version"] == 4
- assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.849Z"
+ assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.849085Z"
  assert objs[0]["event.duration"] == 12192000
  self.assert_common_fields(objs)
  self.assert_common_icmp4_fields(objs[0])
@@ -48,8 +55,9 @@ def test_icmp6_ping(self):
  objs = self.read_output()
 
  assert len(objs) == 1
+ print(objs[0])
  assert objs[0]["icmp.version"] == 6
- assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.872Z"
+ assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.872952Z"
  assert objs[0]["event.duration"] == 16439000
  self.assert_common_fields(objs)
  self.assert_common_icmp6_fields(objs[0])
@@ -60,8 +68,9 @@ def test_icmp6_ping_over_vlan(self):
  objs = self.read_output()
 
  assert len(objs) == 1
+ print(objs[0])
  assert objs[0]["icmp.version"] == 6
- assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.901Z"
+ assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.901349Z"
  assert objs[0]["event.duration"] == 12333000
  self.assert_common_fields(objs)
  self.assert_common_icmp6_fields(objs[0])

packetbeat/tests/system/test_0060_flows.py

@@ -1,6 +1,6 @@
 from packetbeat import (BaseTest, FLOWS_REQUIRED_FIELDS)
 from pprint import PrettyPrinter
-from datetime import datetime
+from datetime import datetime, timedelta
 import six
 import os
 
@@ -14,7 +14,14 @@ def check_fields(flow, fields):
 
 
 def parse_timestamp(ts):
- return datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%fZ")
+ if ts[-1] != 'Z':
+ raise Exception("missing time zone marker Z")
+ ts = ts[:-1]
+ parts = ts.split(".")
+ ts = datetime.strptime(parts[0], "%Y-%m-%dT%H:%M:%S")
+ if len(parts[1]) > 6:
+ parts[1] = parts[1][:6] # ensure we always parse microseconds
+ return ts + timedelta(microseconds=datetime.strptime(parts[1], "%f").microsecond)
 
 
 class Test(BaseTest):

packetbeat/tests/system/test_0062_cassandra.py

@@ -21,12 +21,13 @@ def test_create_keyspace(self):
  self.run_packetbeat(pcap="cassandra/v4/cassandra_create_keyspace.pcap", debug_selectors=["*"])
  objs = self.read_output()
  o = objs[0]
+ print(o)
 
  assert o["type"] == "cassandra"
  assert o["event.dataset"] == "cassandra"
  assert o["event.duration"] == 62453000
  assert o["event.start"] == o["@timestamp"]
- assert o["event.end"] == "2016-06-28T09:03:53.502Z"
+ assert o["event.end"] == "2016-06-28T09:03:53.502299Z"
  assert o["client.ip"] == "127.0.0.1"
  assert o["client.port"] == 52749
  assert o["client.bytes"] == 133