diff --git a/manager/orchestrator/jobs/orchestrator_restart_test.go b/manager/orchestrator/jobs/orchestrator_restart_test.go index e09ccfa3f6..345f72072f 100644 --- a/manager/orchestrator/jobs/orchestrator_restart_test.go +++ b/manager/orchestrator/jobs/orchestrator_restart_test.go @@ -230,6 +230,91 @@ var _ = Describe("Jobs RestartSupervisor Integration", func() { Expect(task.Status.State).To(Equal(api.TaskStateFailed)) } }) + + It("should only restart global job tasks MaxAttempts times", func() { + service.Spec.Mode = &api.ServiceSpec_GlobalJob{ + GlobalJob: &api.GlobalJob{}, + } + service.Spec.Task.Restart = &api.RestartPolicy{ + Condition: api.RestartOnFailure, + MaxAttempts: 3, + // set a low but non-zero delay duration, so we avoid default + // duration, which may be long. + Delay: gogotypes.DurationProto(100 * time.Millisecond), + } + err := s.Update(func(tx store.Tx) error { + if err := store.CreateNode(tx, &api.Node{ + ID: "node1", + Spec: api.NodeSpec{ + Availability: api.NodeAvailabilityActive, + }, + Status: api.NodeStatus{ + State: api.NodeStatus_READY, + }, + }); err != nil { + return err + } + return store.CreateService(tx, service) + }) + Expect(err).ToNot(HaveOccurred()) + + passEventsUntil(o, serviceCreated(service)) + + // fail the original task and its first 2 replacements, waiting each + // time for the replacement task to be started. + for range 3 { + err = s.Update(func(tx store.Tx) error { + tasks, err := store.FindTasks(tx, store.ByTaskState(api.TaskStateNew)) + if err != nil { + return err + } + if len(tasks) != 1 { + return fmt.Errorf("expected 1 new task, there are %v", len(tasks)) + } + tasks[0].Status.State = api.TaskStateFailed + return store.UpdateTask(tx, tasks[0]) + }) + Expect(err).ToNot(HaveOccurred()) + + passEventsUntil(o, taskFailed) + + passEventsUntil(o, func(event events.Event) bool { + updated, ok := event.(api.EventUpdateTask) + return ok && + updated.Task.DesiredState == api.TaskStateCompleted && + updated.OldTask.DesiredState == api.TaskStateReady + }) + } + + // fail the third replacement. MaxAttempts is used up, so no new + // task may be created. + err = s.Update(func(tx store.Tx) error { + tasks, err := store.FindTasks(tx, store.ByTaskState(api.TaskStateNew)) + if err != nil { + return err + } + if len(tasks) != 1 { + return fmt.Errorf("expected 1 new task, there are %v", len(tasks)) + } + tasks[0].Status.State = api.TaskStateFailed + return store.UpdateTask(tx, tasks[0]) + }) + Expect(err).ToNot(HaveOccurred()) + + passEventsUntil(o, taskFailed) + + var tasks []*api.Task + s.View(func(tx store.ReadTx) { + tasks, err = store.FindTasks(tx, store.All) + }) + Expect(err).ToNot(HaveOccurred()) + Expect(tasks).To(HaveLen(4)) + + for _, task := range tasks { + Expect(task.NodeID).To(Equal("node1")) + Expect(task.Status.State).To(Equal(api.TaskStateFailed)) + } + }) }) func serviceCreated(service *api.Service) func(events.Event) bool { diff --git a/manager/orchestrator/restart/restart.go b/manager/orchestrator/restart/restart.go index a13340112b..5f215ee0b0 100644 --- a/manager/orchestrator/restart/restart.go +++ b/manager/orchestrator/restart/restart.go @@ -248,9 +248,9 @@ func (r *Supervisor) shouldRestart(ctx context.Context, t *api.Task, service *ap ServiceID: t.ServiceID, } - // Slot is not meaningful for "global" tasks, so they need to be - // indexed by NodeID. - if orchestrator.IsGlobalService(service) { + // Slot is not meaningful for "global" tasks (global services and global + // jobs), so they need to be indexed by NodeID. + if orchestrator.IsGlobalService(service) || orchestrator.IsGlobalJob(service) { instanceTuple.NodeID = t.NodeID }