Repository: matrixorigin/matrixone
Pull request #5138 (2022-09-21 04:05:39)

add start&stop cron tasks in hakeeper
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:


Patch:
diff --git a/pkg/hakeeper/checker.go b/pkg/hakeeper/checker.go
index 819ebf009..48b5a56e0 100644
--- a/pkg/hakeeper/checker.go
+++ b/pkg/hakeeper/checker.go
@@ -49,4 +49,11 @@ type BootstrapManager interface {
 
 type TaskScheduler interface {
 	Schedule(cnState pb.CNState, currentTick uint64)
+
+	// StartScheduleCronTask start schedule cron tasks. A timer will be started to pull the latest CronTask
+	// from the TaskStore at regular intervals, and a timer will be maintained in memory for all Cron's to be
+	// triggered at regular intervals.
+	StartScheduleCronTask()
+	// StopScheduleCronTask stop schedule cron tasks.
+	StopScheduleCronTask()
 }
diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go
index 4f55ecc83..bf8cbecbb 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -115,29 +115,33 @@ func (l *store) hakeeperCheck() {
 		return
 	}
 
-	if isLeader {
-		state, err := l.getCheckerState()
-		if err != nil {
-			// TODO: check whether this is temp error
-			logger.Error("failed to get checker state", zap.Error(err))
-			return
-		}
-		switch state.State {
-		case pb.HAKeeperCreated:
-			logger.Warn("waiting for initial cluster info to be set, check skipped")
-			return
-		case pb.HAKeeperBootstrapping:
-			l.bootstrap(term, state)
-		case pb.HAKeeperBootstrapCommandsReceived:
-			l.checkBootstrap(state)
-		case pb.HAKeeperBootstrapFailed:
-			l.handleBootstrapFailure()
-		case pb.HAKeeperRunning:
-			l.healthCheck(term, state)
-			l.taskSchedule(state)
-		default:
-			panic("unknown HAKeeper state")
-		}
+	if !isLeader {
+		l.taskScheduler.StopScheduleCronTask()
+		return
+	}
+
+	l.taskScheduler.StartScheduleCronTask()
+	state, err := l.getCheckerState()
+	if err != nil {
+		// TODO: check whether this is temp error
+		logger.Error("failed to get checker state", zap.Error(err))
+		return
+	}
+	switch state.State {
+	case pb.HAKeeperCreated:
+		logger.Warn("waiting for initial cluster info to be set, check skipped")
+		return
+	case pb.HAKeeperBootstrapping:
+		l.bootstrap(term, state)
+	case pb.HAKeeperBootstrapCommandsReceived:
+		l.checkBootstrap(state)
+	case pb.HAKeeperBootstrapFailed:
+		l.handleBootstrapFailure()
+	case pb.HAKeeperRunning:
+		l.healthCheck(term, state)
+		l.taskSchedule(state)
+	default:
+		panic("unknown HAKeeper state")
 	}
 }
 
diff --git a/pkg/taskservice/task_service.go b/pkg/taskservice/task_service.go
index 1fe1ada0a..c7fd4f6ae 100644
--- a/pkg/taskservice/task_service.go
+++ b/pkg/taskservice/task_service.go
@@ -36,6 +36,7 @@ type taskService struct {
 		sync.Mutex
 
 		started  bool
+		stopping bool
 		stopper  *stopper.Stopper
 		cron     *cron.Cron
 		retryC   chan task.CronTask
diff --git a/pkg/taskservice/task_service_cron.go b/pkg/taskservice/task_service_cron.go
index a159e2523..0d0dd09ab 100644
--- a/pkg/taskservice/task_service_cron.go
+++ b/pkg/taskservice/task_service_cron.go
@@ -35,7 +35,7 @@ func (s *taskService) StartScheduleCronTask() {
 	s.crons.Lock()
 	defer s.crons.Unlock()
 
-	if s.crons.started {
+	if s.crons.started || s.crons.stopping {
 		return
 	}
 
@@ -62,6 +62,7 @@ func (s *taskService) StopScheduleCronTask() {
 	}
 	stopper := s.crons.stopper
 	s.crons.started = false
+	s.crons.stopping = true
 	s.crons.stopper = nil
 	s.crons.Unlock()
 
@@ -76,6 +77,7 @@ func (s *taskService) StopScheduleCronTask() {
 	close(s.crons.retryC)
 	s.crons.jobs = nil
 	s.crons.entryIDs = nil
+	s.crons.stopping = false
 }
 
 func (s *taskService) fetchCronTasks(ctx context.Context) {


Test patch:
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index e5ee5718d..9dabbc112 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -484,6 +484,9 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 				t.Fatalf("unexpected error %v", err)
 			}
 			if completed {
+				for _, s := range services[:3] {
+					s.store.taskScheduler.StopScheduleCronTask()
+				}
 				return
 			}
 			time.Sleep(5 * time.Millisecond)
@@ -805,6 +808,7 @@ func TestTaskSchedulerCanReScheduleExpiredTasks(t *testing.T) {
 			}
 			completed := tn()
 			if completed {
+				store.taskScheduler.StopScheduleCronTask()
 				return
 			}
 			time.Sleep(100 * time.Millisecond)