Repository: vanus-labs/vanus
Pull request #471 (2023-03-03 13:20:38)

feat: modify the default version of cluster creation
### What problem does this PR solve?

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


codecov[bot]: # [Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) Report
> Merging [#471](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) (9998f36) into [main](https://codecov.io/gh/vanus-labs/vanus/commit/2a78cb5c98e21ede2d084739bfadd4151ac48054?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) (2a78cb5) will **decrease** coverage by `0.87%`.
> The diff coverage is `39.67%`.

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://codecov.io/gh/vanus-labs/vanus/pull/471/graphs/tree.svg?width=650&height=150&src=pr&token=RSXSIMEY4V&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)

```diff
@@            Coverage Diff             @@
##             main     #471      +/-   ##
==========================================
- Coverage   57.71%   56.85%   -0.87%     
==========================================
  Files         180      187       +7     
  Lines       14614    15115     +501     
==========================================
+ Hits         8434     8593     +159     
- Misses       5552     5885     +333     
- Partials      628      637       +9     
```


| [Impacted Files](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) | Coverage Δ | |
|---|---|---|
| [...ernal/controller/trigger/validation/subscripton.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvY29udHJvbGxlci90cmlnZ2VyL3ZhbGlkYXRpb24vc3Vic2NyaXB0b24uZ28=) | `69.29% <ø> (+0.85%)` | :arrow_up: |
| [internal/gateway/proxy/deadletter.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvZ2F0ZXdheS9wcm94eS9kZWFkbGV0dGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [internal/gateway/proxy/direct.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvZ2F0ZXdheS9wcm94eS9kaXJlY3QuZ28=) | `73.33% <0.00%> (-5.24%)` | :arrow_down: |
| [.../primitive/interceptor/memberinterceptor/member.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvcHJpbWl0aXZlL2ludGVyY2VwdG9yL21lbWJlcmludGVyY2VwdG9yL21lbWJlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/store/io/file\_linux.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvaW8vZmlsZV9saW51eC5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/store/io/zone/segmentedfile/recovery.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvaW8vem9uZS9zZWdtZW50ZWRmaWxlL3JlY292ZXJ5Lmdv) | `18.98% <0.00%> (ø)` | |
| [internal/store/raft/block/executor.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvcmFmdC9ibG9jay9leGVjdXRvci5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/store/segment/server.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvc2VnbWVudC9zZXJ2ZXIuZ28=) | `14.10% <0.00%> (-0.03%)` | :arrow_down: |
| [internal/store/vsb/engine\_open.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvdnNiL2VuZ2luZV9vcGVuLmdv) | `2.98% <0.00%> (+0.12%)` | :arrow_up: |
| [internal/store/wal/recovery.go](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvd2FsL3JlY292ZXJ5Lmdv) | `57.14% <0.00%> (-2.51%)` | :arrow_down: |
| ... and [50 more](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs). Last update [2f658f2...9998f36](https://codecov.io/gh/vanus-labs/vanus/pull/471?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs).

</details>
JieDing: Great job

Patch:
diff --git a/internal/controller/eventbus/controller.go b/internal/controller/eventbus/controller.go
index f8961d9..22d3456 100644
--- a/internal/controller/eventbus/controller.go
+++ b/internal/controller/eventbus/controller.go
@@ -94,7 +94,8 @@ type controller struct {
 	eventbusDeletedCount int64
 }
 
-func (ctrl *controller) GetEventbusWithHumanFriendly(ctx context.Context, request *ctrlpb.GetEventbusWithHumanFriendlyRequest) (*metapb.Eventbus, error) {
+func (ctrl *controller) GetEventbusWithHumanFriendly(ctx context.Context,
+	request *ctrlpb.GetEventbusWithHumanFriendlyRequest) (*metapb.Eventbus, error) {
 	for id, eventbus := range ctrl.eventbusMap {
 		if eventbus.Name == request.EventbusName {
 			return ctrl.getEventbus(id)
diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go
index b63dc96..971e6d6 100644
--- a/internal/gateway/gateway.go
+++ b/internal/gateway/gateway.go
@@ -16,30 +16,25 @@ package gateway
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
-	"strconv"
 	"strings"
-	"sync"
-	"time"
 
 	v2 "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/client"
 	"github.com/cloudevents/sdk-go/v2/protocol"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
-	"github.com/cloudevents/sdk-go/v2/types"
-	"github.com/google/uuid"
-	"go.opentelemetry.io/otel/trace"
-
 	eb "github.com/vanus-labs/vanus/client"
-	"github.com/vanus-labs/vanus/client/pkg/api"
+	"github.com/vanus-labs/vanus/internal/gateway/proxy"
+	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 	"github.com/vanus-labs/vanus/observability/log"
-	"github.com/vanus-labs/vanus/observability/metrics"
 	"github.com/vanus-labs/vanus/observability/tracing"
-
-	"github.com/vanus-labs/vanus/internal/gateway/proxy"
-	"github.com/vanus-labs/vanus/internal/primitive"
+	"github.com/vanus-labs/vanus/proto/pkg/cloudevents"
+	"github.com/vanus-labs/vanus/proto/pkg/codec"
+	proxypb "github.com/vanus-labs/vanus/proto/pkg/proxy"
+	"go.opentelemetry.io/otel/trace"
 )
 
 const (
@@ -55,7 +50,6 @@ type EventData struct {
 
 type ceGateway struct {
 	// ceClient  v2.Client
-	busWriter  sync.Map
 	config     Config
 	client     eb.Client
 	proxySrv   *proxy.ControllerProxy
@@ -112,114 +106,40 @@ func (ga *ceGateway) startCloudEventsReceiver(ctx context.Context) error {
 }
 
 func (ga *ceGateway) receive(ctx context.Context, event v2.Event) (re *v2.Event, result protocol.Result) {
-	_ctx, span := ga.tracer.Start(ctx, "receive")
-
-	responseCode := 200
-	start := time.Now()
-	ebName := getEventbusFromPath(requestDataFromContext(_ctx))
-	defer func() {
-		used := float64(time.Since(start)) / float64(time.Millisecond)
-		metrics.GatewayEventReceivedCountVec.WithLabelValues(
-			ebName,
-			metrics.LabelValueProtocolHTTP,
-			strconv.FormatInt(1, 10),
-			strconv.Itoa(responseCode),
-		).Inc()
-
-		metrics.GatewayEventWriteLatencySummaryVec.WithLabelValues(
-			ebName,
-			metrics.LabelValueProtocolHTTP,
-			strconv.FormatInt(1, 10),
-		).Observe(used)
-		span.End()
-	}()
-
-	if ebName == "" {
-		responseCode = http.StatusBadRequest
-		return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid eventbus name")
-	}
-
-	extensions := event.Extensions()
-	err := checkExtension(extensions)
+	eventbusID, err := getEventbusFromPath(requestDataFromContext(ctx))
 	if err != nil {
-		responseCode = http.StatusBadRequest
-		return nil, v2.NewHTTPResult(http.StatusBadRequest, err.Error())
-	}
-
-	event.SetExtension(primitive.XVanusEventbus, ebName)
-	if eventTime, ok := extensions[primitive.XVanusDeliveryTime]; ok {
-		// validate event time
-		if _, err := types.ParseTime(eventTime.(string)); 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")
-		}
-		ebName = primitive.TimerEventbusName
+		return nil, v2.NewHTTPResult(http.StatusInternalServerError, err.Error())
 	}
 
-	v, exist := ga.busWriter.Load(ebName)
-	if !exist {
-		v, _ = ga.busWriter.LoadOrStore(ebName, ga.client.Eventbus(ctx, ebName).Writer())
-	}
-	writer, _ := v.(api.BusWriter)
-	eventID, err := api.AppendOne(ctx, writer, &event)
+	e, err := codec.ToProto(&event)
 	if err != nil {
-		log.Warning(_ctx, "append to failed", map[string]interface{}{
-			log.KeyError: err,
-			"eventbus":   ebName,
-		})
-		responseCode = http.StatusInternalServerError
 		return nil, v2.NewHTTPResult(http.StatusInternalServerError, err.Error())
 	}
-	eventData := EventData{
-		BusName: ebName,
-		EventID: eventID,
-	}
-	re, err = createResponseEvent(eventData)
+
+	_, err = ga.proxySrv.Publish(ctx, &proxypb.PublishRequest{
+		Events: &cloudevents.CloudEventBatch{
+			Events: []*cloudevents.CloudEvent{e},
+		},
+		EventbusId: eventbusID.Uint64(),
+	})
+
 	if err != nil {
-		responseCode = http.StatusInternalServerError
 		return nil, v2.NewHTTPResult(http.StatusInternalServerError, err.Error())
 	}
-	return re, v2.ResultACK
-}
 
-func checkExtension(extensions map[string]interface{}) error {
-	if len(extensions) == 0 {
-		return nil
-	}
-	for name := range extensions {
-		if name == primitive.XVanusDeliveryTime {
-			continue
-		}
-		// event attribute can not prefix with vanus system use
-		if strings.HasPrefix(name, primitive.XVanus) {
-			return fmt.Errorf("invalid ce attribute [%s] perfix %s", name, primitive.XVanus)
-		}
-	}
-	return nil
+	return re, v2.ResultACK
 }
 
-func getEventbusFromPath(reqData *cehttp.RequestData) string {
+func getEventbusFromPath(reqData *cehttp.RequestData) (vanus.ID, error) {
 	// TODO validate
 	reqPathStr := reqData.URL.String()
 	if !strings.HasPrefix(reqPathStr, httpRequestPrefix) {
-		return ""
+		return vanus.EmptyID(), errors.New("invalid eventbus id")
 	}
-	return strings.TrimLeft(reqPathStr[len(httpRequestPrefix):], "/")
-}
-
-func createResponseEvent(eventData EventData) (*v2.Event, error) {
-	e := v2.NewEvent("1.0")
-	e.SetID(uuid.NewString())
-	e.SetType("ai.vanus.event.stored")
-	e.SetSource("https://vanus.ai/vanus")
-
-	err := e.SetData(v2.ApplicationJSON, eventData)
+	id, err := vanus.NewIDFromString(strings.TrimLeft(reqPathStr[len(httpRequestPrefix):], "/"))
 	if err != nil {
-		return nil, err
+		return vanus.EmptyID(), err
 	}
-	return &e, nil
+
+	return id, nil
 }
diff --git a/internal/gateway/proxy/deadletter.go b/internal/gateway/proxy/deadletter.go
index bb71276..b540670 100644
--- a/internal/gateway/proxy/deadletter.go
+++ b/internal/gateway/proxy/deadletter.go
@@ -140,7 +140,7 @@ func (cp *ControllerProxy) getDealLetterEventbusID(
 	return vanus.NewIDFromUint64(eb.Id), nil
 }
 
-func (cp *ControllerProxy) ResendDeadLetterEvent(
+func (cp *ControllerProxy) ResendDeadLetterEvent( //nolint:funlen // ok
 	ctx context.Context, req *proxypb.ResendDeadLetterEventRequest,
 ) (*emptypb.Empty, error) {
 	if req.GetSubscriptionId() == 0 {
@@ -249,8 +249,12 @@ loop:
 func (cp *ControllerProxy) writeDeadLetterEvent(
 	ctx context.Context, subscriptionID uint64, offset uint64, events []*cloudevents.CloudEvent,
 ) error {
+	meta, err := cp.ctrl.EventbusService().GetSystemEventbusByName(ctx, primitive.RetryEventbusName)
+	if err != nil {
+		return err
+	}
 	// write to retry eventbus
-	err := cp.writeEvents(ctx, primitive.RetryEventbusName, &cloudevents.CloudEventBatch{
+	err = cp.writeEvents(ctx, vanus.NewIDFromUint64(meta.Id), &cloudevents.CloudEventBatch{
 		Events: events,
 	})
 	if err != nil {
diff --git a/internal/gateway/proxy/direct.go b/internal/gateway/proxy/direct.go
index 32f7e4c..5bf2f3c 100644
--- a/internal/gateway/proxy/direct.go
+++ b/internal/gateway/proxy/direct.go
@@ -18,10 +18,10 @@ import (
 	"context"
 	"errors"
 
-	"google.golang.org/protobuf/types/known/emptypb"
-
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 	metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 var errMethodNotImplemented = errors.New("the method hasn't implemented")
@@ -33,15 +33,15 @@ func (cp *ControllerProxy) CreateEventbus(
 }
 
 func (cp *ControllerProxy) DeleteEventbus(
-	ctx context.Context, req *ctrlpb.DeleteEventbusRequest,
+	ctx context.Context, id *wrapperspb.UInt64Value,
 ) (*emptypb.Empty, error) {
-	return cp.eventbusCtrl.DeleteEventbus(ctx, req)
+	return cp.eventbusCtrl.DeleteEventbus(ctx, id)
 }
 
 func (cp *ControllerProxy) GetEventbus(
-	ctx context.Context, req *ctrlpb.GetEventbusRequest,
+	ctx context.Context, id *wrapperspb.UInt64Value,
 ) (*metapb.Eventbus, error) {
-	return cp.eventbusCtrl.GetEventbus(ctx, req)
+	return cp.eventbusCtrl.GetEventbus(ctx, id)
 }
 
 func (cp *ControllerProxy) ListEventbus(
@@ -50,6 +50,11 @@ func (cp *ControllerProxy) ListEventbus(
 	return cp.eventbusCtrl.ListEventbus(ctx, req)
 }
 
+func (cp *ControllerProxy) GetEventbusWithHumanFriendly(ctx context.Context,
+	request *ctrlpb.GetEventbusWithHumanFriendlyRequest) (*metapb.Eventbus, error) {
+	return cp.eventbusCtrl.GetEventbusWithHumanFriendly(ctx, request)
+}
+
 func (cp *ControllerProxy) UpdateEventbus(
 	_ context.Context, _ *ctrlpb.UpdateEventbusRequest,
 ) (*metapb.Eventbus, error) {
diff --git a/internal/gateway/proxy/proxy.go b/internal/gateway/proxy/proxy.go
index 253ebcc..5add404 100644
--- a/internal/gateway/proxy/proxy.go
+++ b/internal/gateway/proxy/proxy.go
@@ -147,8 +147,9 @@ type ControllerProxy struct {
 }
 
 func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequest) (*emptypb.Empty, error) {
-	if req.EventbusName == "" {
-		return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid eventbus name")
+	meta, _ := cp.ctrl.EventbusService().GetEventbus(ctx, req.EventbusId)
+	if meta == nil {
+		return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid eventbus id")
 	}
 
 	responseCode := 200
@@ -158,20 +159,20 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 		span.End()
 		used := float64(stdtime.Since(start)) / float64(stdtime.Millisecond)
 		metrics.GatewayEventReceivedCountVec.WithLabelValues(
-			req.EventbusName,
+			meta.Name,
 			metrics.LabelValueProtocolHTTP,
 			strconv.FormatInt(int64(len(req.Events.Events)), 10),
 			strconv.Itoa(responseCode),
 		).Add(float64(len(req.Events.Events)))
 
 		metrics.GatewayEventWriteLatencySummaryVec.WithLabelValues(
-			req.EventbusName,
+			meta.Name,
 			metrics.LabelValueProtocolHTTP,
 			strconv.FormatInt(int64(len(req.Events.Events)), 10),
 		).Observe(used)
 	}()
 
-	eventbusName := req.EventbusName
+	eventbusName := meta.Name
 	for idx := range req.Events.Events {
 		e := req.Events.Events[idx]
 		err := checkExtension(e.Attributes)
@@ -195,11 +196,15 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 				responseCode = http.StatusBadRequest
 				return nil, v2.NewHTTPResult(http.StatusBadRequest, "invalid delivery time")
 			}
-			req.EventbusName = primitive.TimerEventbusName
+			tID, err := cp.ctrl.EventbusService().GetSystemEventbusByName(ctx, primitive.TimerEventbusName)
+			if err != nil {
+				return nil, err
+			}
+			req.EventbusId = tID.Id
 		}
 	}
 
-	err := cp.writeEvents(ctx, req.EventbusName, req.Events)
+	err := cp.writeEvents(ctx, vanus.NewIDFromUint64(req.EventbusId), req.Events)
 	if err != nil {
 		return nil, err
 	}
@@ -207,20 +212,20 @@ func (cp *ControllerProxy) Publish(ctx context.Context, req *proxypb.PublishRequ
 }
 
 func (cp *ControllerProxy) writeEvents(ctx context.Context,
-	eventbusName string,
+	eventbusID vanus.ID,
 	events *cloudevents.CloudEventBatch,
 ) error {
-	val, exist := cp.writerMap.Load(eventbusName)
+	val, exist := cp.writerMap.Load(eventbusID)
 	if !exist {
-		val, _ = cp.writerMap.LoadOrStore(eventbusName,
-			cp.client.Eventbus(ctx, eventbusName).Writer())
+		val, _ = cp.writerMap.LoadOrStore(eventbusID,
+			cp.client.Eventbus(ctx, api.WithID(eventbusID.Uint64())).Writer())
 	}
 	w, _ := val.(api.BusWriter)
 	_, err := w.Append(ctx, events)
 	if err != nil {
 		log.Warning(ctx, "append to failed", map[string]interface{}{
 			log.KeyError: err,
-			"eventbus":   eventbusName,
+			"eventbus":   eventbusID.Key(),
 		})
 		return v2.NewHTTPResult(http.StatusInternalServerError, err.Error())
 	}
@@ -448,7 +453,7 @@ func newSubscription(
 			Filters:     info.Filters,
 			Sink:        newsink,
 			Protocol:    info.Protocol,
-			Eventbus:    info.Eventbus,
+			EventbusId:  info.EventbusId,
 			Transformer: info.Transformer,
 			Name:        info.Name,
 			Description: info.Description,
@@ -662,13 +667,13 @@ func (cp *ControllerProxy) LookupOffset(ctx context.Context,
 	elList := make([]api.Eventlog, 0)
 	if req.EventlogId > 0 {
 		id := vanus.NewIDFromUint64(req.EventlogId)
-		l, err := cp.client.Eventbus(ctx, req.GetEventbus()).GetLog(ctx, id.Uint64())
+		l, err := cp.client.Eventbus(ctx, api.WithID(req.EventbusId)).GetLog(ctx, id.Uint64())
 		if err != nil {
 			return nil, err
 		}
 		elList = append(elList, l)
 	} else {
-		ls, err := cp.client.Eventbus(ctx, req.GetEventbus()).ListLog(ctx)
+		ls, err := cp.client.Eventbus(ctx, api.WithID(req.EventbusId)).ListLog(ctx)
 		if err != nil {
 			return nil, err
 		}
@@ -694,7 +699,8 @@ func (cp *ControllerProxy) LookupOffset(ctx context.Context,
 func (cp *ControllerProxy) GetEvent(ctx context.Context,
 	req *proxypb.GetEventRequest,
 ) (*proxypb.GetEventResponse, error) {
-	if req.GetEventbus() == "" {
+	vid := vanus.NewIDFromUint64(req.EventbusId)
+	if vid == vanus.EmptyID() {
 		return nil, errInvalidEventbus
 	}
 
@@ -715,11 +721,11 @@ func (cp *ControllerProxy) GetEvent(ctx context.Context,
 		num = maximumNumberPerGetRequest
 	}
 
-	ls, err := cp.client.Eventbus(ctx, req.GetEventbus()).ListLog(ctx)
+	ls, err := cp.client.Eventbus(ctx, api.WithID(vid.Uint64())).ListLog(ctx)
 	if err != nil {
 		return nil, err
 	}
-	reader := cp.client.Eventbus(ctx, req.GetEventbus()).Reader(
+	reader := cp.client.Eventbus(ctx, api.WithID(vid.Uint64())).Reader(
 		option.WithDisablePolling(),
 		option.WithReadPolicy(policy.NewManuallyReadPolicy(ls[0], offset)),
 		option.WithBatchSize(int(num)),
@@ -744,7 +750,7 @@ func (cp *ControllerProxy) ValidateSubscription(ctx context.Context,
 ) (*proxypb.ValidateSubscriptionResponse, error) {
 	if req.GetEvent() == nil {
 		res, err := cp.GetEvent(ctx, &proxypb.GetEventRequest{
-			Eventbus:   req.Eventbus,
+			EventbusId: req.EventbusId,
 			EventlogId: req.Eventlog,
 			Offset:     req.Offset,
 			Number:     1,
@@ -801,12 +807,12 @@ func (cp *ControllerProxy) getByEventID(ctx context.Context,
 		return nil, err
 	}
 
-	l, err := cp.client.Eventbus(ctx, req.GetEventbus()).GetLog(ctx, logID)
+	l, err := cp.client.Eventbus(ctx, api.WithID(req.GetEventbusId())).GetLog(ctx, logID)
 	if err != nil {
 		return nil, err
 	}
 
-	reader := cp.client.Eventbus(ctx, req.GetEventbus()).Reader(
+	reader := cp.client.Eventbus(ctx, api.WithID(req.GetEventbusId())).Reader(
 		option.WithReadPolicy(policy.NewManuallyReadPolicy(l, off)),
 		option.WithDisablePolling(),
 	)
diff --git a/internal/primitive/transform/common/cast.go b/internal/primitive/transform/common/cast.go
index 375c66e..11ae917 100644
--- a/internal/primitive/transform/common/cast.go
+++ b/internal/primitive/transform/common/cast.go
@@ -22,7 +22,7 @@ import (
 	"strings"
 )
 
-func Cast(val interface{}, target Type) (interface{}, error) {
+func Cast(val interface{}, target Type) (interface{}, error) { //nolint:gocyclo,cyclop // ok
 	if target.IsSameType(val) {
 		return val, nil
 	}
diff --git a/internal/trigger/filter/common.go b/internal/trigger/filter/common.go
index 471cc49..4f0f746 100644
--- a/internal/trigger/filter/common.go
+++ b/internal/trigger/filter/common.go
@@ -18,11 +18,11 @@ import (
 	"context"
 
 	ce "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/types"
 	"github.com/tidwall/gjson"
 
-	"github.com/vanus-labs/vanus/observability/log"
-
 	"github.com/vanus-labs/vanus/internal/trigger/util"
+	"github.com/vanus-labs/vanus/observability/log"
 )
 
 type commonFilter struct {
@@ -31,7 +31,7 @@ type commonFilter struct {
 	meetCondition meetCondition
 }
 
-type meetCondition func(exist bool, value interface{}, compareValue string) bool
+type meetCondition func(value, compareValue string) bool
 
 func newCommonFilter(value map[string]string, meetCondition meetCondition) *commonFilter {
 	attribute := map[string]string{}
@@ -65,22 +65,37 @@ func newCommonFilter(value map[string]string, meetCondition meetCondition) *comm
 func (filter *commonFilter) Filter(event ce.Event) Result {
 	for attr, v := range filter.attribute {
 		value, ok := util.LookupAttribute(event, attr)
-		if !filter.meetCondition(ok, value, v) {
+		if !ok {
+			return false
+		}
+		if !filter.meetCondition(attrValue2String(value), v) {
 			return FailFilter
 		}
 	}
 	for attr, v := range filter.data {
 		if attr == "" {
 			// event data
-			if !filter.meetCondition(true, string(event.Data()), v) {
+			if !filter.meetCondition(string(event.Data()), v) {
 				return FailFilter
 			}
 			continue
 		}
 		result := gjson.GetBytes(event.Data(), attr)
-		if !filter.meetCondition(result.Exists(), result.Value(), v) {
+		if !result.Exists() {
+			return false
+		}
+		if !filter.meetCondition(result.String(), v) {
 			return FailFilter
 		}
 	}
 	return PassFilter
 }
+
+func attrValue2String(value interface{}) string {
+	v, ok := value.(string)
+	if ok {
+		return v
+	}
+	v, _ = types.Format(value)
+	return v
+}
diff --git a/internal/trigger/filter/exact_filter.go b/internal/trigger/filter/exact_filter.go
index 211c2bd..1dd100f 100644
--- a/internal/trigger/filter/exact_filter.go
+++ b/internal/trigger/filter/exact_filter.go
@@ -19,10 +19,7 @@ type exactFilter struct {
 }
 
 func NewExactFilter(exact map[string]string) Filter {
-	f := newCommonFilter(exact, func(exist bool, value interface{}, compareValue string) bool {
-		if !exist {
-			return false
-		}
+	f := newCommonFilter(exact, func(value, compareValue string) bool {
 		return value == compareValue
 	})
 	if f == nil {
diff --git a/internal/trigger/filter/prefix_filter.go b/internal/trigger/filter/prefix_filter.go
index 5c10835..b305f58 100644
--- a/internal/trigger/filter/prefix_filter.go
+++ b/internal/trigger/filter/prefix_filter.go
@@ -16,8 +16,6 @@ package filter
 
 import (
 	"strings"
-
-	pkgUtil "github.com/vanus-labs/vanus/pkg/util"
 )
 
 type prefixFilter struct {
@@ -25,11 +23,8 @@ type prefixFilter struct {
 }
 
 func NewPrefixFilter(prefix map[string]string) Filter {
-	f := newCommonFilter(prefix, func(exist bool, value interface{}, compareValue string) bool {
-		if !exist {
-			return false
-		}
-		return strings.HasPrefix(pkgUtil.StringValue(value), compareValue)
+	f := newCommonFilter(prefix, func(value, compareValue string) bool {
+		return strings.HasPrefix(value, compareValue)
 	})
 	if f == nil {
 		return nil
diff --git a/internal/trigger/filter/suffix_filter.go b/internal/trigger/filter/suffix_filter.go
index 802e149..64f8ebc 100644
--- a/internal/trigger/filter/suffix_filter.go
+++ b/internal/trigger/filter/suffix_filter.go
@@ -16,8 +16,6 @@ package filter
 
 import (
 	"strings"
-
-	pkgUtil "github.com/vanus-labs/vanus/pkg/util"
 )
 
 type suffixFilter struct {
@@ -25,11 +23,8 @@ type suffixFilter struct {
 }
 
 func NewSuffixFilter(suffix map[string]string) Filter {
-	f := newCommonFilter(suffix, func(exist bool, value interface{}, compareValue string) bool {
-		if !exist {
-			return false
-		}
-		return strings.HasSuffix(pkgUtil.StringValue(value), compareValue)
+	f := newCommonFilter(suffix, func(value, compareValue string) bool {
+		return strings.HasSuffix(value, compareValue)
 	})
 	if f == nil {
 		return nil
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 9547b73..6dbda8b 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -26,11 +26,3 @@ func GetIDByAddr(addr string) string {
 func IsSpace(c byte) bool {
 	return c <= ' ' && (c == ' ' || c == '\t' || c == '\r' || c == '\n')
 }
-
-func StringValue(value interface{}) string {
-	v, ok := value.(string)
-	if ok {
-		return v
-	}
-	return fmt.Sprintf("%v", value)
-}
diff --git a/vsctl/command/cluster.go b/vsctl/command/cluster.go
index dec3475..39a4787 100644
--- a/vsctl/command/cluster.go
+++ b/vsctl/command/cluster.go
@@ -33,16 +33,18 @@ import (
 	"gopkg.in/yaml.v2"
 )
 
-var clusterVersionList = []string{
-	"v0.5.0", "v0.5.1", "v0.5.2", "v0.5.3", "v0.5.4", "v0.5.5", "v0.5.6", "v0.5.7", "v0.6.0",
-}
+var (
+	clusterVersionList = []string{"v0.7.0"}
+)
 
 type ClusterCreate struct {
-	ControllerReplicas    int32  `json:"controller_replicas,omitempty"`
-	ControllerStorageSize string `json:"controller_storage_size,omitempty"`
-	StoreReplicas         int32  `json:"store_replicas,omitempty"`
-	StoreStorageSize      string `json:"store_storage_size,omitempty"`
-	Version               string `json:"version,omitempty"`
+	Version            string `json:"version,omitempty"`
+	ControllerReplicas int32  `json:"controller_replicas,omitempty"`
+	StoreReplicas      int32  `json:"store_replicas,omitempty"`
+	GatewayReplicas    int32  `json:"gateway_replicas,omitempty"`
+	TriggerReplicas    int32  `json:"trigger_replicas,omitempty"`
+	TimerReplicas      int32  `json:"timer_replicas,omitempty"`
+	StoreStorageSize   string `json:"store_storage_size,omitempty"`
 }
 
 type ClusterDelete struct {
@@ -71,12 +73,13 @@ type ClusterSpec struct {
 	Version    string      `yaml:"version"`
 	Controller *Controller `yaml:"controller"`
 	Store      *Store      `yaml:"store"`
+	Gateway    *Gateway    `yaml:"gateway"`
 	Trigger    *Trigger    `yaml:"trigger"`
+	Timer      *Timer      `yaml:"timer"`
 }
 
 type Controller struct {
-	Replicas    int32  `yaml:"replicas"`
-	StorageSize string `yaml:"storage_size"`
+	Replicas int32 `yaml:"replicas"`
 }
 
 type Store struct {
@@ -84,10 +87,18 @@ type Store struct {
 	StorageSize string `yaml:"storage_size"`
 }
 
+type Gateway struct {
+	Replicas int32 `yaml:"replicas"`
+}
+
 type Trigger struct {
 	Replicas int32 `yaml:"replicas"`
 }
 
+type Timer struct {
+	Replicas int32 `yaml:"replicas"`
+}
+
 func NewClusterCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "cluster sub-command ",
@@ -185,14 +196,15 @@ func createClusterCommand() *cobra.Command {
 
 			clusterspec := table.NewWriter()
 			clusterspec.AppendHeader(table.Row{"Cluster", "Version", "Component", "Replicas", "StorageSize"})
-			clusterspec.AppendRow(table.Row{
-				"vanus", c.Version, "controller",
-				c.Controller.Replicas, c.Controller.StorageSize,
-			})
+			clusterspec.AppendRow(table.Row{"vanus", c.Version, "controller", c.Controller.Replicas, "-"})
 			clusterspec.AppendSeparator()
 			clusterspec.AppendRow(table.Row{"vanus", c.Version, "store", c.Store.Replicas, c.Store.StorageSize})
 			clusterspec.AppendSeparator()
-			clusterspec.AppendRow(table.Row{"vanus", c.Version, "trigger", c.Trigger.Replicas})
+			clusterspec.AppendRow(table.Row{"vanus", c.Version, "gateway", c.Gateway.Replicas, "-"})
+			clusterspec.AppendSeparator()
+			clusterspec.AppendRow(table.Row{"vanus", c.Version, "trigger", c.Trigger.Replicas, "-"})
+			clusterspec.AppendSeparator()
+			clusterspec.AppendRow(table.Row{"vanus", c.Version, "timer", c.Timer.Replicas, "-"})
 			clusterspec.AppendSeparator()
 			clusterspec.SetColumnConfigs(clusterColConfigs())
 			fmt.Println(clusterspec.Render())
@@ -211,11 +223,13 @@ func createClusterCommand() *cobra.Command {
 			client := &http.Client{}
 			url := fmt.Sprintf("%s%s%s/cluster", HttpPrefix, operatorEndpoint, BaseUrl)
 			cluster := ClusterCreate{
-				ControllerReplicas:    c.Controller.Replicas,
-				ControllerStorageSize: c.Controller.StorageSize,
-				StoreReplicas:         c.Store.Replicas,
-				StoreStorageSize:      c.Store.StorageSize,
-				Version:               c.Version,
+				Version:            c.Version,
+				ControllerReplicas: c.Controller.Replicas,
+				GatewayReplicas:    c.Gateway.Replicas,
+				StoreReplicas:      c.Store.Replicas,
+				TriggerReplicas:    c.Trigger.Replicas,
+				TimerReplicas:      c.Timer.Replicas,
+				StoreStorageSize:   c.Store.StorageSize,
 			}
 			dataByte, err := json.Marshal(cluster)
 			if err != nil {
@@ -794,18 +808,23 @@ func genClusterCommand() *cobra.Command {
 		Short: "generate cluster config file template",
 		Run: func(cmd *cobra.Command, args []string) {
 			cluster := &ClusterSpec{
-				Version: "v0.6.0",
+				Version: "v0.7.0",
 				Controller: &Controller{
-					Replicas:    3,
-					StorageSize: "1Gi",
+					Replicas: 3,
 				},
 				Store: &Store{
 					Replicas:    3,
-					StorageSize: "1Gi",
+					StorageSize: "10Gi",
+				},
+				Gateway: &Gateway{
+					Replicas: 1,
 				},
 				Trigger: &Trigger{
 					Replicas: 1,
 				},
+				Timer: &Timer{
+					Replicas: 2,
+				},
 			}
 			data, err := yaml.Marshal(cluster)
 			if err != nil {
@@ -885,3 +904,4 @@ func getUpgradableVersionList(curVersion string) []string {
 	}
 	return clusterVersionList[curIdx+1:]
 }
+
diff --git a/vsctl/command/eventbus.go b/vsctl/command/eventbus.go
index d7a73d5..1da0d32 100644
--- a/vsctl/command/eventbus.go
+++ b/vsctl/command/eventbus.go
@@ -17,6 +17,7 @@ package command
 import (
 	"context"
 	"encoding/json"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 	"os"
 	"time"
 
@@ -96,9 +97,8 @@ func deleteEventbusCommand() *cobra.Command {
 				cmdFailedf(cmd, "the --name flag MUST be set")
 			}
 
-			_, err := client.DeleteEventbus(context.Background(), &ctrlpb.DeleteEventbusRequest{
-				Id: mustGetEventbusID(namespace, name).Uint64(),
-			})
+			_, err := client.DeleteEventbus(context.Background(),
+				wrapperspb.UInt64(mustGetEventbusID(namespace, eventbus).Uint64()))
 			if err != nil {
 				cmdFailedf(cmd, "delete eventbus failed: %s", err)
 			}
@@ -133,9 +133,8 @@ func getEventbusInfoCommand() *cobra.Command {
 
 			segs := make(map[uint64][]*metapb.Segment)
 			ctx := context.Background()
-			res, err := client.GetEventbus(ctx, &ctrlpb.GetEventbusRequest{
-				Id: mustGetEventbusID(namespace, eventbus).Uint64(),
-			})
+			res, err := client.GetEventbus(ctx,
+				wrapperspb.UInt64(mustGetEventbusID(namespace, eventbus).Uint64()))
 			if err != nil {
 				cmdFailedf(cmd, "get eventbus failed: %s", err)
 			}


Test patch:
diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go
index 03ca68e..ea190fe 100644
--- a/internal/gateway/gateway_test.go
+++ b/internal/gateway/gateway_test.go
@@ -26,11 +26,10 @@ import (
 	. "github.com/golang/mock/gomock"
 	. "github.com/prashantv/gostub"
 	. "github.com/smartystreets/goconvey/convey"
-
 	"github.com/vanus-labs/vanus/client"
 	"github.com/vanus-labs/vanus/client/pkg/api"
-
 	"github.com/vanus-labs/vanus/internal/primitive"
+	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 )
 
 func TestGateway_NewGateway(t *testing.T) {
@@ -110,20 +109,6 @@ func TestGateway_receive(t *testing.T) {
 	// })
 }
 
-func TestGateway_checkExtension(t *testing.T) {
-	Convey("test check extensions", t, func() {
-		e := ce.NewEvent()
-		err := checkExtension(e.Extensions())
-		So(err, ShouldBeNil)
-		e.SetExtension(primitive.XVanusDeliveryTime, "test")
-		err = checkExtension(e.Extensions())
-		So(err, ShouldBeNil)
-		e.SetExtension(primitive.XVanus+"fortest", "test")
-		err = checkExtension(e.Extensions())
-		So(err, ShouldNotBeNil)
-	})
-}
-
 func TestGateway_getEventbusFromPath(t *testing.T) {
 	Convey("test get eventbus from path return nil ", t, func() {
 		reqData := &cehttp.RequestData{
@@ -131,17 +116,19 @@ func TestGateway_getEventbusFromPath(t *testing.T) {
 				Opaque: "/test",
 			},
 		}
-		ret := getEventbusFromPath(reqData)
-		So(ret, ShouldEqual, "")
+		_, err := getEventbusFromPath(reqData)
+		So(err.Error(), ShouldEqual, "invalid eventbus id")
 	})
 	Convey("test get eventbus from path return path ", t, func() {
+		vid := vanus.NewTestID()
 		reqData := &cehttp.RequestData{
 			URL: &url.URL{
-				Opaque: "/gateway/test",
+				Opaque: fmt.Sprintf("/gateway/%d", vid.Uint64()),
 			},
 		}
-		ret := getEventbusFromPath(reqData)
-		So(ret, ShouldEqual, "test")
+		id, err := getEventbusFromPath(reqData)
+		So(id, ShouldEqual, vid)
+		So(err, ShouldBeNil)
 	})
 }
 
diff --git a/internal/gateway/proxy/direct_test.go b/internal/gateway/proxy/direct_test.go
index a48de59..cf8a9b8 100644
--- a/internal/gateway/proxy/direct_test.go
+++ b/internal/gateway/proxy/direct_test.go
@@ -8,6 +8,7 @@ import (
 	. "github.com/smartystreets/goconvey/convey"
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 func TestControllerProxy_ProxyMethod(t *testing.T) {
@@ -30,8 +31,8 @@ func TestControllerProxy_ProxyMethod(t *testing.T) {
 		eventbusCtrl.EXPECT().GetEventbus(gomock.Any(), gomock.Any(), gomock.Any()).Times(1)
 		eventbusCtrl.EXPECT().ListEventbus(gomock.Any(), gomock.Any(), gomock.Any()).Times(1)
 		_, _ = cp.CreateEventbus(stdCtx.Background(), &ctrlpb.CreateEventbusRequest{})
-		_, _ = cp.DeleteEventbus(stdCtx.Background(), &ctrlpb.DeleteEventbusRequest{})
-		_, _ = cp.GetEventbus(stdCtx.Background(), &ctrlpb.GetEventbusRequest{})
+		_, _ = cp.DeleteEventbus(stdCtx.Background(), &wrapperspb.UInt64Value{})
+		_, _ = cp.GetEventbus(stdCtx.Background(), &wrapperspb.UInt64Value{})
 		_, _ = cp.ListEventbus(stdCtx.Background(), &ctrlpb.ListEventbusRequest{})
 		_, err := cp.UpdateEventbus(stdCtx.Background(), &ctrlpb.UpdateEventbusRequest{})
 		So(err, ShouldEqual, errMethodNotImplemented)
diff --git a/internal/gateway/proxy/proxy_test.go b/internal/gateway/proxy/proxy_test.go
index 65e56b9..e94d859 100644
--- a/internal/gateway/proxy/proxy_test.go
+++ b/internal/gateway/proxy/proxy_test.go
@@ -118,18 +118,18 @@ func TestControllerProxy_GetEvent(t *testing.T) {
 			}
 			reader.EXPECT().Read(gomock.Any()).Times(2).Return(batchret, int64(0), id, nil)
 			res, err := cp.GetEvent(stdCtx.Background(), &proxypb.GetEventRequest{
-				Eventbus: "ut1",
-				Offset:   -123,
-				Number:   1,
+				EventbusId: vanus.NewTestID().Uint64(),
+				Offset:     -123,
+				Number:     1,
 			})
 
 			So(err, ShouldBeNil)
 			So(res.Events, ShouldHaveLength, 1)
 
 			res, err = cp.GetEvent(stdCtx.Background(), &proxypb.GetEventRequest{
-				Eventbus: "ut1",
-				Offset:   0,
-				Number:   1,
+				EventbusId: vanus.NewTestID().Uint64(),
+				Offset:     0,
+				Number:     1,
 			})
 
 			So(err, ShouldBeNil)
@@ -181,9 +181,9 @@ func TestControllerProxy_GetEvent(t *testing.T) {
 			}
 			reader.EXPECT().Read(gomock.Any()).Times(1).Return(ret, int64(1234), id, nil)
 			res, err := cp.GetEvent(stdCtx.Background(), &proxypb.GetEventRequest{
-				Eventbus: "ut1",
-				Offset:   1234,
-				Number:   128,
+				EventbusId: vanus.NewTestID().Uint64(),
+				Offset:     1234,
+				Number:     128,
 			})
 
 			So(err, ShouldBeNil)
@@ -233,8 +233,8 @@ func TestControllerProxy_GetEvent(t *testing.T) {
 			}
 			reader.EXPECT().Read(gomock.Any()).Times(1).Return(ret, offset, id, nil)
 			res, err := cp.GetEvent(stdCtx.Background(), &proxypb.GetEventRequest{
-				Eventbus: "ut1",
-				EventId:  base64.StdEncoding.EncodeToString(b),
+				EventbusId: vanus.NewTestID().Uint64(),
+				EventId:    base64.StdEncoding.EncodeToString(b),
 			})
 
 			So(err, ShouldBeNil)
@@ -394,7 +394,7 @@ func TestControllerProxy_ValidateSubscription(t *testing.T) {
 
 			res, err := cp.ValidateSubscription(ctx, &proxypb.ValidateSubscriptionRequest{
 				SubscriptionId: vanus.NewTestID().Uint64(),
-				Eventbus:       "test",
+				EventbusId:     vanus.NewTestID().Uint64(),
 				Eventlog:       vanus.NewTestID().Uint64(),
 				Offset:         123,
 			})