Improve timestamp handling (#1889)

* Update testcases to test timestamps with timezone specified

This is the default @timestamp format used by kibana json log for example

* Make timestamp conversation based on RFC3339

Allow timestamps without timezone. See #1831 for more details

filebeat/input/event.go

@@ -69,14 +69,15 @@ func mergeJSONFields(f *FileEvent, event common.MapStr) {
 						event[jsonErrorKey] = "@timestamp not overwritten (not string)"
 						continue
 					}
-					// @timestamp must be of time common.Time
-					ts, err := common.ParseTime(vstr)
+
+					// @timestamp must be of format RFC3339
+					ts, err := time.Parse(time.RFC3339, vstr)
 					if err != nil {
 						logp.Err("JSON: Won't overwrite @timestamp because of parsing error: %v", err)
 						event[jsonErrorKey] = fmt.Sprintf("@timestamp not overwritten (parse error on %s)", vstr)
 						continue
 					}
-					event[k] = ts
+					event[k] = common.Time(ts)
 				} else if k == "type" {
 					vstr, ok := v.(string)
 					if !ok {

filebeat/tests/files/logs/json_timestamp.log

@@ -1,3 +1,5 @@
-{"@timestamp":"2016-04-05T18:47:18.444Z","level":"INFO","logger":"iapi.logger","thread":"JobCourier4","appInfo":{"appname":"SessionManager","appid":"Pooler","host":"demohost.mydomain.com","ip":"192.168.128.113","pid":13982},"userFields":{"ApplicationId":"PROFAPP_001","RequestTrackingId":"RetrieveTBProfileToken-6066477"},"source":"DataAccess\/FetchActiveSessionToken.process","msg":"FetchActiveSessionToken process ended"}
-{"@timestamp":"invalid","level":"INFO","logger":"iapi.logger","thread":"JobCourier4","appInfo":{"appname":"SessionManager","appid":"Pooler","host":"demohost.mydomain.com","ip":"192.168.128.113","pid":13982},"userFields":{"ApplicationId":"PROFAPP_001","RequestTrackingId":"RetrieveTBProfileToken-6066477"},"source":"DataAccess\/FetchActiveSessionToken.process","msg":"FetchActiveSessionToken process ended"}
-{"@timestamp":{"hello": "test"},"level":"INFO","logger":"iapi.logger","thread":"JobCourier4","appInfo":{"appname":"SessionManager","appid":"Pooler","host":"demohost.mydomain.com","ip":"192.168.128.113","pid":13982},"userFields":{"ApplicationId":"PROFAPP_001","RequestTrackingId":"RetrieveTBProfileToken-6066477"},"source":"DataAccess\/FetchActiveSessionToken.process","msg":"FetchActiveSessionToken process ended"}
+{"@timestamp":"2016-04-05T18:47:18.444Z"}
+{"@timestamp":"invalid"}
+{"@timestamp":{"hello": "test"}}
+{"@timestamp":"2016-04-05T18:47:18.444+00:00"}
+{"@timestamp":"2016-04-05T18:47:18+00:00"}

filebeat/tests/system/test_json.py

@@ -188,12 +188,12 @@ def test_timestamp_in_message(self):
 
         proc = self.start_beat()
         self.wait_until(
-            lambda: self.output_has(lines=3),
+            lambda: self.output_has(lines=5),
             max_timeout=10)
         proc.check_kill_and_wait()
 
         output = self.read_output()
-        assert len(output) == 3
+        assert len(output) == 5
         assert all(isinstance(o["@timestamp"], basestring) for o in output)
         assert all(isinstance(o["type"], basestring) for o in output)
         assert output[0]["@timestamp"] == "2016-04-05T18:47:18.444Z"
@@ -205,6 +205,12 @@ def test_timestamp_in_message(self):
         assert output[2]["json_error"] == \
             "@timestamp not overwritten (not string)"
 
+        assert "json_error" not in output[3]
+        assert output[3]["@timestamp"] == "2016-04-05T18:47:18.444Z", output[3]["@timestamp"]
+
+        assert "json_error" not in output[4]
+        assert output[4]["@timestamp"] == "2016-04-05T18:47:18.000Z", output[4]["@timestamp"]
+
     def test_type_in_message(self):
         """
         If overwrite_keys is true and type is in the message, we have to