Repository: vanus-labs/vanus
Pull request #518 (2023-03-11 09:30:16)

feat: change deliverytime type to ce timestamp
<!--

Thank you for contributing to Vanus!

PR Title Format: 

feat/fix/refactor/docs/...: what's changed

Note: see https://github.com/vanus-labs/vanus/blob/main/CONTRIBUTING.md to know details about title format

-->

### What problem does this PR solve?

<!--

Please create an issue first to describe the problem

There MUST be one line starting with "Issue Number:  ".

-->

Issue Number: close #xxx

### Problem Summary

### What is changed and how does it work?

### Check List

<!-- At least one of them must be included. -->

#### Tests

- [ ] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No code



Patch:
diff --git a/internal/gateway/proxy/proxy.go b/internal/gateway/proxy/proxy.go
index 1c55567..421472a 100644
--- a/internal/gateway/proxy/proxy.go
+++ b/internal/gateway/proxy/proxy.go
@@ -183,12 +183,23 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 			Attr: &cloudevents.CloudEvent_CloudEventAttributeValue_CeString{CeString: eventbusID.Key()},
 		}
 		if eventTime, ok := e.Attributes[primitive.XVanusDeliveryTime]; ok {
-			// validate event time
-			if _, err := types.ParseTime(eventTime.GetCeString()); err != nil {
-				log.Error(_ctx, "invalid format of event time", map[string]interface{}{
-					log.KeyError: err,
-					"eventTime":  eventTime.String(),
-				})
+			switch attr := eventTime.GetAttr().(type) {
+			case *cloudevents.CloudEvent_CloudEventAttributeValue_CeTimestamp:
+			case *cloudevents.CloudEvent_CloudEventAttributeValue_CeString:
+				// validate event time
+				t, err := types.ParseTime(attr.CeString)
+				if err != nil {
+					log.Error(_ctx, "invalid format of event time", map[string]interface{}{
+						log.KeyError: err,
+						"eventTime":  eventTime.String(),
+					})
+					responseCode = http.StatusBadRequest
+					return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid delivery time")
+				}
+				eventTime.Attr = &cloudevents.CloudEvent_CloudEventAttributeValue_CeTimestamp{
+					CeTimestamp: timestamppb.New(t),
+				}
+			default:
 				responseCode = http.StatusBadRequest
 				return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid delivery time")
 			}
diff --git a/internal/kv/constant.go b/internal/kv/constant.go
index e98fc58..2bd8423 100644
--- a/internal/kv/constant.go
+++ b/internal/kv/constant.go
@@ -23,18 +23,11 @@ const (
 	ResourceVolumeBlock       = "/vanus/core/eventbus_controller/volume/block"
 	ResourceVolumeInstance    = "/vanus/core/eventbus_controller/volume/instance"
 	ResourceSubscription      = "/vanus/core/trigger_controller/subscriptions"
-	MetadataSecret            = "/vanus/core/trigger_controller/secrets"
+	MetadataSecret            = "/vanus/core/trigger_controller/secrets" //nolint:gosec // ok
 	MetadataOffset            = "/vanus/core/trigger_controller/offsets"
 	TriggerWorker             = "/vanus/core/trigger_controller/trigger_workers"
 	LeaderLock                = "/vanus/core/cluster/resource_lock"
 	LeaderInfo                = "/vanus/core/cluster/leader_info"
 	ClusterNode               = "/vanus/core/cluster/nodes"
 	ClusterStart              = "/vanus/core/cluster/start_at"
-
-	namespace             = "/vanus/core/tenants/namespaces"
-	namespaceEventbus     = "/vanus/core/tenants/{ns_id}/eventbus/{eventbus_id}"
-	namespaceSubscription = "/vanus/core/tenants/{ns_id}/subscription/{subscription_id}"
-	userRoles             = "/vanus/core/tenants/users/{user_id}/roles/{role_id}"
-	roleUsers             = "/vanus/core/tenants/roles/{role_id}/users/{user_id}"
-	rolePermissions       = "/vanus/core/tenants/roles/{role_id}/permissions"
 )
diff --git a/internal/timer/timingwheel/bucket.go b/internal/timer/timingwheel/bucket.go
index 1b09ff5..8a3a395 100644
--- a/internal/timer/timingwheel/bucket.go
+++ b/internal/timer/timingwheel/bucket.go
@@ -24,13 +24,13 @@ import (
 	"time"
 
 	ce "github.com/cloudevents/sdk-go/v2"
-	"github.com/cloudevents/sdk-go/v2/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/vanus-labs/vanus/client"
 	"github.com/vanus-labs/vanus/client/pkg/api"
 	"github.com/vanus-labs/vanus/client/pkg/option"
 	"github.com/vanus-labs/vanus/client/pkg/policy"
+	"github.com/vanus-labs/vanus/internal/primitive"
 	"github.com/vanus-labs/vanus/observability/log"
 	"github.com/vanus-labs/vanus/pkg/errors"
 
@@ -42,8 +42,8 @@ const (
 	timerBuiltInEventbusReceivingStation    = "__Timer_RS"
 	timerBuiltInEventbusDistributionStation = "__Timer_DS"
 	timerBuiltInEventbus                    = "__Timer_%d_%d"
-	xVanusEventbus                          = "xvanuseventbus"
-	xVanusDeliveryTime                      = "xvanusdeliverytime"
+	xVanusEventbus                          = primitive.XVanusEventbus
+	xVanusDeliveryTime                      = primitive.XVanusDeliveryTime
 	sleepDuration                           = 100 * time.Millisecond
 )
 
@@ -53,19 +53,16 @@ type timingMsg struct {
 }
 
 func newTimingMsg(ctx context.Context, e *ce.Event) *timingMsg {
-	var (
-		err        error
-		expiration time.Time
-	)
+	var expiration time.Time
 	extensions := e.Extensions()
 	if deliveryTime, ok := extensions[xVanusDeliveryTime]; ok {
-		expiration, err = types.ParseTime(deliveryTime.(string))
-		if err != nil {
+		if t, ok := deliveryTime.(ce.Timestamp); !ok {
 			log.Error(ctx, "parse time failed", map[string]interface{}{
-				log.KeyError: err,
-				"time":       deliveryTime,
+				"time": deliveryTime,
 			})
 			expiration = time.Now()
+		} else {
+			expiration = t.Time
 		}
 	} else {
 		log.Error(ctx, "xvanusdeliverytime not found, set to current time", nil)
diff --git a/internal/trigger/trigger/trigger.go b/internal/trigger/trigger/trigger.go
index 9c23719..0be3b5a 100644
--- a/internal/trigger/trigger/trigger.go
+++ b/internal/trigger/trigger/trigger.go
@@ -456,8 +456,7 @@ func (t *trigger) writeEventToRetry(ctx context.Context, e *ce.Event, attempts i
 	attempts++
 	ec.Extensions[primitive.XVanusRetryAttempts] = attempts
 	delayTime := calDeliveryTime(attempts)
-	ec.Extensions[primitive.XVanusDeliveryTime] =
-		ce.Timestamp{Time: time.Now().Add(delayTime).UTC()}.Format(time.RFC3339)
+	ec.Extensions[primitive.XVanusDeliveryTime] = ce.Timestamp{Time: time.Now().Add(delayTime)}
 	ec.Extensions[primitive.XVanusSubscriptionID] = t.subscriptionIDStr
 	ec.Extensions[primitive.XVanusEventbus] = t.subscription.RetryEventbusID.Key()
 	var writeAttempt int
@@ -492,8 +491,7 @@ func (t *trigger) writeEventToDeadLetter(ctx context.Context, e *ce.Event, reaso
 	ec, _ := e.Context.(*ce.EventContextV1)
 	delete(ec.Extensions, primitive.XVanusEventbus)
 	ec.Extensions[primitive.XVanusSubscriptionID] = t.subscriptionIDStr
-	ec.Extensions[primitive.LastDeliveryTime] =
-		ce.Timestamp{Time: time.Now().UTC()}.Format(time.RFC3339)
+	ec.Extensions[primitive.LastDeliveryTime] = ce.Timestamp{Time: time.Now()}
 	ec.Extensions[primitive.LastDeliveryError] = errorMsg
 	ec.Extensions[primitive.DeadLetterReason] = reason
 	var writeAttempt int


Test patch:
diff --git a/internal/controller/eventbus/block/block_test.go b/internal/controller/eventbus/block/block_test.go
index 980e8ad..102a481 100644
--- a/internal/controller/eventbus/block/block_test.go
+++ b/internal/controller/eventbus/block/block_test.go
@@ -170,7 +170,7 @@ func TestAllocator_RunWithoutDynamic(t *testing.T) {
 }
 
 // disable on 3.10
-//func TestAllocator_RunWithDynamic(t *testing.T) {
+// func TestAllocator_RunWithDynamic(t *testing.T) {
 //	Convey("test run without dynamic allocate block", t, func() {
 //		ctrl := gomock.NewController(t)
 //		srv1 := server.NewMockInstance(ctrl)
@@ -262,7 +262,7 @@ func TestAllocator_RunWithoutDynamic(t *testing.T) {
 //		So(list2.Len(), ShouldEqual, 8)
 //		So(list3.Len(), ShouldEqual, 8)
 //	})
-//}
+// }
 
 func TestAllocator_UpdateBlock(t *testing.T) {
 	Convey("update update block in kv", t, func() {
diff --git a/internal/timer/timingwheel/bucket_test.go b/internal/timer/timingwheel/bucket_test.go
index 75b9770..94ade65 100644
--- a/internal/timer/timingwheel/bucket_test.go
+++ b/internal/timer/timingwheel/bucket_test.go
@@ -43,10 +43,10 @@ func TestTimingMsg_newTimingMsg(t *testing.T) {
 		e.SetID("1")
 
 		Convey("test timing message new1", func() {
-			t := time.Now().Add(time.Second).UTC().Format("2006-01-02T15:04:05")
+			t := time.Now().Add(time.Second)
 			e.SetExtension(xVanusDeliveryTime, t)
 			tm := newTimingMsg(ctx, &e)
-			So(time.Now().Add(100*time.Millisecond).After(tm.expiration), ShouldBeTrue)
+			So(time.Now().Add(100*time.Millisecond).Before(tm.expiration), ShouldBeTrue)
 		})
 
 		Convey("test timing message new2", func() {
@@ -60,7 +60,7 @@ func TestTimingMsg_hasExpired(t *testing.T) {
 	Convey("test timing message is expired", t, func() {
 		ctx := context.Background()
 		e := ce.NewEvent()
-		e.SetExtension(xVanusDeliveryTime, time.Now().Add(2*time.Second).UTC().Format(time.RFC3339))
+		e.SetExtension(xVanusDeliveryTime, time.Now().Add(2*time.Second))
 		tm := newTimingMsg(ctx, &e)
 		So(tm.hasExpired(), ShouldEqual, false)
 	})
@@ -558,7 +558,7 @@ func TestBucket_getOffset(t *testing.T) {
 
 func event(i int64) *ce.Event {
 	e := ce.NewEvent()
-	t := time.Now().Add(time.Duration(i) * time.Millisecond).UTC().Format(time.RFC3339)
+	t := time.Now().Add(time.Duration(i) * time.Millisecond)
 	e.SetExtension(xVanusDeliveryTime, t)
 	e.SetExtension(xVanusEventbus, "quick-start")
 	return &e
@@ -566,7 +566,7 @@ func event(i int64) *ce.Event {
 
 func batch(i int64) *cloudevents.CloudEventBatch {
 	e := ce.NewEvent()
-	t := time.Now().Add(time.Duration(i) * time.Millisecond).UTC().Format(time.RFC3339)
+	t := time.Now().Add(time.Duration(i) * time.Millisecond)
 	e.SetExtension(xVanusDeliveryTime, t)
 	e.SetExtension(xVanusEventbus, "quick-start")
 	events := make([]*cloudevents.CloudEvent, 1)