diff --git a/agent/reconcile_test.go b/agent/reconcile_test.go index 8e479445b..9d4fb38cc 100644 --- a/agent/reconcile_test.go +++ b/agent/reconcile_test.go @@ -275,48 +275,48 @@ func TestAbleToRun(t *testing.T) { tests := []struct { dState *AgentState job *job.Job - want bool + want job.JobAction }{ // nothing to worry about { dState: NewAgentState(&machine.MachineState{ID: "123"}), job: &job.Job{Name: "easy-street.service", Unit: unit.UnitFile{}}, - want: true, + want: job.JobActionSchedule, }, // match MachineID { dState: NewAgentState(&machine.MachineState{ID: "XYZ"}), job: newTestJobWithXFleetValues(t, "MachineID=XYZ"), - want: true, + want: job.JobActionSchedule, }, // mismatch MachineID { dState: NewAgentState(&machine.MachineState{ID: "123"}), job: newTestJobWithXFleetValues(t, "MachineID=XYZ"), - want: false, + want: job.JobActionUnschedule, }, // match MachineMetadata { dState: NewAgentState(&machine.MachineState{ID: "123", Metadata: map[string]string{"region": "us-west"}}), job: newTestJobWithXFleetValues(t, "MachineMetadata=region=us-west"), - want: true, + want: job.JobActionSchedule, }, // Machine metadata ignored when no MachineMetadata in Job { dState: NewAgentState(&machine.MachineState{ID: "123", Metadata: map[string]string{"region": "us-west"}}), job: &job.Job{Name: "easy-street.service", Unit: unit.UnitFile{}}, - want: true, + want: job.JobActionSchedule, }, // mismatch MachineMetadata { dState: NewAgentState(&machine.MachineState{ID: "123", Metadata: map[string]string{"region": "us-west"}}), job: newTestJobWithXFleetValues(t, "MachineMetadata=region=us-east"), - want: false, + want: job.JobActionUnschedule, }, // peer scheduled locally @@ -328,7 +328,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "MachineOf=pong.service"), - want: true, + want: job.JobActionSchedule, }, // multiple peers scheduled locally @@ -341,14 +341,14 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "MachineOf=pong.service\nMachineOf=ping.service"), - want: true, + want: job.JobActionSchedule, }, // peer not scheduled locally { dState: NewAgentState(&machine.MachineState{ID: "123"}), job: newTestJobWithXFleetValues(t, "MachineOf=ping.service"), - want: false, + want: job.JobActionUnschedule, }, // one of multiple peers not scheduled locally @@ -360,7 +360,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "MachineOf=pong.service\nMachineOf=ping.service"), - want: false, + want: job.JobActionUnschedule, }, // no conflicts found @@ -372,7 +372,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "Conflicts=pong.service"), - want: true, + want: job.JobActionSchedule, }, // conflicts found @@ -384,7 +384,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "Conflicts=ping.service"), - want: false, + want: job.JobActionUnschedule, }, // no replaces found @@ -396,7 +396,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "Replaces=pong.service"), - want: true, + want: job.JobActionSchedule, }, // replaces found @@ -408,7 +408,7 @@ func TestAbleToRun(t *testing.T) { }, }, job: newTestJobWithXFleetValues(t, "Replaces=ping.service"), - want: false, + want: job.JobActionReschedule, }, } diff --git a/agent/state.go b/agent/state.go index a1e5cd899..c254147fa 100644 --- a/agent/state.go +++ b/agent/state.go @@ -124,15 +124,15 @@ func globMatches(pattern, target string) bool { // - Agent must have all required Peers of the Job scheduled locally (if any) // - Job must not conflict with any other Units scheduled to the agent // - Job must specially handle replaced units to be rescheduled -func (as *AgentState) AbleToRun(j *job.Job) (bool, string) { +func (as *AgentState) AbleToRun(j *job.Job) (jobAction job.JobAction, errstr string) { if tgt, ok := j.RequiredTarget(); ok && !as.MState.MatchID(tgt) { - return false, fmt.Sprintf("agent ID %q does not match required %q", as.MState.ID, tgt) + return job.JobActionUnschedule, fmt.Sprintf("agent ID %q does not match required %q", as.MState.ID, tgt) } metadata := j.RequiredTargetMetadata() if len(metadata) != 0 { if !machine.HasMetadata(as.MState, metadata) { - return false, "local Machine metadata insufficient" + return job.JobActionUnschedule, "local Machine metadata insufficient" } } @@ -140,20 +140,27 @@ func (as *AgentState) AbleToRun(j *job.Job) (bool, string) { if len(peers) != 0 { for _, peer := range peers { if !as.unitScheduled(peer) { - return false, fmt.Sprintf("required peer Unit(%s) is not scheduled locally", peer) + return job.JobActionUnschedule, fmt.Sprintf("required peer Unit(%s) is not scheduled locally", peer) } } } if cExists, cJobName := as.HasConflict(j.Name, j.Conflicts()); cExists { - return false, fmt.Sprintf("found conflict with locally-scheduled Unit(%s)", cJobName) + return job.JobActionUnschedule, fmt.Sprintf("found conflict with locally-scheduled Unit(%s)", cJobName) } - // Handle Replace option specially, by returning a special string - // "jobreschedule" as reason. - if cExists, _ := as.hasReplace(j.Name, j.Replaces()); cExists { - return false, job.JobReschedule + // Handle Replace option specially for rescheduling the unit + if cExists, cJobName := as.hasReplace(j.Name, j.Replaces()); cExists { + return job.JobActionReschedule, fmt.Sprintf("found replace with locally-scheduled Unit(%s)", cJobName) } - return true, "" + return job.JobActionSchedule, "" +} + +func (as *AgentState) GetReplacedUnit(j *job.Job) (string, error) { + cExists, replaced := as.hasReplace(j.Name, j.Replaces()) + if !cExists { + return "", fmt.Errorf("cannot find units to be replaced for Unit(%s)", j.Name) + } + return replaced, nil } diff --git a/engine/reconciler.go b/engine/reconciler.go index 67472e6dc..3d98660cf 100644 --- a/engine/reconciler.go +++ b/engine/reconciler.go @@ -84,56 +84,111 @@ func (r *Reconciler) calculateClusterTasks(clust *clusterState, stopchan chan st return true } - go func() { - defer close(taskchan) + decide := func(j *job.Job) (jobAction job.JobAction, reason string) { + if j.TargetState == job.JobStateInactive { + return job.JobActionUnschedule, "target state inactive" + } agents := clust.agents() + as, ok := agents[j.TargetMachineID] + if !ok { + metrics.ReportEngineReconcileFailure(metrics.MachineAway) + return job.JobActionUnschedule, fmt.Sprintf("target Machine(%s) went away", j.TargetMachineID) + } + + if act, ableReason := as.AbleToRun(j); act != job.JobActionSchedule { + metrics.ReportEngineReconcileFailure(metrics.RunFailure) + return act, fmt.Sprintf("target Machine(%s) unable to run unit: %v", + j.TargetMachineID, ableReason) + } + + return job.JobActionSchedule, "" + } + + handle_reschedule := func(j *job.Job, reason string) bool { + isRescheduled := false + + agents := clust.agents() + + as, ok := agents[j.TargetMachineID] + if !ok { + metrics.ReportEngineReconcileFailure(metrics.MachineAway) + return false + } + + for _, cj := range clust.jobs { + if !cj.Scheduled() { + continue + } + if j.Name != cj.Name { + continue + } + + replacedUnit, err := as.GetReplacedUnit(j) + if err != nil { + log.Debugf("No unit to reschedule: %v", err) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) + continue + } + + if !send(taskTypeUnscheduleUnit, reason, replacedUnit, j.TargetMachineID) { + log.Infof("Job(%s) unschedule send failed", replacedUnit) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) + continue + } + + dec, err := r.sched.DecideReschedule(clust, j) + if err != nil { + log.Debugf("Unable to schedule Job(%s): %v", j.Name, err) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) + continue + } + + if !send(taskTypeAttemptScheduleUnit, reason, replacedUnit, dec.machineID) { + log.Infof("Job(%s) attemptschedule send failed", replacedUnit) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) + continue + } + clust.schedule(replacedUnit, dec.machineID) + log.Debugf("rescheduling unit %s to machine %s", replacedUnit, dec.machineID) + + clust.schedule(j.Name, j.TargetMachineID) + log.Debugf("scheduling unit %s to machine %s", j.Name, j.TargetMachineID) + + isRescheduled = true + } + + return isRescheduled + } + + go func() { + defer close(taskchan) + for _, j := range clust.jobs { if !j.Scheduled() { continue } - decide := func() (unschedule bool, reason string) { - if j.TargetState == job.JobStateInactive { - unschedule = true - reason = "target state inactive" - return - } - - as, ok := agents[j.TargetMachineID] - if !ok { - unschedule = true - reason = fmt.Sprintf("target Machine(%s) went away", j.TargetMachineID) - metrics.ReportEngineReconcileFailure(metrics.MachineAway) - return - } - - var able bool - var ableReason string - if able, ableReason = as.AbleToRun(j); !able { - unschedule = true - if ableReason == job.JobReschedule { - reason = ableReason - } else { - reason = fmt.Sprintf("target Machine(%s) unable to run unit", j.TargetMachineID) - metrics.ReportEngineReconcileFailure(metrics.RunFailure) - } - return - } - - return + act, reason := decide(j) + if act == job.JobActionReschedule && handle_reschedule(j, reason) { + log.Debugf("Job(%s) is rescheduled: %v", j.Name, reason) + continue } - unschedule, reason := decide() - if !unschedule { + if act != job.JobActionUnschedule { + log.Debugf("Job(%s) is not to be unscheduled, reason: %v", j.Name, reason) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) continue } if !send(taskTypeUnscheduleUnit, reason, j.Name, j.TargetMachineID) { + log.Infof("Job(%s) send failed.", j.Name) + metrics.ReportEngineReconcileFailure(metrics.ScheduleFailure) return } + log.Debugf("Job(%s) unscheduling.", j.Name) clust.unschedule(j.Name) } diff --git a/engine/scheduler.go b/engine/scheduler.go index 2b7119496..069e69b7c 100644 --- a/engine/scheduler.go +++ b/engine/scheduler.go @@ -28,6 +28,7 @@ type decision struct { type Scheduler interface { Decide(*clusterState, *job.Job) (*decision, error) + DecideReschedule(*clusterState, *job.Job) (*decision, error) } type leastLoadedScheduler struct{} @@ -41,7 +42,7 @@ func (lls *leastLoadedScheduler) Decide(clust *clusterState, j *job.Job) (*decis var target *agent.AgentState for _, as := range agents { - if able, _ := as.AbleToRun(j); !able { + if act, _ := as.AbleToRun(j); act == job.JobActionUnschedule { continue } @@ -61,6 +62,42 @@ func (lls *leastLoadedScheduler) Decide(clust *clusterState, j *job.Job) (*decis return &dec, nil } +// DecideReschedule() decides scheduling in a much simpler way than +// Decide(). It just tries to find out another free machine to be scheduled, +// except for the current target machine. It does not have to run +// as.AbleToRun(), because its job action must have been already decided +// before getting into the function. +func (lls *leastLoadedScheduler) DecideReschedule(clust *clusterState, j *job.Job) (*decision, error) { + agents := lls.sortedAgents(clust) + + if len(agents) == 0 { + return nil, fmt.Errorf("zero agents available") + } + + found := false + var target *agent.AgentState + for _, as := range agents { + if as.MState.ID == j.TargetMachineID { + continue + } + + as := as + target = as + found = true + break + } + + if !found { + return nil, fmt.Errorf("no agents able to run job") + } + + dec := decision{ + machineID: target.MState.ID, + } + + return &dec, nil +} + // sortedAgents returns a list of AgentState objects sorted ascending // by the number of scheduled units func (lls *leastLoadedScheduler) sortedAgents(clust *clusterState) []*agent.AgentState { diff --git a/functional/fixtures/units/replace-kick0.service b/functional/fixtures/units/replace-kick0.service new file mode 100644 index 000000000..5f1d5a448 --- /dev/null +++ b/functional/fixtures/units/replace-kick0.service @@ -0,0 +1,8 @@ +[Unit] +Description=Test Unit + +[Service] +ExecStart=/bin/bash -c "while true; do echo Hello, World!; sleep 1; done" + +[X-Fleet] +Replaces=replace.0.service diff --git a/functional/fixtures/units/replace.1.service b/functional/fixtures/units/replace.1.service index 5f1d5a448..94e06cf90 100644 --- a/functional/fixtures/units/replace.1.service +++ b/functional/fixtures/units/replace.1.service @@ -3,6 +3,3 @@ Description=Test Unit [Service] ExecStart=/bin/bash -c "while true; do echo Hello, World!; sleep 1; done" - -[X-Fleet] -Replaces=replace.0.service diff --git a/functional/fixtures/units/replace.2.service b/functional/fixtures/units/replace.2.service new file mode 100644 index 000000000..f8e9aa70c --- /dev/null +++ b/functional/fixtures/units/replace.2.service @@ -0,0 +1,8 @@ +[Unit] +Description=Test Unit + +[Service] +ExecStart=/bin/bash -c "while true; do echo Hello, World!; sleep 1; done" + +[X-Fleet] +MachineOf=replace.1.service diff --git a/functional/scheduling_test.go b/functional/scheduling_test.go index 46d29a01a..d29066484 100644 --- a/functional/scheduling_test.go +++ b/functional/scheduling_test.go @@ -333,9 +333,9 @@ func TestScheduleOneWayConflict(t *testing.T) { } -// TestScheduleReplace starts 1 unit, followed by starting another unit -// that replaces the 1st unit. Then it verifies that the 2 units are -// started on different machines. +// TestScheduleReplace starts 3 units, followed by starting another unit +// that replaces the 1st unit. Then it verifies that the original unit +// got rescheduled on a different machine. func TestScheduleReplace(t *testing.T) { cluster, err := platform.NewNspawnCluster("smoke") if err != nil { @@ -348,43 +348,30 @@ func TestScheduleReplace(t *testing.T) { t.Fatal(err) } m0 := members[0] + m1 := members[1] if _, err := cluster.WaitForNMachines(m0, 2); err != nil { t.Fatal(err) } - // Start a unit without Replaces + // Start 3 units without Replaces, replace.0.service on m0, while both 1 and 2 on m1. + // That's possible as replace.2.service has an option "MachineOf=replace.1.service". uNames := []string{ "fixtures/units/replace.0.service", "fixtures/units/replace.1.service", + "fixtures/units/replace.2.service", + "fixtures/units/replace-kick0.service", } if stdout, stderr, err := cluster.Fleetctl(m0, "start", "--no-block", uNames[0]); err != nil { t.Fatalf("Failed starting unit %s: \nstdout: %s\nstderr: %s\nerr: %v", uNames[0], stdout, stderr, err) } - - active, err := cluster.WaitForNActiveUnits(m0, 1) - if err != nil { - t.Fatal(err) - } - _, err = util.ActiveToSingleStates(active) - if err != nil { - t.Fatal(err) - } - - // Start a unit that replaces the former one, replace.0.service - if stdout, stderr, err := cluster.Fleetctl(m0, "start", "--no-block", uNames[1]); err != nil { + if stdout, stderr, err := cluster.Fleetctl(m1, "start", "--no-block", uNames[1]); err != nil { t.Fatalf("Failed starting unit %s: \nstdout: %s\nstderr: %s\nerr: %v", uNames[1], stdout, stderr, err) } - - // Check that both units should show up - stdout, stderr, err := cluster.Fleetctl(m0, "list-unit-files", "--no-legend") - if err != nil { - t.Fatalf("Failed to run list-unit-files:\nstdout: %s\nstderr: %s\nerr: %v", stdout, stderr, err) - } - units := strings.Split(strings.TrimSpace(stdout), "\n") - if len(units) != 2 { - t.Fatalf("Did not find two units in cluster: \n%s", stdout) + if stdout, stderr, err := cluster.Fleetctl(m1, "start", "--no-block", uNames[2]); err != nil { + t.Fatalf("Failed starting unit %s: \nstdout: %s\nstderr: %s\nerr: %v", uNames[2], stdout, stderr, err) } - active, err = cluster.WaitForNActiveUnits(m0, 2) + + active, err := cluster.WaitForNActiveUnits(m0, 3) if err != nil { t.Fatal(err) } @@ -393,16 +380,66 @@ func TestScheduleReplace(t *testing.T) { t.Fatal(err) } - // Check that the unit 1 is located on a different machine from that of unit 0 - nUnits := 2 - uNameBase := make([]string, nUnits) - machs := make([]string, nUnits) - for i, uName := range uNames { - uNameBase[i] = path.Base(uName) - machs[i] = states[uNameBase[i]].Machine + oldMach := states[path.Base(uNames[0])].Machine + + // Start a unit replace-kick0.service that replaces replace.0.service + // Then the kick0 unit will be scheduled to m0, as m0 is least loaded than m1. + // So it's possible to trigger a situation where kick0 could kick the original unit 0. + if stdout, stderr, err := cluster.Fleetctl(m0, "start", "--no-block", uNames[3]); err != nil { + t.Fatalf("Failed starting unit %s: \nstdout: %s\nstderr: %s\nerr: %v", uNames[3], stdout, stderr, err) } - if machs[0] == machs[1] { - t.Fatalf("machine for %s is %s, the same as that of %s.", uNameBase[0], machs[0], uNameBase[1]) + + // Here we need to wait up to 15 seconds, to avoid races, because the unit state + // publisher could otherwise report unit states with old machine IDs to registry. + checkReplacedMachines := func() bool { + // Check that 4 units show up + nUnits := 4 + stdout, stderr, err := cluster.Fleetctl(m0, "list-unit-files", "--no-legend") + if err != nil { + t.Logf("Failed to run list-unit-files:\nstdout: %s\nstderr: %s\nerr: %v", stdout, stderr, err) + return false + } + units := strings.Split(strings.TrimSpace(stdout), "\n") + if len(units) != nUnits { + t.Logf("Did not find two units in cluster: \n%s", stdout) + return false + } + active, err = cluster.WaitForNActiveUnits(m0, nUnits) + if err != nil { + t.Log(err) + return false + } + states, err = util.ActiveToSingleStates(active) + if err != nil { + t.Log(err) + return false + } + + // Check that replace.0.service is located on a different machine from + // that of replace-kick0.service. + uNameBase := make([]string, nUnits) + machs := make([]string, nUnits) + for i, uName := range uNames { + uNameBase[i] = path.Base(uName) + machs[i] = states[uNameBase[i]].Machine + } + if machs[0] == machs[3] { + t.Logf("machine for %s is %s, the same as that of %s.", uNameBase[0], machs[0], uNameBase[3]) + return false + } + if machs[3] != oldMach { + t.Logf("machine for %s is %s, different from old machine %s.", uNameBase[3], machs[3], oldMach) + return false + } + if machs[0] == oldMach { + t.Logf("machine for %s is %s, the same as that of %s.", uNameBase[0], machs[0], oldMach) + return false + } + + return true + } + if timeout, err := util.WaitForState(checkReplacedMachines); err != nil { + t.Fatalf("Cannot verify replaced units within %v\nerr: %v", timeout, err) } } @@ -429,7 +466,7 @@ func TestScheduleCircularReplace(t *testing.T) { // it under /tmp. Also store the original service 1 that replace 0. uNames := []string{ "fixtures/units/replace.0.service", - "fixtures/units/replace.1.service", + "fixtures/units/replace-kick0.service", } nUnits := 2 nActiveUnits := 1 @@ -439,12 +476,12 @@ func TestScheduleCircularReplace(t *testing.T) { } uName0tmp := path.Join("/tmp", uNameBase[0]) err = util.GenNewFleetService(uName0tmp, uNames[1], - "Replaces=replace.1.service", "Replaces=replace.0.service") + "Replaces=replace-kick0.service", "Replaces=replace.0.service") if err != nil { t.Fatalf("Failed to generate a temp fleet service: %v", err) } - // Start replace.0 unit that replaces replace.1.service, + // Start replace.0 unit that replaces replace-kick0.service, // then fleetctl list-unit-files should show only return 1 launched unit. stdout, stderr, err := cluster.Fleetctl(m0, "start", "--no-block", uName0tmp) if err != nil { @@ -469,7 +506,7 @@ func TestScheduleCircularReplace(t *testing.T) { t.Fatalf("Failed to run list-unit-files: %v", err) } - // Start replace.1 unit that replaces replace.0.service, + // Start replace-kick0 unit that replaces replace.0.service, // and then check that only 1 unit is active if stdout, stderr, err := cluster.Fleetctl(m0, "start", "--no-block", uNames[1]); err != nil { t.Fatalf("Failed starting unit %s: \nstdout: %s\nstderr: %s\nerr: %v", uNames[1], stdout, stderr, err) diff --git a/job/job.go b/job/job.go index 907f14a59..1ed001f14 100644 --- a/job/job.go +++ b/job/job.go @@ -23,13 +23,16 @@ import ( ) type JobState string +type JobAction string const ( JobStateInactive = JobState("inactive") JobStateLoaded = JobState("loaded") JobStateLaunched = JobState("launched") - JobReschedule = "jobreschedule" + JobActionSchedule = JobAction("job_action_schedule") + JobActionUnschedule = JobAction("job_action_unschedule") + JobActionReschedule = JobAction("job_action_reschedule") ) // fleet-specific unit file requirement keys.