chore: add some test for routine

Author: Ethereal-O

vermeer/apps/master/bl/scheduler_bl.go

@@ -244,10 +244,11 @@ func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) erro
 	s.taskManager.RemoveTask(taskId)
 	// release the worker group
 	s.resourceManager.ReleaseByTaskID(taskId)
-	// stop the cron job if exists
-	s.cronManager.DeleteTask(taskId)
 
 	if len(removeWorkerName) > 0 {
+		// stop the cron job if exists
+		s.cronManager.DeleteTask(taskId)
+		// remove the worker from resource manager
 		workerName := removeWorkerName[0]
 		if workerName == "" {
 			return errors.New("the argument `removeWorkerName` is empty")
@@ -408,6 +409,16 @@ func (s *ScheduleBl) handleCancelTask(taskInfo *structure.TaskInfo) error {
 	return nil
 }
 
+func (s *ScheduleBl) CancelCronTask(taskInfo *structure.TaskInfo) error {
+	if taskInfo == nil {
+		return errors.New("the argument `taskInfo` is nil")
+	}
+
+	s.cronManager.DeleteTask(taskInfo.ID)
+
+	return nil
+}
+
 // ** Other Methods **
 
 func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {

vermeer/apps/master/bl/task_bl.go

@@ -182,6 +182,9 @@ func (tb *TaskBl) CancelTask(taskID int32) error {
 		return fmt.Errorf("cannot cancel the task with id '%v' as it was not created by you", taskID)
 	}
 
+	// stop the cron job if exists
+	Scheduler.CancelCronTask(task)
+
 	if task.State == structure.TaskStateCanceled {
 		return fmt.Errorf("task had been in state canceled")
 	}

vermeer/apps/master/schedules/scheduler_cron_manager.go

@@ -27,6 +27,7 @@ func (t *SchedulerCronManager) CheckCronExpression(cronExpr string) error {
 		return errors.New("cron expression is empty")
 	}
 	if _, err := cron.ParseStandard(cronExpr); err != nil {
+		logrus.Errorf("Failed to parse cron expression: %v", err)
 		return errors.New("invalid cron expression: " + err.Error())
 	}
 	return nil
@@ -41,22 +42,44 @@ func (t *SchedulerCronManager) AddCronTask(taskInfo *structure.TaskInfo) error {
 		return errors.New("the property `CronExpr` of taskInfo is empty")
 	}
 
+	// add to cron tasks
 	t.cronTasks[taskInfo.ID] = append(t.cronTasks[taskInfo.ID], taskInfo)
 	cronJob := cron.New()
 	_, err := cronJob.AddFunc(taskInfo.CronExpr, func() {
 		if taskInfo == nil {
 			return
 		}
-		if _, err := t.queueHandler(taskInfo); err != nil {
+
+		// TODO: CREATE a new task from the original task, using taskbl
+		// copy a new taskInfo
+		task, err := structure.TaskManager.CreateTask(taskInfo.SpaceName, taskInfo.Type, 0)
+		task.CreateType = structure.TaskCreateAsync
+		task.GraphName = taskInfo.GraphName
+		task.CreateUser = taskInfo.CreateUser
+		task.Params = taskInfo.Params
+		task.CronExpr = "" // clear cron expression for the new task
+		task.Priority = taskInfo.Priority
+		task.Preorders = taskInfo.Preorders
+		task.Exclusive = taskInfo.Exclusive
+		if err != nil {
+			logrus.Errorf("Failed to create task from cron job for task %d: %v", taskInfo.ID, err)
+			return
+		}
+		structure.TaskManager.AddTask(task)
+		structure.TaskManager.SaveTask(task.ID)
+		if _, err := t.queueHandler(task); err != nil {
 			logrus.Errorf("Failed to queue task %d in cron job: %v", taskInfo.ID, err)
 			return
 		}
+		logrus.Infof("Successfully queued task %d from cron job", task.ID)
 	})
 	if err != nil {
 		logrus.Errorf("Failed to add cron job for task %d: %v", taskInfo.ID, err)
 		return err
 	}
 	t.crons[taskInfo.ID] = append(t.crons[taskInfo.ID], cronJob)
+	cronJob.Start()
+	logrus.Infof("Added cron task for task ID %d with expression %s", taskInfo.ID, taskInfo.CronExpr)
 	return nil
 }
 
@@ -73,3 +96,28 @@ func (t *SchedulerCronManager) DeleteTask(taskID int32) error {
 	logrus.Infof("Deleted cron task for task ID %d", taskID)
 	return nil
 }
+
+func (t *SchedulerCronManager) DeleteTaskByGraph(spaceName, graphName string) error {
+	if spaceName == "" || graphName == "" {
+		return errors.New("the argument `spaceName` or `graphName` is empty")
+	}
+
+	var toDelete []int32
+	for taskID, tasks := range t.cronTasks {
+		for _, task := range tasks {
+			if task.SpaceName == spaceName && task.GraphName == graphName {
+				toDelete = append(toDelete, taskID)
+				break
+			}
+		}
+	}
+
+	for _, taskID := range toDelete {
+		if err := t.DeleteTask(taskID); err != nil {
+			logrus.Errorf("Failed to delete cron task for task ID %d: %v", taskID, err)
+			return err
+		}
+	}
+	logrus.Infof("Deleted cron tasks for space %s and graph %s", spaceName, graphName)
+	return nil
+}

vermeer/test/scheduler/priority.go

@@ -86,6 +86,7 @@ func SubTestConcurrent(t *testing.T, expectRes *functional.ExpectRes, healthChec
 
 	// send two tasks with different size
 	params := make([]map[string]string, 0)
+	// default is false, actually do not need to set
 	taskComputeBody["exclusive"] = "false"
 	params = append(params, taskComputeBody)
 	params = append(params, taskComputeBody)

vermeer/test/scheduler/routine.go

@@ -19,22 +19,28 @@ func SubTestRoutine(t *testing.T, expectRes *functional.ExpectRes, healthCheck *
 	computeTest.Init(graphName[0], computeTask, expectRes, waitSecond, masterHttp, t, healthCheck)
 	taskComputeBody := computeTest.TaskComputeBody()
 
-	taskComputeBody["cron_expr"] = "*/1 * * * * *" // every second
+	// every 1 minute
+	taskComputeBody["cron_expr"] = "* * * * *"
 
 	logrus.Infof("params for routine test: %+v", taskComputeBody)
 
-	computeTest.SendComputeReqAsync(taskComputeBody)
-	computeTest.CheckRes()
+	taskid := computeTest.SendComputeReqAsyncNotWait(taskComputeBody)
+	// computeTest.CheckRes()
 
 	// wait for a while and check again
-	time.Sleep(10 * time.Second)
+	time.Sleep(2 * time.Minute)
+
+	masterHttp.GetTaskCancel(int(taskid))
+
 	fmt.Printf("Test Routine: %-30s [OK], cost: %v\n", computeTask, time.Since(bTime))
 }
 
 func TestRoutine(t *testing.T, expectRes *functional.ExpectRes, healthCheck *functional.HealthCheck, masterHttp *client.VermeerClient, graphName []string, factor string, waitSecond int) {
+	var computeTask = "pagerank"
+
 	// TEST GROUP: ROUTINE
 	// 1. send tasks to single graph
 	// expect: the tasks should be executed timely
 
-	SubTestRoutine(t, expectRes, healthCheck, masterHttp, graphName, factor, waitSecond)
+	SubTestRoutine(t, expectRes, healthCheck, masterHttp, graphName, computeTask, waitSecond)
 }