Repository: numaproj/numaflow (go)

Issue:
Support extracting event time from an epoch value from payload
# Summary

Our builtin event time extractor uses [time.Parse](https://pkg.go.dev/time#Parse) to convert a time string representation to a time.Time object. Currently, the `time.Parse` only supports converting the time string using a user-provided format string, e.g. `2006-01-02T15:04:05Z07:00`. It doesn't support converting an epoch to time.

A common use case is that the event time extracted from payload is in format of epoch instead of string. We should support parsing epoch in our builtin extractor.

What change needs making?
* Update the argument map of builtin extractor to allow specifying the time string representation is expected in "epoch" format.
* Unit Tests and manual E2E tests,

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request:
chore: use dateparse for extracting event time from payload
Closes #589 

This change uses https://github.com/araddon/dateparse for parsing event time string to `time.Time` object. The benefit of using this library is that it can parse many date strings without knowing format in advance, which makes our job easier when we want to parse epoch time string, because we don't need to implement parsing logics for different granularities.

This change also marks the `format` specification as optional. If specified, we will continue using time.Parse to parse the data. But if user doesn't specify `format`, it's also fine as we use `dateparse` to find the format based on time string.

The only uncertainty is that we haven't tested the performance of `dateparse` under a high-traffic numaflow environment. Based on https://github.com/araddon/dateparse/blob/master/bench_test.go, it seems performing pretty well. If we ever find `dateparse` being the bottleneck of performance, we should fallback and implement our own approach to parse epoch.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


Patch:
diff --git a/go.mod b/go.mod
index 4dc744fa..8b2dcc1f 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/Shopify/sarama v1.30.1
 	github.com/ahmetb/gen-crd-api-reference-docs v0.3.0
 	github.com/antonmedv/expr v1.9.0
+	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	github.com/fsnotify/fsnotify v1.5.1
 	github.com/gavv/httpexpect/v2 v2.3.1
 	github.com/gin-contrib/static v0.0.2-0.20220606235426-ae09b2ea7e39
diff --git a/go.sum b/go.sum
index 1b32b326..3bc3a892 100644
--- a/go.sum
+++ b/go.sum
@@ -91,6 +91,8 @@ github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kd
 github.com/antlr/antlr4/runtime/Go/antlr v0.0.0-20210826220005-b48c857c3a0e/go.mod h1:F7bn7fEU90QkQ3tnmaTx3LTKLEDqnwWODIYppRQ5hnY=
 github.com/antonmedv/expr v1.9.0 h1:j4HI3NHEdgDnN9p6oI6Ndr0G5QryMY0FNxT4ONrFDGU=
 github.com/antonmedv/expr v1.9.0/go.mod h1:5qsM3oLGDND7sDmQGDXHkYfkjYMUX14qsgqmHhwGEk8=
+github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de h1:FxWPpzIjnTlhPwqqXc4/vE0f7GvRjuAsbW+HOIe8KnA=
+github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de/go.mod h1:DCaWoUhZrYW9p1lxo/cm8EmUOOzAPSEZNGF2DK1dJgw=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
@@ -639,6 +641,7 @@ github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
 github.com/mattn/go-runewidth v0.0.4/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/mattn/go-runewidth v0.0.8/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/mattn/go-runewidth v0.0.10/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
@@ -787,6 +790,7 @@ github.com/sagikazarmark/crypt v0.1.0/go.mod h1:B/mN0msZuINBtQ1zZLEQcegFJJf9vnYI
 github.com/sanity-io/litter v1.2.0/go.mod h1:JF6pZUFgu2Q0sBZ+HSV35P8TVPI1TTzEwyu9FXAw2W4=
 github.com/savsgio/gotils v0.0.0-20210617111740-97865ed5a873 h1:N3Af8f13ooDKcIhsmFT7Z05CStZWu4C7Md0uDEy4q6o=
 github.com/savsgio/gotils v0.0.0-20210617111740-97865ed5a873/go.mod h1:dmPawKuiAeG/aFYVs2i+Dyosoo7FNcm+Pi8iK6ZUrX8=
+github.com/scylladb/termtables v0.0.0-20191203121021-c4c0b6d42ff4/go.mod h1:C1a7PQSMz9NShzorzCiG2fk9+xuCgLkPeCvMHYR2OWg=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
diff --git a/pkg/sources/transformer/builtin/event_time/event_time_extractor.go b/pkg/sources/transformer/builtin/event_time/event_time_extractor.go
index 828d7ce8..ce53e71c 100644
--- a/pkg/sources/transformer/builtin/event_time/event_time_extractor.go
+++ b/pkg/sources/transformer/builtin/event_time/event_time_extractor.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/araddon/dateparse"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 
 	"github.com/numaproj/numaflow/pkg/shared/expr"
@@ -32,7 +33,8 @@ type eventTimeExtractor struct {
 	// e.g. `json(payload).metadata.time`
 	expression string
 	// format specifies the layout of extracted time string.
-	// With format, we use the time.Parse function to translate the event time string representation to time.Time object.
+	// with format, eventTimeExtractor uses the time.Parse function to translate the event time string representation to time.Time object.
+	// otherwise if format is not specified, eventTimeExtractor uses dateparse to find format based on the time string.
 	format string
 }
 
@@ -42,9 +44,9 @@ func New(args map[string]string) (functionsdk.MapTFunc, error) {
 		return nil, fmt.Errorf(`missing "expression"`)
 	}
 
-	format, existing := args["format"]
-	if !existing {
-		return nil, fmt.Errorf("missing \"format\"")
+	var format string
+	if format, existing = args["format"]; !existing {
+		format = ""
 	}
 
 	e := eventTimeExtractor{
@@ -63,13 +65,20 @@ func New(args map[string]string) (functionsdk.MapTFunc, error) {
 }
 
 // apply compiles the payload to extract the new event time. If there is any error during extraction,
-// we pass on the default event time. Otherwise, we assign the new event time to the message.
+// we pass on the original input event time. Otherwise, we assign the new event time to the message.
 func (e eventTimeExtractor) apply(et time.Time, payload []byte) (functionsdk.MessageT, error) {
 	timeStr, err := expr.EvalStr(e.expression, payload)
 	if err != nil {
 		return functionsdk.MessageTToAll(et, payload), err
 	}
-	newEventTime, err := time.Parse(e.format, timeStr)
+
+	var newEventTime time.Time
+	time.Local, _ = time.LoadLocation("UTC")
+	if e.format != "" {
+		newEventTime, err = time.Parse(e.format, timeStr)
+	} else {
+		newEventTime, err = dateparse.ParseStrict(timeStr)
+	}
 	if err != nil {
 		return functionsdk.MessageTToAll(et, payload), err
 	} else {


Test patch:
diff --git a/pkg/sources/transformer/builtin/event_time/event_time_extractor_test.go b/pkg/sources/transformer/builtin/event_time/event_time_extractor_test.go
index 0e382434..87085715 100644
--- a/pkg/sources/transformer/builtin/event_time/event_time_extractor_test.go
+++ b/pkg/sources/transformer/builtin/event_time/event_time_extractor_test.go
@@ -49,13 +49,27 @@ func TestEventTimeExtractor(t *testing.T) {
 		assert.Contains(t, err.Error(), "missing \"expression\"")
 	})
 
-	t.Run("Missing format, return error", func(t *testing.T) {
-		_, err := New(map[string]string{"expression": "json(payload).item[1].time"})
-		assert.Error(t, err)
-		assert.Contains(t, err.Error(), "missing \"format\"")
+	t.Run("Json expression valid, assign a new event time to the message - format not specified", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].time"}
+		handle, err := New(args)
+		assert.NoError(t, err)
+
+		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-02-18T21:54:42.123Z"}]}`
+		result := handle(context.Background(), "test-key", &testDatum{
+			value:     []byte(testJsonMsg),
+			eventTime: time.Time{},
+			watermark: time.Time{},
+		})
+
+		time.Local, _ = time.LoadLocation("UTC")
+		expected, _ := time.Parse(time.RFC3339, "2021-02-18T21:54:42.123Z")
+		// Verify new event time is assigned to the message.
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
+		// Verify the payload remains unchanged.
+		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
 	})
 
-	t.Run("Json expression valid, assign a new event time to the message", func(t *testing.T) {
+	t.Run("Json expression valid, assign a new event time to the message - format specified", func(t *testing.T) {
 		args := map[string]string{"expression": "json(payload).item[1].time", "format": time.RFC3339}
 		handle, err := New(args)
 		assert.NoError(t, err)
@@ -67,19 +81,21 @@ func TestEventTimeExtractor(t *testing.T) {
 			watermark: time.Time{},
 		})
 
+		time.Local, _ = time.LoadLocation("UTC")
 		expected, _ := time.Parse(time.RFC3339, "2021-02-18T21:54:42.123Z")
 		// Verify new event time is assigned to the message.
-		assert.Equal(t, expected, result.Items()[0].EventTime)
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
 		// Verify the payload remains unchanged.
 		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
 	})
 
-	t.Run("Cannot compile json expression, pass on the message", func(t *testing.T) {
-		args := map[string]string{"expression": "json(payload).item[1].non-exist-field-name", "format": time.RFC3339}
+	t.Run("Time string not matching user-provided format, pass on the message without assigning new event time", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].time", "format": time.ANSIC}
 		handle, err := New(args)
 		assert.NoError(t, err)
 
 		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
+		// Handler receives format as time.ANSIC but in the message, we use time.RFC3339. Format is not matched.
 		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-02-18T21:54:42.123Z"}]}`
 		result := handle(context.Background(), "test-key", &testDatum{
 			value:     []byte(testJsonMsg),
@@ -93,13 +109,12 @@ func TestEventTimeExtractor(t *testing.T) {
 		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
 	})
 
-	t.Run("Time string not matching user-provided format, pass on the message", func(t *testing.T) {
-		args := map[string]string{"expression": "json(payload).item[1].time", "format": time.ANSIC}
+	t.Run("Cannot compile json expression, pass on the message without assigning new event time", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].non-exist-field-name"}
 		handle, err := New(args)
 		assert.NoError(t, err)
 
 		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
-		// Handler receives format as time.ANSIC but in the message, we use time.RFC3339. Format is not matched.
 		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-02-18T21:54:42.123Z"}]}`
 		result := handle(context.Background(), "test-key", &testDatum{
 			value:     []byte(testJsonMsg),
@@ -107,8 +122,74 @@ func TestEventTimeExtractor(t *testing.T) {
 			watermark: time.Time{},
 		})
 
+		expected := testInputEventTime
 		// Verify event time remains unchanged.
-		assert.Equal(t, testInputEventTime, result.Items()[0].EventTime)
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
+		// Verify the payload remains unchanged.
+		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
+	})
+
+	t.Run("The time string is in epoch format with a granularity of seconds, assign a new event time to the message", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].time"}
+		handle, err := New(args)
+		assert.NoError(t, err)
+
+		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
+		// Handler receives format as time.ANSIC but in the message, we use time.RFC3339. Format is not matched.
+		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "1673239888"},{"id": 2, "name": "numa", "time": "1673239888"}]}`
+		result := handle(context.Background(), "test-key", &testDatum{
+			value:     []byte(testJsonMsg),
+			eventTime: testInputEventTime,
+			watermark: time.Time{},
+		})
+
+		time.Local, _ = time.LoadLocation("UTC")
+		expected := time.Unix(1673239888, 0)
+		// Verify new event time is assigned to the message.
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
+		// Verify the payload remains unchanged.
+		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
+	})
+
+	t.Run("The time string is in epoch format with a granularity of milliseconds, assign a new event time to the message", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].time"}
+		handle, err := New(args)
+		assert.NoError(t, err)
+
+		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
+		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "1673239888123"},{"id": 2, "name": "numa", "time": "1673239888123"}]}`
+		result := handle(context.Background(), "test-key", &testDatum{
+			value:     []byte(testJsonMsg),
+			eventTime: testInputEventTime,
+			watermark: time.Time{},
+		})
+
+		time.Local, _ = time.LoadLocation("UTC")
+		expected := time.Unix(1673239888, 123000000)
+		// Verify new event time is assigned to the message.
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
+		// Verify the payload remains unchanged.
+		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
+	})
+
+	t.Run("The time string is ambiguous, pass on the message without assigning new event time", func(t *testing.T) {
+		args := map[string]string{"expression": "json(payload).item[1].time"}
+		handle, err := New(args)
+		assert.NoError(t, err)
+
+		testInputEventTime := time.Date(2022, 1, 4, 2, 3, 4, 5, time.UTC)
+		// 04/08/2014 is ambiguous because it could be mm/dd/yyyy or dd/mm/yyyy.
+		testJsonMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "04/08/2014 22:05"},{"id": 2, "name": "numa", "time": "04/08/2014 22:05"}]}`
+		result := handle(context.Background(), "test-key", &testDatum{
+			value:     []byte(testJsonMsg),
+			eventTime: testInputEventTime,
+			watermark: time.Time{},
+		})
+
+		time.Local, _ = time.LoadLocation("UTC")
+		expected := testInputEventTime
+		// Verify event time remains unchanged.
+		assert.True(t, expected.Equal(result.Items()[0].EventTime))
 		// Verify the payload remains unchanged.
 		assert.Equal(t, testJsonMsg, string(result.Items()[0].Value))
 	})
diff --git a/test/e2e/testdata/extract-event-time-from-payload.yaml b/test/e2e/testdata/extract-event-time-from-payload.yaml
index 2187ab50..dbb7f2f6 100644
--- a/test/e2e/testdata/extract-event-time-from-payload.yaml
+++ b/test/e2e/testdata/extract-event-time-from-payload.yaml
@@ -12,9 +12,6 @@ spec:
             name: eventTimeExtractor
             kwargs:
               expression: json(payload).item[1].time
-              # Use RFC3339 time format for testing.
-              # More types of format in https://cs.opensource.google/go/go/+/refs/tags/go1.19.5:src/time/format.go
-              format: 2006-01-02T15:04:05Z07:00
     - name: out
       sink:
         log: {}