Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix(task): fix an issue where tasks fail to resume on claim (#14356)

pull/14360/head
Lyon Hill 2019-07-16 14:10:28 -06:00 committed by GitHub
parent c6f66e1e21
commit 10bfc91562
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
task/backend/scheduler.go
View File

@ -307,8 +307,6 @@ func (s *TickScheduler) ClaimTask(authCtx context.Context, task *platform.Task)
return err
}
s.taskSchedulers[task.ID] = ts
// pickup any runs that are still "running from a previous failure"
runs, err := s.taskControlService.CurrentlyRunning(authCtx, task.ID)
if err != nil {
@ -320,6 +318,8 @@ func (s *TickScheduler) ClaimTask(authCtx context.Context, task *platform.Task)
}
}
s.taskSchedulers[task.ID] = ts
next, hasQueue := ts.NextDue()
if now := atomic.LoadInt64(&s.now); now >= next || hasQueue {
ts.Work()
@ -509,8 +509,9 @@ func (ts *taskScheduler) Work() {
}
func (ts *taskScheduler) WorkCurrentlyRunning(runs []*platform.Run) error {
foundWorker := false
for _, cr := range runs {
foundWorker := false
for _, r := range ts.runners {
t, err := time.Parse(time.RFC3339, cr.ScheduledFor)
if err != nil {
@ -523,9 +524,10 @@ func (ts *taskScheduler) WorkCurrentlyRunning(runs []*platform.Run) error {
}
}
if !foundWorker {
return errors.New("worker not found to resume work")
}
}
if !foundWorker {
return errors.New("worker not found to resume work")
}
return nil