Repository: vanus-labs/vanus
Pull request #508 (2023-03-09 13:17:34)

feat(eventbus): eventbus name change to id
<!--

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/controller/eventbus/controller.go b/internal/controller/eventbus/controller.go
index fdcedab..f8961d9 100644
--- a/internal/controller/eventbus/controller.go
+++ b/internal/controller/eventbus/controller.go
@@ -20,7 +20,6 @@ import (
 	stdErr "errors"
 	"fmt"
 	"io"
-	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -30,6 +29,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	"github.com/vanus-labs/vanus/observability/log"
 	"github.com/vanus-labs/vanus/observability/metrics"
@@ -64,7 +64,7 @@ func NewController(cfg Config, mem member.Member) *controller {
 	c := &controller{
 		cfg:         &cfg,
 		ssMgr:       server.NewServerManager(),
-		eventbusMap: map[string]*metadata.Eventbus{},
+		eventbusMap: map[vanus.ID]*metadata.Eventbus{},
 		member:      mem,
 		isLeader:    false,
 		readyNotify: make(chan error, 1),
@@ -81,7 +81,7 @@ type controller struct {
 	volumeMgr            volume.Manager
 	eventlogMgr          eventlog.Manager
 	ssMgr                server.Manager
-	eventbusMap          map[string]*metadata.Eventbus
+	eventbusMap          map[vanus.ID]*metadata.Eventbus
 	member               member.Member
 	cancelCtx            context.Context
 	cancelFunc           context.CancelFunc
@@ -94,6 +94,15 @@ type controller struct {
 	eventbusDeletedCount int64
 }
 
+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)
+		}
+	}
+	return nil, errors.ErrResourceNotFound.WithMessage("eventbus not found")
+}
+
 func (ctrl *controller) Start(_ context.Context) error {
 	store, err := etcd.NewEtcdClientV3(ctrl.cfg.KVStoreEndpoints, ctrl.cfg.KVKeyPrefix)
 	if err != nil {
@@ -131,7 +140,7 @@ func (ctrl *controller) CreateEventbus(
 	// TODO async create
 	// create dead letter eventbus
 	_, err = ctrl.createEventbus(context.Background(), &ctrlpb.CreateEventbusRequest{
-		Name:        primitive.GetDeadLetterEventbusName(req.Name),
+		Name:        primitive.GetDeadLetterEventbusName(vanus.NewIDFromUint64(eb.Id)),
 		LogNumber:   1,
 		Description: "System DeadLetter Eventbus For " + req.Name,
 	})
@@ -222,7 +231,7 @@ func (ctrl *controller) createEventbus(
 		}
 		eb.Eventlogs[idx] = el
 	}
-	ctrl.eventbusMap[eb.Name] = eb
+	ctrl.eventbusMap[eb.ID] = eb
 
 	{
 		data, _ := json.Marshal(eb)
@@ -230,41 +239,53 @@ func (ctrl *controller) createEventbus(
 			return nil, err
 		}
 	}
-	return ctrl.getEventbus(eb.Name)
+	return ctrl.getEventbus(eb.ID)
 }
 
-func (ctrl *controller) DeleteEventbus(ctx context.Context, eb *ctrlpb.DeleteEventbusRequest) (*emptypb.Empty, error) {
+func (ctrl *controller) getDeadLetterEventbusID(ctx context.Context, id vanus.ID) vanus.ID {
+	deadLetterEventbusName := primitive.GetDeadLetterEventbusName(id)
+	//
+	for _id, eb := range ctrl.eventbusMap {
+		if eb.Name == deadLetterEventbusName {
+			return _id
+		}
+	}
+	return vanus.EmptyID()
+}
+
+func (ctrl *controller) DeleteEventbus(ctx context.Context, eb *wrapperspb.UInt64Value) (*emptypb.Empty, error) {
 	ctrl.mutex.Lock()
 	defer ctrl.mutex.Unlock()
-	err := ctrl.deleteEventbus(ctx, eb.Name)
+	eventbusID := vanus.NewIDFromUint64(eb.GetValue())
+	err := ctrl.deleteEventbus(ctx, eventbusID)
 	if err != nil {
 		return nil, err
 	}
 	// TODO async delete
 	// delete dead letter eventbus
-	err = ctrl.deleteEventbus(context.Background(),
-		primitive.GetDeadLetterEventbusName(eb.Name))
+	deadLetterEventbusID := ctrl.getDeadLetterEventbusID(ctx, eventbusID)
+	err = ctrl.deleteEventbus(context.Background(), deadLetterEventbusID)
 	if err != nil {
 		log.Error(context.Background(), "delete dead letter eventbus error", map[string]interface{}{
-			log.KeyError:        err,
-			log.KeyEventbusName: eb.Name,
+			log.KeyError:      err,
+			log.KeyEventbusID: eventbusID,
 		})
 	}
 	return &emptypb.Empty{}, nil
 }
 
-func (ctrl *controller) deleteEventbus(ctx context.Context, name string) error {
-	bus, exist := ctrl.eventbusMap[name]
+func (ctrl *controller) deleteEventbus(ctx context.Context, id vanus.ID) error {
+	bus, exist := ctrl.eventbusMap[id]
 	if !exist {
 		return errors.ErrResourceNotFound.WithMessage("the eventbus doesn't exist")
 	}
-	err := ctrl.kvStore.Delete(ctx, metadata.GetEventbusMetadataKey(name))
+	err := ctrl.kvStore.Delete(ctx, metadata.GetEventbusMetadataKey(bus.Name))
 	if err != nil {
 		return errors.ErrInternal.WithMessage("delete eventbus metadata in kv failed").Wrap(err)
 	}
 
 	// TODO(wenfeng.wang) notify gateway to cut flow
-	delete(ctrl.eventbusMap, name)
+	delete(ctrl.eventbusMap, id)
 	wg := sync.WaitGroup{}
 
 	for _, v := range bus.Eventlogs {
@@ -279,12 +300,12 @@ func (ctrl *controller) deleteEventbus(ctx context.Context, name string) error {
 	return nil
 }
 
-func (ctrl *controller) GetEventbus(ctx context.Context, eb *ctrlpb.GetEventbusRequest) (*metapb.Eventbus, error) {
-	return ctrl.getEventbus(eb.Name)
+func (ctrl *controller) GetEventbus(ctx context.Context, eb *wrapperspb.UInt64Value) (*metapb.Eventbus, error) {
+	return ctrl.getEventbus(vanus.NewIDFromUint64(eb.GetValue()))
 }
 
-func (ctrl *controller) getEventbus(name string) (*metapb.Eventbus, error) {
-	_eb, exist := ctrl.eventbusMap[name]
+func (ctrl *controller) getEventbus(id vanus.ID) (*metapb.Eventbus, error) {
+	_eb, exist := ctrl.eventbusMap[id]
 	if !exist {
 		return nil, errors.ErrResourceNotFound.WithMessage("eventbus not found")
 	}
@@ -664,7 +685,7 @@ func (ctrl *controller) loadEventbus(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		ctrl.eventbusMap[filepath.Base(pair.Key)] = busInfo
+		ctrl.eventbusMap[busInfo.ID] = busInfo
 	}
 	return nil
 }
diff --git a/internal/store/raft/block/appender.go b/internal/store/raft/block/appender.go
index 839936b..0b05607 100644
--- a/internal/store/raft/block/appender.go
+++ b/internal/store/raft/block/appender.go
@@ -81,6 +81,7 @@ type appender struct {
 	node    *raft.RawNode
 	storage *storage.Storage
 	host    transport.Host
+	e       *Engine
 
 	hint map[uint64]string
 
diff --git a/internal/store/raft/block/appender_transport.go b/internal/store/raft/block/appender_transport.go
index dc535ef..4b0dd44 100644
--- a/internal/store/raft/block/appender_transport.go
+++ b/internal/store/raft/block/appender_transport.go
@@ -49,6 +49,7 @@ func (a *appender) Receive(ctx context.Context, msg *raftpb.Message, from uint64
 	a.transportExecutor.Execute(func() {
 		if endpoint != "" && a.hint[from] != endpoint {
 			a.hint[from] = endpoint
+			_ = a.e.RegisterNodeRecord(from, endpoint)
 		}
 
 		a.step(msg)
diff --git a/internal/store/raft/block/engine.go b/internal/store/raft/block/engine.go
index 9163ab9..a7fcbc6 100644
--- a/internal/store/raft/block/engine.go
+++ b/internal/store/raft/block/engine.go
@@ -171,6 +171,7 @@ func (e *Engine) newAppender(raw block.Raw, s *storage.Storage) Appender {
 		leaderLis:         e.leaderLis,
 		storage:           s,
 		host:              e.host,
+		e:                 e,
 		hint:              make(map[uint64]string, 3),
 		raftExecutor:      e.executor.newRaftFlow(),
 		appendExecutor:    e.executor.newAppendFlow(),
diff --git a/internal/store/raft/transport/host.go b/internal/store/raft/transport/host.go
index 5626480..fcfbe79 100644
--- a/internal/store/raft/transport/host.go
+++ b/internal/store/raft/transport/host.go
@@ -19,10 +19,6 @@ import (
 	"context"
 	"sync"
 
-	// third-party libraries.
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-
 	// first-party libraries.
 	"github.com/vanus-labs/vanus/observability/log"
 	"github.com/vanus-labs/vanus/raft/raftpb"
@@ -88,8 +84,6 @@ func (h *host) resolveMultiplexer(ctx context.Context, to uint64, endpoint strin
 		if endpoint = h.resolver.Resolve(to); endpoint == "" {
 			return nil
 		}
-		span := trace.SpanFromContext(ctx)
-		span.SetAttributes(attribute.String("endpoint", endpoint))
 	}
 
 	if endpoint == h.callback {
diff --git a/internal/timer/timingwheel/bucket.go b/internal/timer/timingwheel/bucket.go
index 1e49ccf..1b09ff5 100644
--- a/internal/timer/timingwheel/bucket.go
+++ b/internal/timer/timingwheel/bucket.go
@@ -332,13 +332,14 @@ func (b *bucket) createEventbus(ctx context.Context) error {
 	if !b.isLeader() {
 		return nil
 	}
-	return b.timingwheel.ctrl.EventbusService().CreateSystemEventbusIfNotExist(ctx, b.eventbus,
+	_, err := b.timingwheel.ctrl.EventbusService().CreateSystemEventbusIfNotExist(ctx, b.eventbus,
 		"System Eventbus For Timing Service")
+	return err
 }
 
 func (b *bucket) connectEventbus(ctx context.Context) {
-	b.eventbusWriter = b.client.Eventbus(ctx, b.eventbus).Writer()
-	b.eventbusReader = b.client.Eventbus(ctx, b.eventbus).Reader()
+	b.eventbusWriter = b.client.Eventbus(ctx, api.WithName(b.eventbus)).Writer()
+	b.eventbusReader = b.client.Eventbus(ctx, api.WithName(b.eventbus)).Reader()
 }
 
 func (b *bucket) putEvent(ctx context.Context, tm *timingMsg) (err error) {
@@ -385,7 +386,7 @@ func (b *bucket) getEvent(ctx context.Context, number int16) (events []*ce.Event
 		time.Sleep(time.Second)
 		return []*ce.Event{}, errors.ErrOffsetOnEnd
 	}
-	ls, err := b.client.Eventbus(ctx, b.eventbus).ListLog(ctx)
+	ls, err := b.client.Eventbus(ctx, api.WithName(b.eventbus)).ListLog(ctx)
 	if err != nil {
 		return []*ce.Event{}, err
 	}
@@ -485,7 +486,9 @@ func (b *bucket) hasOnEnd(ctx context.Context) bool {
 }
 
 func (b *bucket) recycle(ctx context.Context) {
-	_ = b.timingwheel.ctrl.EventbusService().Delete(ctx, b.eventbus)
+	// TODO(jiangkai): check for errors
+	metaEventbus, _ := b.timingwheel.ctrl.EventbusService().GetSystemEventbusByName(ctx, b.eventbus)
+	_ = b.timingwheel.ctrl.EventbusService().Delete(ctx, metaEventbus.Id)
 	_ = b.deleteOffsetMeta(ctx)
 }
 
diff --git a/internal/timer/timingwheel/timingwheel.go b/internal/timer/timingwheel/timingwheel.go
index 31bcafc..325498b 100644
--- a/internal/timer/timingwheel/timingwheel.go
+++ b/internal/timer/timingwheel/timingwheel.go
@@ -38,6 +38,7 @@ import (
 
 	"github.com/vanus-labs/vanus/internal/kv"
 	"github.com/vanus-labs/vanus/internal/kv/etcd"
+	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 	"github.com/vanus-labs/vanus/internal/timer/metadata"
 )
 
@@ -235,8 +236,14 @@ func (tw *timingWheel) IsLeader() bool {
 }
 
 func (tw *timingWheel) IsDeployed(ctx context.Context) bool {
-	return tw.ctrl.EventbusService().IsExist(ctx, tw.receivingStation.eventbus) &&
-		tw.ctrl.EventbusService().IsExist(ctx, tw.distributionStation.eventbus)
+	var err error
+	if _, err = tw.ctrl.EventbusService().GetSystemEventbusByName(ctx, tw.receivingStation.eventbus); err != nil {
+		return false
+	}
+	if _, err = tw.ctrl.EventbusService().GetSystemEventbusByName(ctx, tw.distributionStation.eventbus); err != nil {
+		return false
+	}
+	return true
 }
 
 func (tw *timingWheel) Recover(ctx context.Context) error {
@@ -569,20 +576,28 @@ func (tw *timingWheel) runDistributionStation(ctx context.Context) {
 
 func (tw *timingWheel) deliver(ctx context.Context, e *ce.Event) error {
 	var (
-		err    error
-		ebName string
+		err  error
+		ebID string
 	)
 
-	err = e.ExtensionAs(xVanusEventbus, &ebName)
+	err = e.ExtensionAs(xVanusEventbus, &ebID)
 	if err != nil {
 		log.Error(ctx, "get eventbus failed when delivering", map[string]interface{}{
 			log.KeyError: err,
 		})
 		return err
 	}
-	v, exist := tw.cache.Load(ebName)
+	eventbusID, err := vanus.NewIDFromString(ebID)
+	if err != nil {
+		log.Error(ctx, "eventbus id string to uint64 failed when delivering", map[string]interface{}{
+			log.KeyError:  err,
+			"eventbus_id": ebID,
+		})
+		return err
+	}
+	v, exist := tw.cache.Load(eventbusID)
 	if !exist {
-		v, _ = tw.cache.LoadOrStore(ebName, tw.client.Eventbus(ctx, ebName).Writer())
+		v, _ = tw.cache.LoadOrStore(eventbusID, tw.client.Eventbus(ctx, api.WithID(eventbusID.Uint64())).Writer())
 	}
 	writer, _ := v.(api.BusWriter)
 	_, err = api.AppendOne(ctx, writer, e)
@@ -590,21 +605,21 @@ func (tw *timingWheel) deliver(ctx context.Context, e *ce.Event) error {
 		if errors.Is(err, errors.ErrOffsetOnEnd) {
 			log.Warning(ctx, "eventbus not found, discard this event", map[string]interface{}{
 				log.KeyError:    err,
-				"eventbus":      ebName,
 				"event_id":      e.ID(),
+				"eventbus_id":   ebID,
 				"delivery_time": newTimingMsg(ctx, e).getExpiration().Format(time.RFC3339Nano),
 			})
 			return nil
 		}
 		log.Error(ctx, "append failed", map[string]interface{}{
-			log.KeyError: err,
-			"eventbus":   ebName,
+			log.KeyError:  err,
+			"eventbus_id": ebID,
 		})
 		return err
 	}
 	log.Debug(ctx, "event delivered", map[string]interface{}{
 		"event_id":      e.ID(),
-		"eventbus":      e.Extensions()[xVanusEventbus],
+		"eventbus_id":   ebID,
 		"delivery_time": newTimingMsg(ctx, e).getExpiration().Format(time.RFC3339Nano),
 	})
 	return nil


Test patch:
diff --git a/internal/controller/eventbus/controller_test.go b/internal/controller/eventbus/controller_test.go
index 8e3b293..72f9f43 100644
--- a/internal/controller/eventbus/controller_test.go
+++ b/internal/controller/eventbus/controller_test.go
@@ -22,12 +22,12 @@ import (
 
 	"github.com/golang/mock/gomock"
 	. "github.com/smartystreets/goconvey/convey"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	"github.com/vanus-labs/vanus/internal/controller/eventbus/eventlog"
 	"github.com/vanus-labs/vanus/internal/controller/eventbus/metadata"
 	"github.com/vanus-labs/vanus/internal/controller/member"
 	"github.com/vanus-labs/vanus/internal/kv"
-	"github.com/vanus-labs/vanus/internal/primitive"
 	"github.com/vanus-labs/vanus/internal/primitive/vanus"
 	"github.com/vanus-labs/vanus/pkg/errors"
 	ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
@@ -55,9 +55,8 @@ func TestController_CreateEventbus(t *testing.T) {
 			kvCli.EXPECT().Exists(ctx, metadata.GetEventbusMetadataKey("test-1")).Times(1).Return(false, nil)
 			kvCli.EXPECT().Set(ctx, metadata.GetEventbusMetadataKey("test-1"), gomock.Any()).
 				Times(1).Return(nil)
-			dlEventbusName := primitive.GetDeadLetterEventbusName("test-1")
-			kvCli.EXPECT().Exists(ctx, metadata.GetEventbusMetadataKey(dlEventbusName)).Times(1).Return(false, nil)
-			kvCli.EXPECT().Set(ctx, metadata.GetEventbusMetadataKey(dlEventbusName), gomock.Any()).
+			kvCli.EXPECT().Exists(ctx, gomock.Any()).Times(1).Return(false, nil)
+			kvCli.EXPECT().Set(ctx, gomock.Any(), gomock.Any()).
 				Times(1).Return(nil)
 			el := &metadata.Eventlog{
 				ID: vanus.NewTestID(),
@@ -86,7 +85,7 @@ func TestController_CreateEventbus(t *testing.T) {
 			So(res.Logs[0].ServerAddress, ShouldResemble, []string{"a", "b"})
 			So(res.Logs[0].CurrentSegmentNumbers, ShouldEqual, 2)
 
-			_, exist := ctrl.eventbusMap["test-1"]
+			_, exist := ctrl.eventbusMap[vanus.NewIDFromUint64(res.Id)]
 			So(exist, ShouldBeTrue)
 		})
 
@@ -121,7 +120,7 @@ func TestController_DeleteEventbus(t *testing.T) {
 		ctx := stdCtx.Background()
 
 		Convey("deleting a doesn't exist eventbus", func() {
-			res, err := ctrl.DeleteEventbus(ctx, &ctrlpb.DeleteEventbusRequest{Name: "test-1"})
+			res, err := ctrl.DeleteEventbus(ctx, &wrapperspb.UInt64Value{Value: vanus.NewTestID().Uint64()})
 			So(err, ShouldNotBeNil)
 			So(res, ShouldBeNil)
 			et, ok := err.(*errors.ErrorType)
@@ -149,8 +148,8 @@ func TestController_DeleteEventbus(t *testing.T) {
 			kvCli.EXPECT().Delete(ctx, metadata.GetEventbusMetadataKey("test-1")).Times(1).
 				Return(fmt.Errorf("test"))
 
-			ctrl.eventbusMap["test-1"] = md
-			res, err := ctrl.DeleteEventbus(stdCtx.Background(), &ctrlpb.DeleteEventbusRequest{Name: "test-1"})
+			ctrl.eventbusMap[md.ID] = md
+			res, err := ctrl.DeleteEventbus(stdCtx.Background(), &wrapperspb.UInt64Value{Value: md.ID.Uint64()})
 			So(err, ShouldNotBeNil)
 			So(res, ShouldBeNil)
 			et, ok := err.(*errors.ErrorType)
@@ -167,11 +166,11 @@ func TestController_DeleteEventbus(t *testing.T) {
 			elMgr.EXPECT().DeleteEventlog(ctx, md.Eventlogs[0].ID).Times(1)
 			elMgr.EXPECT().DeleteEventlog(ctx, md.Eventlogs[1].ID).Times(1)
 
-			ctrl.eventbusMap["test-1"] = md
-			_, err := ctrl.DeleteEventbus(stdCtx.Background(), &ctrlpb.DeleteEventbusRequest{Name: "test-1"})
+			ctrl.eventbusMap[md.ID] = md
+			_, err := ctrl.DeleteEventbus(stdCtx.Background(), &wrapperspb.UInt64Value{Value: md.ID.Uint64()})
 			So(err, ShouldBeNil)
 
-			_, exist := ctrl.eventbusMap["test-1"]
+			_, exist := ctrl.eventbusMap[md.ID]
 			So(exist, ShouldBeFalse)
 		})
 	})