feat: abstract cluster operation

client/client.go

@@ -40,7 +40,7 @@ type client struct {
 }
 
 func (c *client) Eventbus(ctx context.Context, ebName string) api.Eventbus {
-	_, span := c.tracer.Start(ctx, "Eventbus")
+	_, span := c.tracer.Start(ctx, "EventbusService")
 	defer span.End()
 
 	bus := func() api.Eventbus {

client/internal/vanus/eventbus/name_service.go

@@ -17,12 +17,11 @@ package eventbus
 import (
 	// standard libraries
 	"context"
+	"github.com/linkall-labs/vanus/pkg/cluster"
 
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
-	// third-party libraries
-	"github.com/linkall-labs/vanus/pkg/controller"
 	"google.golang.org/grpc/credentials/insecure"
 
 	// first-party libraries
@@ -33,7 +32,7 @@ import (
 
 func NewNameService(endpoints []string) *NameService {
 	return &NameService{
-		client: controller.NewEventbusClient(endpoints, insecure.NewCredentials()),
+		client: cluster.NewClusterController(endpoints, insecure.NewCredentials()).EventbusService().RawClient(),
 		tracer: tracing.NewTracer("internal.discovery.eventbus", trace.SpanKindClient),
 	}
 }

client/internal/vanus/eventlog/name_service.go

@@ -27,7 +27,7 @@ import (
 	// first-party libraries.
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
-	"github.com/linkall-labs/vanus/pkg/controller"
+	"github.com/linkall-labs/vanus/pkg/cluster"
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
 	metapb "github.com/linkall-labs/vanus/proto/pkg/meta"
 
@@ -38,7 +38,7 @@ import (
 
 func NewNameService(endpoints []string) *NameService {
 	return &NameService{
-		client: controller.NewEventlogClient(endpoints, insecure.NewCredentials()),
+		client: cluster.NewClusterController(endpoints, insecure.NewCredentials()).EventlogService().RawClient(),
 		tracer: tracing.NewTracer("internal.discovery.eventlog", trace.SpanKindClient),
 	}
 }

cmd/controller/main.go

@@ -87,7 +87,7 @@ func main() {
 
 	segmentCtrl := eventbus.NewController(cfg.GetEventbusCtrlConfig(), etcd)
 	if err = segmentCtrl.Start(ctx); err != nil {
-		log.Error(ctx, "start Eventbus Controller failed", map[string]interface{}{
+		log.Error(ctx, "start EventbusService Controller failed", map[string]interface{}{
 			log.KeyError: err,
 		})
 		os.Exit(-1)

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/iceber/iouring-go v0.0.0-20220609112130-b1dc8dd9fbfd
 	github.com/jedib0t/go-pretty/v6 v6.3.1
 	github.com/json-iterator/go v1.1.12
-	github.com/linkall-labs/embed-etcd v0.1.1
+	github.com/linkall-labs/embed-etcd v0.1.2
 	github.com/linkall-labs/vanus/client v0.5.1
 	github.com/linkall-labs/vanus/observability v0.5.1
 	github.com/linkall-labs/vanus/pkg v0.5.1

go.sum

@@ -292,8 +292,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/linkall-labs/embed-etcd v0.1.1 h1:WxV9wbnRtNf7DMW8SJauVYqhFLXzRfY5wpplFypXK9k=
-github.com/linkall-labs/embed-etcd v0.1.1/go.mod h1:dmleSy0Myllw6W5awwjyDMipgICVDHTHuTcRT4cqaIc=
+github.com/linkall-labs/embed-etcd v0.1.2 h1:1mTdXLwVvn9gi3XWh/PGhaEAfG8Zmxvjqwnfontb+fA=
+github.com/linkall-labs/embed-etcd v0.1.2/go.mod h1:QnecHaKt3WQBO9YGBckCDUTBd44VBR2VO8220BtWZ5U=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.11 h1:nQ+aFkoE2TMGc0b68U2OKSexC+eq46+XwZzWXHRmPYs=
 github.com/mattn/go-colorable v0.1.11/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=

internal/controller/eventbus/controller.go

@@ -32,6 +32,7 @@ import (
 	"github.com/linkall-labs/vanus/internal/controller/eventbus/volume"
 	"github.com/linkall-labs/vanus/internal/kv"
 	"github.com/linkall-labs/vanus/internal/kv/etcd"
+	"github.com/linkall-labs/vanus/internal/primitive"
 	"github.com/linkall-labs/vanus/internal/primitive/vanus"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/metrics"
@@ -94,7 +95,6 @@ func (ctrl *controller) Start(_ context.Context) error {
 		return err
 	}
 	ctrl.kvStore = store
-
 	ctrl.cancelCtx, ctrl.cancelFunc = context.WithCancel(context.Background())
 	go ctrl.member.RegisterMembershipChangedProcessor(ctrl.membershipChangedProcessor)
 	return nil
@@ -113,9 +113,48 @@ func (ctrl *controller) StopNotify() <-chan error {
 }
 
 func (ctrl *controller) CreateEventBus(ctx context.Context,
+	req *ctrlpb.CreateEventBusRequest) (*metapb.EventBus, error) {
+	if err := isValidEventbusName(req.Name); err != nil {
+		return nil, err
+	}
+	return ctrl.createEventBus(ctx, req)
+}
+
+func isValidEventbusName(name string) error {
+	name = strings.ToLower(name)
+	for _, v := range name {
+		if v == '.' || v == '_' || v == '-' {
+			continue
+		}
+		c := v - 'a'
+		if c >= 0 || c <= 26 {
+			continue
+		} else {
+			c = v - '0'
+			if c >= 0 || c <= 9 {
+				continue
+			}
+			return errors.ErrInvalidRequest.WithMessage("eventbus name must be insist of 0-9a-zA-Z.-_")
+		}
+	}
+	return nil
+}
+
+func (ctrl *controller) CreateSystemEventBus(ctx context.Context,
+	req *ctrlpb.CreateEventBusRequest) (*metapb.EventBus, error) {
+	if !strings.HasPrefix(req.Name, primitive.SystemEventbusNamePrefix) {
+		return nil, errors.ErrInvalidRequest.WithMessage("system eventbus must start with __")
+	}
+	return ctrl.createEventBus(ctx, req)
+}
+
+func (ctrl *controller) createEventBus(ctx context.Context,
 	req *ctrlpb.CreateEventBusRequest) (*metapb.EventBus, error) {
 	ctrl.mutex.Lock()
 	defer ctrl.mutex.Unlock()
+	if !ctrl.isReady(ctx) {
+		return nil, errors.ErrResourceCanNotOp.WithMessage("the cluster isn't ready to create eventbus")
+	}
 	logNum := req.LogNumber
 	if logNum == 0 {
 		logNum = 1
@@ -451,12 +490,23 @@ func (ctrl *controller) ReportSegmentBlockIsFull(ctx context.Context,
 	return &emptypb.Empty{}, nil
 }
 
-func (ctrl *controller) Ping(_ context.Context, _ *emptypb.Empty) (*ctrlpb.PingResponse, error) {
+func (ctrl *controller) Ping(ctx context.Context, _ *emptypb.Empty) (*ctrlpb.PingResponse, error) {
 	return &ctrlpb.PingResponse{
-		LeaderAddr: ctrl.member.GetLeaderAddr(),
+		LeaderAddr:      ctrl.member.GetLeaderAddr(),
+		IsEventbusReady: ctrl.isReady(ctx),
 	}, nil
 }
 
+func (ctrl *controller) isReady(ctx context.Context) bool {
+	if ctrl.member == nil {
+		return false
+	}
+	if !ctrl.member.IsLeader() && !ctrl.member.IsReady() || ctrl.member.GetLeaderAddr() == "" {
+		return false
+	}
+	return ctrl.ssMgr.CanCreateEventbus(ctx, int(ctrl.cfg.Replicas))
+}
+
 func (ctrl *controller) ReportSegmentLeader(ctx context.Context,
 	req *ctrlpb.ReportSegmentLeaderRequest) (*emptypb.Empty, error) {
 	err := ctrl.eventLogMgr.UpdateSegmentReplicas(ctx, vanus.NewIDFromUint64(req.LeaderId), req.Term)

internal/controller/eventbus/controller_test.go

@@ -20,15 +20,15 @@ import (
 	"sort"
 	"testing"
 
+	"github.com/golang/mock/gomock"
+	embedetcd "github.com/linkall-labs/embed-etcd"
 	"github.com/linkall-labs/vanus/internal/controller/eventbus/eventlog"
 	"github.com/linkall-labs/vanus/internal/controller/eventbus/metadata"
 	"github.com/linkall-labs/vanus/internal/kv"
 	"github.com/linkall-labs/vanus/internal/primitive/vanus"
 	"github.com/linkall-labs/vanus/pkg/errors"
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
 	metapb "github.com/linkall-labs/vanus/proto/pkg/meta"
-
-	"github.com/golang/mock/gomock"
 	. "github.com/smartystreets/goconvey/convey"
 )
 
@@ -44,6 +44,12 @@ func TestController_CreateEventBus(t *testing.T) {
 		ctrl.eventLogMgr = elMgr
 		ctx := stdCtx.Background()
 
+		mockMember := embedetcd.NewMockMember(mockCtrl)
+		ctrl.member = mockMember
+		mockMember.EXPECT().IsLeader().AnyTimes().Return(true)
+		mockMember.EXPECT().IsReady().AnyTimes().Return(true)
+		mockMember.EXPECT().GetLeaderAddr().AnyTimes().Return("test")
+
 		Convey("test create a eventbus two times", func() {
 			kvCli.EXPECT().Exists(ctx, metadata.GetEventbusMetadataKey("test-1")).Times(1).Return(false, nil)
 			kvCli.EXPECT().Set(ctx, metadata.GetEventbusMetadataKey("test-1"), gomock.Any()).

internal/controller/eventbus/server/manager.go

@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:generate mockgen -source=manager.go  -destination=mock_manager.go -package=server
 package server
 
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/golang/protobuf/ptypes/empty"
@@ -36,6 +38,7 @@ type Manager interface {
 	GetServerByServerID(id vanus.ID) Server
 	Run(ctx context.Context) error
 	Stop(ctx context.Context)
+	CanCreateEventbus(ctx context.Context, replicaNum int) bool
 }
 
 const (
@@ -59,6 +62,7 @@ type segmentServerManager struct {
 	cancelCtx            context.Context
 	cancel               func()
 	ticker               *time.Ticker
+	onlineServerNumber   int64
 }
 
 func (mgr *segmentServerManager) AddServer(ctx context.Context, srv Server) error {
@@ -78,9 +82,11 @@ func (mgr *segmentServerManager) AddServer(ctx context.Context, srv Server) erro
 	}
 	mgr.segmentServerMapByIP.Store(srv.Address(), srv)
 	mgr.segmentServerMapByID.Store(srv.ID().Key(), srv)
+	atomic.AddInt64(&mgr.onlineServerNumber, 1)
 	log.Info(ctx, "the segment server added", map[string]interface{}{
 		"server_id": srv.ID(),
 		"addr":      srv.Address(),
+		"online":    atomic.LoadInt64(&mgr.onlineServerNumber),
 	})
 	return nil
 }
@@ -93,6 +99,12 @@ func (mgr *segmentServerManager) RemoveServer(ctx context.Context, srv Server) e
 	defer mgr.mutex.Unlock()
 	mgr.segmentServerMapByIP.Delete(srv.Address())
 	mgr.segmentServerMapByID.Delete(srv.ID().Key())
+	atomic.AddInt64(&mgr.onlineServerNumber, -1)
+	log.Info(ctx, "the segment server was removed", map[string]interface{}{
+		"server_id": srv.ID(),
+		"addr":      srv.Address(),
+		"online":    atomic.LoadInt64(&mgr.onlineServerNumber),
+	})
 	return nil
 }
 
@@ -171,6 +183,18 @@ func (mgr *segmentServerManager) Stop(ctx context.Context) {
 	})
 }
 
+func (mgr *segmentServerManager) CanCreateEventbus(ctx context.Context, replicaNum int) bool {
+	activeNum := 0
+	mgr.segmentServerMapByID.Range(func(_, value any) bool {
+		s, _ := value.(Server)
+		if s.IsActive(ctx) {
+			activeNum++
+		}
+		return true
+	})
+	return activeNum >= replicaNum
+}
+
 type Server interface {
 	RemoteStart(ctx context.Context) error
 	RemoteStop(ctx context.Context)

internal/controller/trigger/controller.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	stdErr "errors"
 	"io"
+	"os"
 	"sync"
 	"time"
 
@@ -33,6 +34,7 @@ import (
 	"github.com/linkall-labs/vanus/internal/primitive/vanus"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/metrics"
+	"github.com/linkall-labs/vanus/pkg/cluster"
 	"github.com/linkall-labs/vanus/pkg/errors"
 	"github.com/linkall-labs/vanus/pkg/util"
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
@@ -74,6 +76,7 @@ type controller struct {
 	ctx                   context.Context
 	stopFunc              context.CancelFunc
 	state                 primitive.ServerState
+	cl                    cluster.Cluster
 }
 
 func (ctrl *controller) CommitOffset(ctx context.Context,
@@ -401,6 +404,7 @@ func (ctrl *controller) requeueSubscription(ctx context.Context, id vanus.ID, ad
 }
 
 func (ctrl *controller) init(ctx context.Context) error {
+	ctrl.initTriggerSystemEventbus()
 	err := ctrl.subscriptionManager.Init(ctx)
 	if err != nil {
 		return err
@@ -464,7 +468,7 @@ func (ctrl *controller) membershipChangedProcessor(ctx context.Context,
 	return nil
 }
 
-func (ctrl *controller) stop(ctx context.Context) error {
+func (ctrl *controller) stop(_ context.Context) error {
 	ctrl.member.ResignIfLeader()
 	ctrl.state = primitive.ServerStateStopping
 	ctrl.stopFunc()
@@ -503,3 +507,33 @@ func (ctrl *controller) Stop(ctx context.Context) {
 		})
 	}
 }
+
+func (ctrl *controller) initTriggerSystemEventbus() {
+	// avoid blocking starting
+	go func() {
+		ctx := context.Background()
+		log.Info(ctx, "trigger controller is ready to check system eventbus", nil)
+		if err := ctrl.cl.WaitForControllerReady(true); err != nil {
+			log.Error(ctx, "trigger controller try to create system eventbus, "+
+				"but Vanus cluster hasn't ready, exit", nil)
+			os.Exit(-1)
+		}
+
+		if err := ctrl.cl.EventbusService().CreateSystemEventbusIfNotExist(ctx, primitive.RetryEventbusName,
+			"System Eventbus For Trigger Service"); err != nil {
+			log.Error(ctx, "failed to create RetryEventbus, exit", map[string]interface{}{
+				log.KeyError: err,
+			})
+			os.Exit(-1)
+		}
+
+		if err := ctrl.cl.EventbusService().CreateSystemEventbusIfNotExist(ctx, primitive.DeadLetterEventbusName,
+			"System Eventbus For Trigger Service"); err != nil {
+			log.Error(ctx, "failed to create DeadLetterEventbus, exit", map[string]interface{}{
+				log.KeyError: err,
+			})
+			os.Exit(-1)
+		}
+		log.Info(ctx, "trigger controller has finished for checking system eventbus", nil)
+	}()
+}