diff --git a/manager/manager.go b/manager/manager.go index 8b14005641..d2ab18e56e 100644 --- a/manager/manager.go +++ b/manager/manager.go @@ -170,6 +170,12 @@ type Manager struct { dekRotator *RaftDEKManager roleManager *roleManager + // becomeFollowerCleanups holds the teardown for each leader-only + // component started by becomeLeader, registered with onBecomeFollower + // next to the code that starts it. becomeFollower runs and clears it. + // Guarded by mu. + becomeFollowerCleanups []func() + cancelFunc context.CancelFunc // mu is a general mutex used to coordinate starting/stopping and @@ -1062,6 +1068,10 @@ func (m *Manager) becomeLeader(ctx context.Context) { log.G(ctx).WithError(err).Error("keymanager failed with an error") } }(m.keyManager) + m.onBecomeFollower(func() { + m.keyManager.Stop() + m.keyManager = nil + }) } go func(d *dispatcher.Dispatcher) { @@ -1080,16 +1090,23 @@ func (m *Manager) becomeLeader(ctx context.Context) { log.G(ctx).WithError(err).Error("Dispatcher exited with an error") } }(m.dispatcher) + // The dispatcher, logbroker and CA server are gRPC services that are + // registered when creating the manager and would need to be re-registered + // if they were recreated. For simplicity, they are stopped but not nilled + // out. + m.onBecomeFollower(func() { m.dispatcher.Stop() }) if err := m.logbroker.Start(ctx); err != nil { log.G(ctx).WithError(err).Error("LogBroker failed to start") } + m.onBecomeFollower(func() { m.logbroker.Stop() }) go func(server *ca.Server) { if err := server.Run(ctx); err != nil { log.G(ctx).WithError(err).Error("CA signer exited with an error") } }(m.caserver) + m.onBecomeFollower(func() { m.caserver.Stop() }) // Start all sub-components in separate goroutines. // TODO(aluzzardi): This should have some kind of error handling so that @@ -1100,6 +1117,10 @@ func (m *Manager) becomeLeader(ctx context.Context) { log.G(ctx).WithError(err).Error("allocator exited with an error") } }(m.allocator) + m.onBecomeFollower(func() { + m.allocator.Stop() + m.allocator = nil + }) } go func(scheduler *scheduler.Scheduler) { @@ -1107,88 +1128,97 @@ func (m *Manager) becomeLeader(ctx context.Context) { log.G(ctx).WithError(err).Error("scheduler exited with an error") } }(m.scheduler) + m.onBecomeFollower(func() { + m.scheduler.Stop() + m.scheduler = nil + }) go func(constraintEnforcer *constraintenforcer.ConstraintEnforcer) { constraintEnforcer.Run() }(m.constraintEnforcer) + m.onBecomeFollower(func() { + m.constraintEnforcer.Stop() + m.constraintEnforcer = nil + }) go func(volumeEnforcer *volumeenforcer.VolumeEnforcer) { volumeEnforcer.Run() }(m.volumeEnforcer) + m.onBecomeFollower(func() { + m.volumeEnforcer.Stop() + m.volumeEnforcer = nil + }) go func(taskReaper *taskreaper.TaskReaper) { taskReaper.Run(ctx) }(m.taskReaper) + m.onBecomeFollower(func() { + m.taskReaper.Stop() + m.taskReaper = nil + }) go func(orchestrator *replicated.Orchestrator) { if err := orchestrator.Run(ctx); err != nil { log.G(ctx).WithError(err).Error("replicated orchestrator exited with an error") } }(m.replicatedOrchestrator) + m.onBecomeFollower(func() { + m.replicatedOrchestrator.Stop() + m.replicatedOrchestrator = nil + }) go func(orchestrator *jobs.Orchestrator) { // jobs orchestrator does not return errors. orchestrator.Run(ctx) }(m.jobsOrchestrator) + m.onBecomeFollower(func() { + m.jobsOrchestrator.Stop() + m.jobsOrchestrator = nil + }) go func(globalOrchestrator *global.Orchestrator) { if err := globalOrchestrator.Run(ctx); err != nil { log.G(ctx).WithError(err).Error("global orchestrator exited with an error") } }(m.globalOrchestrator) + m.onBecomeFollower(func() { + m.globalOrchestrator.Stop() + m.globalOrchestrator = nil + }) go func(roleManager *roleManager) { roleManager.Run(ctx) }(m.roleManager) + m.onBecomeFollower(func() { + m.roleManager.Stop() + m.roleManager = nil + }) go func(volumeManager *csi.Manager) { volumeManager.Run(ctx) }(m.volumeManager) + m.onBecomeFollower(func() { + m.volumeManager.Stop() + m.volumeManager = nil + }) } -// becomeFollower shuts down the subsystems that are only run by the leader. +// becomeFollower shuts down the subsystems that are only run by the leader, +// by running the teardowns that becomeLeader registered with onBecomeFollower. func (m *Manager) becomeFollower() { - // The following components are gRPC services that are - // registered when creating the manager and will need - // to be re-registered if they are recreated. - // For simplicity, they are not nilled out. - m.dispatcher.Stop() - m.logbroker.Stop() - m.caserver.Stop() - - if m.allocator != nil { - m.allocator.Stop() - m.allocator = nil - } - - m.constraintEnforcer.Stop() - m.constraintEnforcer = nil - - m.volumeEnforcer.Stop() - m.volumeEnforcer = nil - - m.replicatedOrchestrator.Stop() - m.replicatedOrchestrator = nil - - m.globalOrchestrator.Stop() - m.globalOrchestrator = nil - - m.taskReaper.Stop() - m.taskReaper = nil - - m.scheduler.Stop() - m.scheduler = nil - - m.roleManager.Stop() - m.roleManager = nil - - if m.keyManager != nil { - m.keyManager.Stop() - m.keyManager = nil + cleanups := m.becomeFollowerCleanups + m.becomeFollowerCleanups = nil + for _, f := range cleanups { + f() } +} - m.volumeManager.Stop() - m.volumeManager = nil +// onBecomeFollower registers f to run when this manager loses leadership. +// becomeLeader calls it immediately after the code that starts each +// leader-only component, so that a component cannot be started on the leader +// without its shutdown being written alongside. +func (m *Manager) onBecomeFollower(f func()) { + m.becomeFollowerCleanups = append(m.becomeFollowerCleanups, f) } // defaultClusterObject creates a default cluster. diff --git a/manager/manager_test.go b/manager/manager_test.go index 50872b1fca..008e45c30c 100644 --- a/manager/manager_test.go +++ b/manager/manager_test.go @@ -432,3 +432,19 @@ func TestManagerLockUnlock(t *testing.T) { // error. <-done } + +// TestBecomeFollowerRunsRegisteredCleanups checks the registry that pairs each +// leader-only component's start with its stop. Cleanups run once, in +// registration order, and a second demotion runs nothing. +func TestBecomeFollowerRunsRegisteredCleanups(t *testing.T) { + m := &Manager{} + var ran []string + m.onBecomeFollower(func() { ran = append(ran, "first") }) + m.onBecomeFollower(func() { ran = append(ran, "second") }) + + m.becomeFollower() + require.Equal(t, []string{"first", "second"}, ran) + + m.becomeFollower() + require.Equal(t, []string{"first", "second"}, ran) +} diff --git a/manager/orchestrator/jobs/fakes_test.go b/manager/orchestrator/jobs/fakes_test.go index a2202248d5..bf39a05ff8 100644 --- a/manager/orchestrator/jobs/fakes_test.go +++ b/manager/orchestrator/jobs/fakes_test.go @@ -36,7 +36,7 @@ type fakeReconciler struct { // ReconcileService implements the reconciler's ReconcileService method, but // just records what arguments it has been passed, and maybe also returns an // error if desired. -func (f *fakeReconciler) ReconcileService(id string) error { +func (f *fakeReconciler) ReconcileService(_ context.Context, id string) error { f.Lock() defer f.Unlock() f.servicesReconciled = append(f.servicesReconciled, id) diff --git a/manager/orchestrator/jobs/global/reconciler.go b/manager/orchestrator/jobs/global/reconciler.go index c519507b81..bfbb06372d 100644 --- a/manager/orchestrator/jobs/global/reconciler.go +++ b/manager/orchestrator/jobs/global/reconciler.go @@ -35,7 +35,7 @@ func NewReconciler(store *store.MemoryStore, restart restartSupervisor) *Reconci } // ReconcileService reconciles one global job service. -func (r *Reconciler) ReconcileService(id string) error { +func (r *Reconciler) ReconcileService(ctx context.Context, id string) error { var ( service *api.Service cluster *api.Cluster @@ -199,9 +199,7 @@ func (r *Reconciler) ReconcileService(id string) error { } // Finally, restart it - // TODO(dperny): pass in context to ReconcileService, so we can - // pass it in here. - return r.restart.Restart(context.Background(), tx, cluster, service, *t) + return r.restart.Restart(ctx, tx, cluster, service, *t) }); err != nil { // TODO(dperny): probably should log like in the other // orchestrators instead of returning here. diff --git a/manager/orchestrator/jobs/global/reconciler_test.go b/manager/orchestrator/jobs/global/reconciler_test.go index 1c9cf4641d..cd9dc09060 100644 --- a/manager/orchestrator/jobs/global/reconciler_test.go +++ b/manager/orchestrator/jobs/global/reconciler_test.go @@ -157,7 +157,7 @@ var _ = Describe("Global Job Reconciler", func() { Expect(err).ToNot(HaveOccurred()) - err = r.ReconcileService(serviceID) + err = r.ReconcileService(context.Background(), serviceID) Expect(err).ToNot(HaveOccurred()) }) @@ -175,7 +175,7 @@ var _ = Describe("Global Job Reconciler", func() { }) Expect(err).ToNot(HaveOccurred()) - err = r.ReconcileService(serviceID) + err = r.ReconcileService(context.Background(), serviceID) Expect(err).ToNot(HaveOccurred()) s.View(func(tx store.ReadTx) { diff --git a/manager/orchestrator/jobs/orchestrator.go b/manager/orchestrator/jobs/orchestrator.go index 5d53e7019c..bebbe26b21 100644 --- a/manager/orchestrator/jobs/orchestrator.go +++ b/manager/orchestrator/jobs/orchestrator.go @@ -23,7 +23,7 @@ import ( type Reconciler interface { taskinit.InitHandler - ReconcileService(id string) error + ReconcileService(ctx context.Context, id string) error } // Orchestrator is the combined orchestrator controlling both Global and @@ -134,7 +134,7 @@ func (o *Orchestrator) init(ctx context.Context) { for _, service := range services { if orchestrator.IsReplicatedJob(service) { - if err := o.replicatedReconciler.ReconcileService(service.ID); err != nil { + if err := o.replicatedReconciler.ReconcileService(ctx, service.ID); err != nil { log.G(ctx).WithField( "service.id", service.ID, ).WithError(err).Error("error reconciling replicated job") @@ -142,7 +142,7 @@ func (o *Orchestrator) init(ctx context.Context) { } if orchestrator.IsGlobalJob(service) { - if err := o.globalReconciler.ReconcileService(service.ID); err != nil { + if err := o.globalReconciler.ReconcileService(ctx, service.ID); err != nil { log.G(ctx).WithField( "service.id", service.ID, ).WithError(err).Error("error reconciling global job") @@ -226,7 +226,7 @@ func (o *Orchestrator) handleEvent(ctx context.Context, event events.Event) { } if orchestrator.IsReplicatedJob(service) { - if err := o.replicatedReconciler.ReconcileService(service.ID); err != nil { + if err := o.replicatedReconciler.ReconcileService(ctx, service.ID); err != nil { log.G(ctx).WithField( "service.id", service.ID, ).WithError(err).Error("error reconciling replicated job") @@ -234,7 +234,7 @@ func (o *Orchestrator) handleEvent(ctx context.Context, event events.Event) { } if orchestrator.IsGlobalJob(service) { - if err := o.globalReconciler.ReconcileService(service.ID); err != nil { + if err := o.globalReconciler.ReconcileService(ctx, service.ID); err != nil { log.G(ctx).WithField( "service.id", service.ID, ).WithError(err).Error("error reconciling global job") diff --git a/manager/orchestrator/jobs/replicated/reconciler.go b/manager/orchestrator/jobs/replicated/reconciler.go index e3b0d5dc69..e50b4b80bd 100644 --- a/manager/orchestrator/jobs/replicated/reconciler.go +++ b/manager/orchestrator/jobs/replicated/reconciler.go @@ -2,9 +2,9 @@ package replicated import ( "context" - "fmt" "github.com/moby/swarmkit/v2/api" + "github.com/moby/swarmkit/v2/log" "github.com/moby/swarmkit/v2/manager/orchestrator" "github.com/moby/swarmkit/v2/manager/state/store" ) @@ -39,7 +39,7 @@ func NewReconciler(store *store.MemoryStore, restart restartSupervisor) *Reconci // checking to see if new replicas should be created. reconcileService returns // an error if there is some case prevent it from correctly reconciling the // service. -func (r *Reconciler) ReconcileService(id string) error { +func (r *Reconciler) ReconcileService(ctx context.Context, id string) error { var ( service *api.Service tasks []*api.Task @@ -157,31 +157,39 @@ func (r *Reconciler) ReconcileService(id string) error { rj := service.Spec.GetReplicatedJob() // possibleNewTasks gives us the upper bound for how many tasks we'll - // create. also, ugh, subtracting uints. there's no way this can ever go - // wrong. - possibleNewTasks := rj.MaxConcurrent - runningTasks + // create. subtractions here are saturating: if more tasks exist than the + // service asks for, we want to create none, not to underflow. + possibleNewTasks := subOrZero(rj.MaxConcurrent, runningTasks) // allowedNewTasks is how many tasks we could create, if there were no // restriction on maximum concurrency. This is the total number of tasks // we want completed, minus the tasks that are already completed, minus // the tasks that are in progress. - // - // seriously, ugh, subtracting unsigned ints. totally a fine and not at all - // risky operation, with no possibility for catastrophe - allowedNewTasks := rj.TotalCompletions - completeTasks - runningTasks + allowedNewTasks := subOrZero(subOrZero(rj.TotalCompletions, completeTasks), runningTasks) // the lower number of allowedNewTasks and possibleNewTasks is how many we // can create. actualNewTasks := min(possibleNewTasks, allowedNewTasks) - // this check might seem odd, but it protects us from an underflow of the - // above subtractions, which, again, is a totally impossible thing that can - // never happen, ever, obviously. - if actualNewTasks > rj.TotalCompletions { - return fmt.Errorf( - "uint64 underflow, we're not going to create %v tasks", - actualNewTasks, - ) + // a job that has overshot its TotalCompletions is not something we can + // undo, but it is also not a reason to stop reconciling: the removal of + // tasks belonging to older job iterations, and the restarting of failed + // tasks, both still need to happen. log it and carry on creating zero new + // tasks. + if completeTasks+runningTasks > rj.TotalCompletions { + log.G(ctx).WithFields(log.Fields{ + "service.id": service.ID, + "job.iteration": jobVersion, + "tasks.complete": completeTasks, + "tasks.running": runningTasks, + "totalCompletions": rj.TotalCompletions, + }).Warn("replicated job has more tasks than TotalCompletions; creating no new tasks") + } + + if completeTasks >= rj.TotalCompletions { + // The job has already reached its goal. Restarting any failed tasks + // now would risk overshooting the desired number of completions. + restartTasks = nil } // finally, we can create these tasks. do this in a batch operation, to @@ -235,8 +243,7 @@ func (r *Reconciler) ReconcileService(id string) error { return nil } - // TODO(dperny): pass in context from above - return r.restart.Restart(context.Background(), tx, cluster, service, *t) + return r.restart.Restart(ctx, tx, cluster, service, *t) }); err != nil { return err } @@ -290,3 +297,12 @@ func (r *Reconciler) SlotTuple(t *api.Task) orchestrator.SlotTuple { Slot: t.Slot, } } + +// subOrZero subtracts b from a, returning 0 rather than underflowing when b is +// greater than a. +func subOrZero(a, b uint64) uint64 { + if b > a { + return 0 + } + return a - b +} diff --git a/manager/orchestrator/jobs/replicated/reconciler_test.go b/manager/orchestrator/jobs/replicated/reconciler_test.go index e4d5e03c16..ba4b0ef9a5 100644 --- a/manager/orchestrator/jobs/replicated/reconciler_test.go +++ b/manager/orchestrator/jobs/replicated/reconciler_test.go @@ -153,7 +153,7 @@ var _ = Describe("Replicated Job reconciler", func() { }) Expect(err).ToNot(HaveOccurred()) - err = r.ReconcileService(serviceID) + err = r.ReconcileService(context.Background(), serviceID) Expect(err).ToNot(HaveOccurred()) // verify there are maxConcurrent tasks @@ -177,7 +177,7 @@ var _ = Describe("Replicated Job reconciler", func() { return store.UpdateService(tx, service) }) Expect(err).ToNot(HaveOccurred()) - err = r.ReconcileService(serviceID) + err = r.ReconcileService(context.Background(), serviceID) Expect(err).ToNot(HaveOccurred()) // fetch the tasks before we get to the test case itself, @@ -231,7 +231,7 @@ var _ = Describe("Replicated Job reconciler", func() { }) Expect(err).ToNot(HaveOccurred()) - reconcileErr = r.ReconcileService(serviceID) + reconcileErr = r.ReconcileService(context.Background(), serviceID) }) When("the job has no tasks yet created", func() { @@ -481,12 +481,15 @@ var _ = Describe("Replicated Job reconciler", func() { }) }) - It("should return an underflow error if there are more running tasks than TotalCompletions", func() { + It("should create no tasks, and no error, if there are more running tasks than TotalCompletions", func() { // this is an error condition which should not happen in real life, // but i want to make sure that we can't accidentally start - // creating nearly the maximum 64-bit unsigned int number of tasks. + // creating nearly the maximum 64-bit unsigned int number of tasks, + // and that overshooting TotalCompletions does not wedge + // reconciliation for the service permanently. maxConcurrent := uint64(10) totalCompletions := uint64(20) + var oldTaskID string err := s.Update(func(tx store.Tx) error { service := &api.Service{ ID: "someService", @@ -498,6 +501,9 @@ var _ = Describe("Replicated Job reconciler", func() { }, }, }, + JobStatus: &api.JobStatus{ + JobIteration: api.Version{Index: 1}, + }, } if err := store.CreateService(tx, service); err != nil { return err @@ -505,20 +511,88 @@ var _ = Describe("Replicated Job reconciler", func() { for range totalCompletions + 10 { task := orchestrator.NewTask(nil, service, 0, "") + task.JobIteration = &api.Version{Index: 1} + task.DesiredState = api.TaskStateCompleted + + if err := store.CreateTask(tx, task); err != nil { + return err + } + } + + // a task left over from the previous iteration of the job. the + // reconcile pass is expected to mark it for removal, which + // happens after the point where an overshoot used to return + // early. + oldTask := orchestrator.NewTask(nil, service, 0, "") + oldTask.JobIteration = &api.Version{Index: 0} + oldTask.DesiredState = api.TaskStateCompleted + oldTaskID = oldTask.ID + return store.CreateTask(tx, oldTask) + }) + Expect(err).ToNot(HaveOccurred()) + + reconcileErr := r.ReconcileService(context.Background(), "someService") + Expect(reconcileErr).ToNot(HaveOccurred()) + + s.View(func(tx store.ReadTx) { + tasks, err := store.FindTasks(tx, store.ByServiceID("someService")) + Expect(err).ToNot(HaveOccurred()) + Expect(tasks).To(HaveLen(int(totalCompletions + 10 + 1))) + + // the rest of the reconcile pass still runs: the task from the + // previous iteration is marked for removal. + oldTask := store.GetTask(tx, oldTaskID) + Expect(oldTask).ToNot(BeNil()) + Expect(oldTask.DesiredState).To(Equal(api.TaskStateRemove)) + }) + }) + + It("should not restart failed tasks once TotalCompletions has been reached", func() { + // restarting a failed task after the job has met its goal would + // push the number of completions past what was asked for. + maxConcurrent := uint64(1) + totalCompletions := uint64(2) + err := s.Update(func(tx store.Tx) error { + service := &api.Service{ + ID: "someService", + Spec: api.ServiceSpec{ + Mode: &api.ServiceSpec_ReplicatedJob{ + ReplicatedJob: &api.ReplicatedJob{ + MaxConcurrent: maxConcurrent, + TotalCompletions: totalCompletions, + }, + }, + }, + } + if err := store.CreateService(tx, service); err != nil { + return err + } + + for i := range totalCompletions { + task := orchestrator.NewTask(nil, service, i, "") task.JobIteration = &api.Version{} task.DesiredState = api.TaskStateCompleted + task.Status.State = api.TaskStateCompleted if err := store.CreateTask(tx, task); err != nil { return err } } - return nil + + // a surplus task that failed. the job is already done, so it + // must not be handed to the restart supervisor. + failed := orchestrator.NewTask(nil, service, totalCompletions, "") + failed.JobIteration = &api.Version{} + failed.DesiredState = api.TaskStateCompleted + failed.Status.State = api.TaskStateFailed + return store.CreateTask(tx, failed) }) Expect(err).ToNot(HaveOccurred()) - reconcileErr := r.ReconcileService("someService") - Expect(reconcileErr).To(HaveOccurred()) - Expect(reconcileErr.Error()).To(ContainSubstring("underflow")) + reconcileErr := r.ReconcileService(context.Background(), "someService") + Expect(reconcileErr).ToNot(HaveOccurred()) + + Expect(f.tasks).To(BeEmpty()) }) }) })