feat: use now as time parser (#2057)

* feat: use now as time parser

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

* add formats: JSISO, ISO8601

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

* fix unit test cases

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

---------

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/gorilla/handlers v1.5.1
 	github.com/gorilla/mux v1.8.0
 	github.com/jhump/protoreflect v1.15.0
+	github.com/jinzhu/now v1.1.5
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1
 	github.com/klauspost/compress v1.16.4
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible

go.sum

@@ -116,6 +116,8 @@ github.com/imdario/mergo v0.3.13 h1:lFzP57bqS/wsqKssCGmtLAb8A0wKjLGrve2q3PPVcBk=
 github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
 github.com/jhump/protoreflect v1.15.0 h1:U5T5/2LF0AZQFP9T4W5GfBjBaTruomrKobiR4E+oA/Q=
 github.com/jhump/protoreflect v1.15.0/go.mod h1:qww51KYjD2hoCl/ohxw5cK2LSssFczrbO1t8Ld2TENs=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
 github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
@@ -186,6 +188,7 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
 github.com/prometheus/client_golang v1.14.0 h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=
 github.com/prometheus/client_golang v1.14.0/go.mod h1:8vpkKitgIVNcqrRBWh1C4TIUQgYNtG/XQE4E/Zae36Y=
 github.com/prometheus/client_model v0.3.0 h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4=

internal/topo/operator/field_processor.go

@@ -129,9 +129,5 @@ func (p *defaultFieldProcessor) validateAndConvertField(sf *ast.JsonStreamField,
 }
 
 func (p *defaultFieldProcessor) parseTime(s string) (time.Time, error) {
-	if p.timestampFormat != "" {
-		return cast.ParseTime(s, p.timestampFormat)
-	} else {
-		return time.Parse(cast.JSISO, s)
-	}
+	return cast.ParseTime(s, p.timestampFormat)
 }

internal/topo/operator/preprocessor_test.go

@@ -613,7 +613,7 @@ func TestPreprocessorTime_Apply(t *testing.T) {
 				},
 			},
 			data:   []byte(`{"abc": "2019-09-19T00:55:1dd5Z", "def" : 111568854573431}`),
-			result: errors.New("error in preprocessor: field abc type mismatch: parsing time \"2019-09-19T00:55:1dd5Z\" as \"2006-01-02T15:04:05.000Z07:00\": cannot parse \"1dd5Z\" as \"05\""),
+			result: errors.New("error in preprocessor: field abc type mismatch: Can't parse string as time: 2019-09-19T00:55:1dd5Z"),
 		},
 		{ // 3
 			stmt: &ast.StreamStmt{
@@ -868,7 +868,7 @@ func TestPreprocessorEventtime_Apply(t *testing.T) {
 				},
 			},
 			data:   []byte(`{"abc": 34, "def" : "2019-09-23AT02:47:29", "ghi": 50}`),
-			result: errors.New("cannot convert timestamp field def to timestamp with error parsing time \"2019-09-23AT02:47:29\" as \"2006-01-02PM15:04:05\": cannot parse \"AT02:47:29\" as \"PM\""),
+			result: errors.New("cannot convert timestamp field def to timestamp with error Can't parse string as time: 2019-09-23AT02:47:29"),
 		},
 	}
 

pkg/cast/time.go

@@ -17,6 +17,8 @@ package cast
 import (
 	"fmt"
 	"time"
+
+	"github.com/jinzhu/now"
 )
 
 //var (
@@ -64,6 +66,10 @@ const (
 	ISO8601 = "2006-01-02T15:04:05"
 )
 
+func init() {
+	now.TimeFormats = append(now.TimeFormats, JSISO, ISO8601)
+}
+
 func TimeToUnixMilli(time time.Time) int64 {
 	return time.UnixNano() / 1e6
 }
@@ -79,20 +85,8 @@ func InterfaceToUnixMilli(i interface{}, format string) (int64, error) {
 	case time.Time:
 		return TimeToUnixMilli(t), nil
 	case string:
-		var ti time.Time
-		var err error
-		f := JSISO
-		if format != "" {
-			f, err = convertFormat(format)
-			if err != nil {
-				return 0, err
-			}
-		}
-		ti, err = time.Parse(f, t)
-		if err != nil {
-			return 0, err
-		}
-		return TimeToUnixMilli(ti), nil
+		ti, err := ParseTime(t, format)
+		return TimeToUnixMilli(ti), err
 	default:
 		return 0, fmt.Errorf("unsupported type to convert to timestamp %v", t)
 	}
@@ -109,20 +103,7 @@ func InterfaceToTime(i interface{}, format string) (time.Time, error) {
 	case time.Time:
 		return t, nil
 	case string:
-		var ti time.Time
-		var err error
-		f := JSISO
-		if format != "" {
-			f, err = convertFormat(format)
-			if err != nil {
-				return ti, err
-			}
-		}
-		ti, err = time.Parse(f, t)
-		if err != nil {
-			return ti, err
-		}
-		return ti, nil
+		return ParseTime(t, format)
 	default:
 		return time.Now(), fmt.Errorf("unsupported type to convert to timestamp %v", t)
 	}
@@ -132,12 +113,18 @@ func TimeFromUnixMilli(t int64) time.Time {
 	return time.Unix(t/1000, (t%1000)*1e6).UTC()
 }
 
-func ParseTime(t string, f string) (time.Time, error) {
-	if f, err := convertFormat(f); err != nil {
-		return time.Now(), err
-	} else {
-		return time.Parse(f, t)
+func ParseTime(t string, f string) (_ time.Time, err error) {
+	if f, err = convertFormat(f); err != nil {
+		return time.Time{}, err
+	}
+	c := &now.Config{
+		TimeLocation: time.UTC,
+		TimeFormats:  now.TimeFormats,
+	}
+	if f != "" {
+		c.TimeFormats = []string{f}
 	}
+	return c.Parse(t)
 }
 
 func FormatTime(time time.Time, f string) (string, error) {

pkg/cast/time_test.go

@@ -94,6 +94,12 @@ func TestParseTime(t *testing.T) {
 			false,
 		},
 		{
+			time.Date(2020, time.January, 16, 2, 14, 24, 0, time.UTC),
+			"2020-01-16 02:14:24",
+			"",
+			false,
+		},
+		{
 			time.Time{},
 			"2020",
 			"YYY",