fix: controller can not start (#354)

* fix: controller can not start

Signed-off-by: wenfeng <[email protected]>

* fix lint

Signed-off-by: wenfeng <[email protected]>

Signed-off-by: wenfeng <[email protected]>

Author: wenfengwang

cmd/controller/main.go

@@ -94,7 +94,7 @@ func main() {
 	}
 
 	//trigger controller
-	triggerCtrlStv := trigger.NewController(cfg.GetTriggerConfig(), etcd)
+	triggerCtrlStv := trigger.NewController(cfg.GetTriggerConfig(), cfg.GetControllerAddrs(), etcd)
 	if err = triggerCtrlStv.Start(); err != nil {
 		log.Error(ctx, "start trigger controller fail", map[string]interface{}{
 			log.KeyError: err,

internal/controller/trigger/controller.go

@@ -39,6 +39,7 @@ import (
 	"github.com/linkall-labs/vanus/pkg/util"
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
 	"github.com/linkall-labs/vanus/proto/pkg/meta"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
@@ -50,12 +51,13 @@ const (
 	defaultGcSubscriptionInterval = time.Second * 10
 )
 
-func NewController(config Config, member embedetcd.Member) *controller {
+func NewController(config Config, controllerAddr []string, member embedetcd.Member) *controller {
 	ctrl := &controller{
 		config:                config,
 		member:                member,
 		needCleanSubscription: map[vanus.ID]string{},
 		state:                 primitive.ServerStateCreated,
+		cl:                    cluster.NewClusterController(controllerAddr, insecure.NewCredentials()),
 	}
 	ctrl.ctx, ctrl.stopFunc = context.WithCancel(context.Background())
 	return ctrl

internal/controller/trigger/controller_test.go

@@ -38,7 +38,7 @@ func TestController_CommitOffset(t *testing.T) {
 	Convey("test reset offset", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -77,7 +77,7 @@ func TestController_ResetOffsetToTimestamp(t *testing.T) {
 	Convey("test reset offset", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -118,7 +118,7 @@ func TestController_CreateSubscription(t *testing.T) {
 	Convey("test create subscription", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -154,7 +154,7 @@ func TestController_UpdateSubscription(t *testing.T) {
 	Convey("test update subscription", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -355,7 +355,7 @@ func TestController_DeleteSubscription(t *testing.T) {
 	Convey("test delete subscription", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -414,7 +414,7 @@ func TestController_GetSubscription(t *testing.T) {
 	Convey("test get subscription", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -451,7 +451,7 @@ func TestController_ListSubscription(t *testing.T) {
 	Convey("test list subscription", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager
@@ -476,7 +476,7 @@ func TestController_TriggerWorkerHeartbeat(t *testing.T) {
 	Convey("test trigger worker heartbeat", t, func() {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
-		ctrl := NewController(Config{}, nil)
+		ctrl := NewController(Config{}, nil, nil)
 		ctx := context.Background()
 		workerManager := worker.NewMockManager(mockCtrl)
 		ctrl.workerManager = workerManager

internal/store/segment/server.go

@@ -260,6 +260,8 @@ func (s *server) reconcileBlocks(ctx context.Context) error {
 
 func (s *server) registerSelf(ctx context.Context) error {
 	// TODO(james.yin): pass information of blocks.
+	start := time.Now()
+	log.Info(ctx, "connecting to controller", nil)
 	if err := s.ctrl.WaitForControllerReady(false); err != nil {
 		return err
 	}
@@ -271,7 +273,9 @@ func (s *server) registerSelf(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-
+	log.Info(ctx, "connected to controller", map[string]interface{}{
+		"used": time.Since(start),
+	})
 	s.id = vanus.NewIDFromUint64(res.ServerId)
 
 	// FIXME(james.yin): some blocks may not be bound to segment.

pkg/cluster/controller.go

@@ -142,7 +142,7 @@ func (c *cluster) IsReady(createEventbus bool) bool {
 	if res.LeaderAddr == "" {
 		return false
 	}
-	return createEventbus && res.GetIsEventbusReady()
+	return !createEventbus || (createEventbus && res.GetIsEventbusReady())
 }
 
 func (c *cluster) Status() Topology {