@@ -16,14 +16,17 @@ package trigger
1616
1717import (
1818 "context"
19- stdErr "errors"
19+ stderr "errors"
2020 "fmt"
2121 "io"
2222 "os"
2323 "strings"
2424 "sync"
2525 "time"
2626
27+ "google.golang.org/grpc/credentials/insecure"
28+ "google.golang.org/protobuf/types/known/emptypb"
29+
2730 eb "github.com/vanus-labs/vanus/client"
2831 "github.com/vanus-labs/vanus/internal/controller/member"
2932 "github.com/vanus-labs/vanus/internal/controller/trigger/metadata"
@@ -42,22 +45,21 @@ import (
4245 "github.com/vanus-labs/vanus/pkg/util"
4346 ctrlpb "github.com/vanus-labs/vanus/proto/pkg/controller"
4447 metapb "github.com/vanus-labs/vanus/proto/pkg/meta"
45- "google.golang.org/grpc/credentials/insecure"
46- "google.golang.org/protobuf/types/known/emptypb"
4748)
4849
4950var _ ctrlpb.TriggerControllerServer = & controller {}
5051
5152const (
5253 defaultGcSubscriptionInterval = time .Second * 10
54+ waitEventbusReadyTime = time .Minute * 3
55+ waitEventbusCheckPeriod = time .Second * 2
5356)
5457
5558func NewController (config Config , mem member.Member ) * controller {
5659 ctrl := & controller {
5760 config : config ,
5861 member : mem ,
5962 needCleanSubscription : map [vanus.ID ]string {},
60- state : primitive .ServerStateCreated ,
6163 cl : cluster .NewClusterController (config .ControllerAddr , insecure .NewCredentials ()),
6264 ebClient : eb .Connect (config .ControllerAddr ),
6365 }
@@ -79,17 +81,13 @@ type controller struct {
7981 isLeader bool
8082 ctx context.Context
8183 stopFunc context.CancelFunc
82- state primitive.ServerState
8384 cl cluster.Cluster
8485 ebClient eb.Client
8586}
8687
8788func (ctrl * controller ) SetDeadLetterEventOffset (
8889 ctx context.Context , request * ctrlpb.SetDeadLetterEventOffsetRequest ,
8990) (* emptypb.Empty , error ) {
90- if ctrl .state != primitive .ServerStateRunning {
91- return nil , errors .ErrServerNotStart
92- }
9391 subID := vanus .ID (request .SubscriptionId )
9492 err := ctrl .subscriptionManager .SaveDeadLetterOffset (ctx , subID , request .GetOffset ())
9593 if err != nil {
@@ -101,9 +99,6 @@ func (ctrl *controller) SetDeadLetterEventOffset(
10199func (ctrl * controller ) GetDeadLetterEventOffset (
102100 ctx context.Context , request * ctrlpb.GetDeadLetterEventOffsetRequest ,
103101) (* ctrlpb.GetDeadLetterEventOffsetResponse , error ) {
104- if ctrl .state != primitive .ServerStateRunning {
105- return nil , errors .ErrServerNotStart
106- }
107102 subID := vanus .ID (request .SubscriptionId )
108103 offset , err := ctrl .subscriptionManager .GetDeadLetterOffset (ctx , subID )
109104 if err != nil {
@@ -115,9 +110,6 @@ func (ctrl *controller) GetDeadLetterEventOffset(
115110func (ctrl * controller ) CommitOffset (
116111 ctx context.Context , request * ctrlpb.CommitOffsetRequest ,
117112) (* ctrlpb.CommitOffsetResponse , error ) {
118- if ctrl .state != primitive .ServerStateRunning {
119- return nil , errors .ErrServerNotStart
120- }
121113 resp := new (ctrlpb.CommitOffsetResponse )
122114 for _ , subInfo := range request .SubscriptionInfo {
123115 if len (subInfo .Offsets ) == 0 {
@@ -139,9 +131,6 @@ func (ctrl *controller) CommitOffset(
139131func (ctrl * controller ) ResetOffsetToTimestamp (
140132 ctx context.Context , request * ctrlpb.ResetOffsetToTimestampRequest ,
141133) (* ctrlpb.ResetOffsetToTimestampResponse , error ) {
142- if ctrl .state != primitive .ServerStateRunning {
143- return nil , errors .ErrServerNotStart
144- }
145134 if request .Timestamp == 0 {
146135 return nil , errors .ErrInvalidRequest .WithMessage ("timestamp is invalid" )
147136 }
@@ -166,15 +155,20 @@ func (ctrl *controller) ResetOffsetToTimestamp(
166155func (ctrl * controller ) CreateSubscription (
167156 ctx context.Context , request * ctrlpb.CreateSubscriptionRequest ,
168157) (* metapb.Subscription , error ) {
169- if ctrl .state != primitive .ServerStateRunning {
170- return nil , errors .ErrServerNotStart
171- }
172158 err := validation .ValidateSubscriptionRequest (ctx , request .Subscription )
173159 if err != nil {
174160 log .Info (ctx ).Err (err ).Msg ("create subscription validate fail" )
175161 return nil , err
176162 }
177163 sub := convert .FromPbSubscriptionRequest (request .Subscription )
164+ _ , err = ctrl .cl .NamespaceService ().GetNamespace (ctx , sub .NamespaceID .Uint64 ())
165+ if err != nil {
166+ return nil , err
167+ }
168+ _ , err = ctrl .cl .EventbusService ().GetEventbus (ctx , sub .EventbusID .Uint64 ())
169+ if err != nil {
170+ return nil , err
171+ }
178172 sub .ID , err = vanus .NewID ()
179173 sub .CreatedAt = time .Now ()
180174 sub .UpdatedAt = time .Now ()
@@ -200,9 +194,6 @@ func (ctrl *controller) CreateSubscription(
200194func (ctrl * controller ) UpdateSubscription (
201195 ctx context.Context , request * ctrlpb.UpdateSubscriptionRequest ,
202196) (* metapb.Subscription , error ) {
203- if ctrl .state != primitive .ServerStateRunning {
204- return nil , errors .ErrServerNotStart
205- }
206197 subID := vanus .ID (request .Id )
207198 sub := ctrl .subscriptionManager .GetSubscription (ctx , subID )
208199 if sub == nil {
@@ -243,9 +234,6 @@ func (ctrl *controller) UpdateSubscription(
243234func (ctrl * controller ) DeleteSubscription (
244235 ctx context.Context , request * ctrlpb.DeleteSubscriptionRequest ,
245236) (* emptypb.Empty , error ) {
246- if ctrl .state != primitive .ServerStateRunning {
247- return nil , errors .ErrServerNotStart
248- }
249237 subID := vanus .ID (request .Id )
250238 sub := ctrl .subscriptionManager .GetSubscription (ctx , subID )
251239 if sub != nil {
@@ -269,9 +257,6 @@ func (ctrl *controller) DeleteSubscription(
269257func (ctrl * controller ) DisableSubscription (
270258 ctx context.Context , request * ctrlpb.DisableSubscriptionRequest ,
271259) (* emptypb.Empty , error ) {
272- if ctrl .state != primitive .ServerStateRunning {
273- return nil , errors .ErrServerNotStart
274- }
275260 subID := vanus .ID (request .Id )
276261 sub := ctrl .subscriptionManager .GetSubscription (ctx , subID )
277262 if sub == nil {
@@ -302,9 +287,6 @@ func (ctrl *controller) DisableSubscription(
302287func (ctrl * controller ) ResumeSubscription (
303288 ctx context.Context , request * ctrlpb.ResumeSubscriptionRequest ,
304289) (* emptypb.Empty , error ) {
305- if ctrl .state != primitive .ServerStateRunning {
306- return nil , errors .ErrServerNotStart
307- }
308290 subID := vanus .ID (request .Id )
309291 sub := ctrl .subscriptionManager .GetSubscription (ctx , subID )
310292 if sub == nil {
@@ -326,9 +308,6 @@ func (ctrl *controller) ResumeSubscription(
326308func (ctrl * controller ) GetSubscription (
327309 ctx context.Context , request * ctrlpb.GetSubscriptionRequest ,
328310) (* metapb.Subscription , error ) {
329- if ctrl .state != primitive .ServerStateRunning {
330- return nil , errors .ErrServerNotStart
331- }
332311 sub := ctrl .subscriptionManager .GetSubscription (ctx , vanus .ID (request .Id ))
333312 if sub == nil {
334313 return nil , errors .ErrResourceNotFound .WithMessage ("subscription not exist" )
@@ -355,7 +334,7 @@ func (ctrl *controller) TriggerWorkerHeartbeat(
355334 }
356335 req , err := heartbeat .Recv ()
357336 if err != nil {
358- if ! stdErr .Is (err , io .EOF ) {
337+ if ! stderr .Is (err , io .EOF ) {
359338 log .Warn (ctx ).Err (err ).Msg ("heartbeat recv error" )
360339 }
361340 log .Info (ctx ).Msg ("heartbeat close" )
@@ -562,7 +541,6 @@ func (ctrl *controller) membershipChangedProcessor(
562541 ctrl .subscriptionManager .Start ()
563542 ctrl .scheduler .Run ()
564543 go ctrl .gcSubscriptions (ctx )
565- ctrl .state = primitive .ServerStateRunning
566544 ctrl .isLeader = true
567545 case member .EventBecomeFollower :
568546 if ! ctrl .isLeader {
@@ -579,13 +557,11 @@ func (ctrl *controller) membershipChangedProcessor(
579557
580558func (ctrl * controller ) stop (_ context.Context ) error {
581559 ctrl .member .ResignIfLeader ()
582- ctrl .state = primitive .ServerStateStopping
583560 ctrl .stopFunc ()
584561 ctrl .scheduler .Stop ()
585562 ctrl .workerManager .Stop ()
586563 ctrl .subscriptionManager .Stop ()
587564 ctrl .storage .Close ()
588- ctrl .state = primitive .ServerStateStopped
589565 return nil
590566}
591567
@@ -621,12 +597,40 @@ func (ctrl *controller) initTriggerSystemEventbus() {
621597 go func () {
622598 ctx := context .Background ()
623599 log .Info (ctx ).Msg ("trigger controller is ready to check system eventbus" )
600+ if err := ctrl .cl .WaitForControllerReady (false ); err != nil {
601+ log .Error ().Err (err ).Msg ("trigger controller check system eventbus, " +
602+ "but Vanus cluster hasn't ready, exit" )
603+ os .Exit (- 1 )
604+ }
605+ ready := util .WaitReady (func () bool {
606+ exist , err := ctrl .cl .EventbusService ().IsSystemEventbusExistByName (ctx , primitive .TimerEventbusName )
607+ if err != nil {
608+ log .Error ().Err (err ).Msg ("check TimerEventbus exist has error" )
609+ return false
610+ }
611+ return exist
612+ }, waitEventbusReadyTime , waitEventbusCheckPeriod )
613+ if ! ready {
614+ log .Error ().Msg ("check TimerEventbus timeout no exist, will exist" )
615+ os .Exit (- 1 )
616+ }
617+
618+ // wait TimerEventbus
619+ exist , err := ctrl .cl .EventbusService ().IsSystemEventbusExistByName (ctx , primitive .RetryEventbusName )
620+ if err != nil {
621+ log .Error ().Err (err ).Msg ("failed to check RetryEventbus exist, exit" )
622+ os .Exit (- 1 )
623+ }
624+ if exist {
625+ log .Info ().Msg ("trigger controller check RetryEventbus exist" )
626+ return
627+ }
628+ log .Info ().Msg ("trigger controller check RetryEventbus no exist, will create" )
624629 if err := ctrl .cl .WaitForControllerReady (true ); err != nil {
625630 log .Error (ctx ).Err (err ).
626631 Msg ("trigger controller try to create system eventbus, but Vanus cluster hasn't ready, exit" )
627632 os .Exit (- 1 )
628633 }
629-
630634 if _ , err := ctrl .cl .EventbusService ().CreateSystemEventbusIfNotExist (ctx , primitive .RetryEventbusName ,
631635 "System Eventbus For Trigger Service" ); err != nil {
632636 log .Error (ctx ).Err (err ).Msg ("failed to create RetryEventbus, exit" )
0 commit comments