@@ -2,10 +2,13 @@ package crew
22
33import (
44 "fmt"
5+ "log"
56 "sort"
67 "strings"
78 "sync"
89 "time"
10+
11+ "github.com/go-co-op/gocron"
912)
1013
1114// A ThrottlePushQuery is a request to the throttler to see if there is enough bandwidth for a worker to run.
@@ -28,11 +31,13 @@ type Throttler struct {
2831
2932// TaskGroup represents a group of tasks.
3033type TaskController struct {
31- Storage TaskStorage
32- Client TaskClient
33- Feed chan interface {}
34- Throttler * Throttler
35- Pending * sync.WaitGroup
34+ Storage TaskStorage
35+ Client TaskClient
36+ Feed chan interface {}
37+ Throttler * Throttler
38+ Pending * sync.WaitGroup
39+ AbandonedCheckScheduler * gocron.Scheduler
40+ AbandonedCheckMutex * sync.Mutex
3641}
3742
3843// NewTaskController returns a new TaskController.
@@ -43,6 +48,8 @@ func NewTaskController(storage TaskStorage, client TaskClient, throttler *Thrott
4348 Feed : make (chan interface {}, 8 ),
4449 Throttler : throttler ,
4550 Pending : & sync.WaitGroup {},
51+ // AbandonedCheckScheduler is created in startup
52+ AbandonedCheckMutex : & sync.Mutex {},
4653 }
4754}
4855
@@ -492,53 +499,75 @@ func (controller *TaskController) UpdateTask(id string, update map[string]interf
492499}
493500
494501func (controller * TaskController ) Startup () (err error ) {
495- // Fire an evaluate for any tasks that are incomplete and unpaused
502+ // Restart tasks on startup and/or check for tasks that may have been abandoned due to crashes (or power outages) during execution.
503+ // Note that for this to work for abandonments the storage mechanism must have expirations on task locks.
504+ // Only the redis storage mechanism currently supports this.
505+ s := gocron .NewScheduler (time .UTC )
506+ // TODO - configure interval with an env var?
507+ s .Every (15 ).Minutes ().Do (func () {
508+ log .Println ("Abandoned task scan starting" )
509+
510+ // Use a mutex to make sure this doesn't run more than once at a time
511+ locked := controller .AbandonedCheckMutex .TryLock ()
512+ if ! locked {
513+ log .Println ("Previous abandoned task scan still running, bailing out." )
514+ return
515+ }
516+ defer controller .AbandonedCheckMutex .Unlock ()
496517
497- go func () {
498- groups , groupsError := controller .Storage .AllTaskGroups ()
499- if groupsError == nil {
500- for _ , group := range groups {
501- fmt .Println ("~~ Bootstrapping tasks for group" , group .Id )
518+ taskGroups , taskGroupsError := controller .Storage .AllTaskGroups ()
519+ if taskGroupsError == nil {
520+ for _ , group := range taskGroups {
502521 tasks , tasksError := controller .Storage .AllTasksInGroup (group .Id )
503522 if tasksError == nil {
504523 for _ , task := range tasks {
505- if ! task .IsComplete && ! task .IsPaused {
524+ // Do a couple of quick checks to prevent unecessary evaluates
525+ if ! task .IsComplete && ! task .IsPaused && task .RunAfter .After (time .Now ()) && task .RemainingAttempts > 0 {
506526 controller .TriggerTaskEvaluate (task .Id )
507527 // Slight pause here to prevent a flood of evaluates
508528 time .Sleep (time .Second / 10 )
509529 }
510530 }
511531 } else {
512- fmt .Println ("~~ Error bootstrapping tasks" , tasksError )
532+ log .Println ("Error scanning for abandoned tasks" , tasksError )
513533 }
514- // Longer pause between groups to prevent overloading ourselves.
534+
535+ // Pause between groups to prevent overloading ourselves.
515536 time .Sleep (time .Second * 1 )
516537 }
517538 } else {
518- fmt .Println ("~~ Error bootstrapping task groups" , groupsError )
539+ log .Println ("Error scanning for abandoned tasks (fetch groups) " , taskGroupsError )
519540 }
520- }()
541+
542+ log .Println ("Abandoned task scan completed" )
543+ })
544+ s .StartAsync ()
545+ controller .AbandonedCheckScheduler = s
521546
522547 return nil
523548}
524549
525550func (controller * TaskController ) Shutdown () (err error ) {
551+ if controller .AbandonedCheckScheduler != nil {
552+ controller .AbandonedCheckScheduler .Stop ()
553+ }
554+
526555 // Wait till all pending task executions are complete
527556 controller .Pending .Wait ()
528557 return nil
529558}
530559
531560func (controller * TaskController ) Evaluate (task * Task ) {
532561 parents , _ := controller .Storage .GetTaskParents (task .Id )
533- fmt .Println ("~~ Evaluating task" , task .Id , len (parents ))
562+ log .Println ("Evaluating task" , task .Id , len (parents ))
534563 canExecute := task .CanExecute (parents )
535564 if canExecute {
536565 controller .Execute (task )
537566 }
538567}
539568
540569func (controller * TaskController ) Execute (taskToExecute * Task ) {
541- fmt .Println ("~~ Executing task" , taskToExecute .Id )
570+ log .Println ("Executing task" , taskToExecute .Id )
542571 parents , _ := controller .Storage .GetTaskParents (taskToExecute .Id )
543572
544573 timer := time .NewTimer (1000 * time .Second )
@@ -548,10 +577,10 @@ func (controller *TaskController) Execute(taskToExecute *Task) {
548577 go func () {
549578 defer controller .Pending .Done ()
550579
551- fmt .Println ("~~ Waiting for task start time (go routine)" , taskToExecute .Id )
580+ log .Println ("Waiting for task start time (go routine)" , taskToExecute .Id )
552581 <- timer .C
553582
554- fmt .Println ("~~ Executing task (go routine)" , taskToExecute .Id )
583+ log .Println ("Executing task (go routine)" , taskToExecute .Id )
555584
556585 // Lock is as close to worker request send as possible (in case task delay is longer than lock timeout)
557586 unlocker , lockError := controller .Storage .TryLockTask (taskToExecute .Id )
@@ -564,7 +593,7 @@ func (controller *TaskController) Execute(taskToExecute *Task) {
564593
565594 if lockError != nil {
566595 // Couldn't lock task, do not execute
567- fmt .Println ("~~ Executing task (lock fail)" , taskToExecute .Id )
596+ log .Println ("Executing task (lock fail)" , taskToExecute .Id )
568597 return
569598 }
570599
@@ -619,10 +648,10 @@ func (controller *TaskController) Execute(taskToExecute *Task) {
619648 task .BusyExecuting = false
620649
621650 if err != nil {
622- fmt .Println ("~~ Got standard error" , err )
651+ log .Println ("Got standard error" , task . Id , err )
623652 controller .HandleExecuteError (task , fmt .Sprintf ("%v" , err ))
624653 } else if workerResponse .Error != nil {
625- fmt .Println ("~~ Got worker response error" , workerResponse .Error )
654+ log .Println ("Got worker response error" , task . Id , workerResponse .Error )
626655 controller .HandleExecuteError (task , fmt .Sprintf ("%v" , workerResponse .Error ))
627656 } else {
628657 // No error!
@@ -674,7 +703,7 @@ func (controller *TaskController) Execute(taskToExecute *Task) {
674703
675704 // Because children failed we have to fail the task so that users will know something went wrong.
676705 task .IsComplete = false
677- fmt .Println ("~~ Got child creation error" , errorCreatingChildren )
706+ log .Println ("Got child creation error" , task . Id , errorCreatingChildren )
678707 controller .HandleExecuteError (task , fmt .Sprintf ("Child create failure : %v" , errorCreatingChildren ))
679708 } else {
680709 for _ , child := range createdChildren {
0 commit comments