From ec664a4c257b5f8bf7142693e76715c0510e45f7 Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Thu, 3 Nov 2016 18:48:15 +0100 Subject: [PATCH 1/4] agent,engine: use tri-state job schedule actions So far decide() returns a bool variable to determine whether it's possible to schedule or not. Reschedule was not handled in a proper way. So let's create a new tri-state for job.JobAction, i.e. JobActionSchedule, JobActionUnschedule, and JobActionReschedule. And return the tri-state variable instead of bool. That way AbleToRun() can check return values from HasConflict() and HasReplace() in a correct manner. --- agent/state.go | 19 +++++++------- engine/reconciler.go | 59 ++++++++++++++++++-------------------------- engine/scheduler.go | 2 +- job/job.go | 5 +++- 4 files changed, 38 insertions(+), 47 deletions(-) diff --git a/agent/state.go b/agent/state.go index a1e5cd899..48d3d7fe5 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,19 @@ 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, "" } diff --git a/engine/reconciler.go b/engine/reconciler.go index 67472e6dc..958dbf5b8 100644 --- a/engine/reconciler.go +++ b/engine/reconciler.go @@ -84,49 +84,38 @@ 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, "" + } + + 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 - } - - unschedule, reason := decide() - if !unschedule { + act, reason := decide(j) + if act != job.JobActionUnschedule { continue } diff --git a/engine/scheduler.go b/engine/scheduler.go index 2b7119496..401c77ae7 100644 --- a/engine/scheduler.go +++ b/engine/scheduler.go @@ -41,7 +41,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.JobActionSchedule { continue } 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. From 13ba083afd829365efae2111dd1289858742b13f Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Fri, 4 Nov 2016 10:53:33 +0100 Subject: [PATCH 2/4] engine: fix bugs in rescheduling for replaced units While the 'Replaces' option has been supported since https://github.com/coreos/fleet/pull/1572, the engine didn't actually unschedule units to be replaced. It was a bug. So let's implement GetReplacedUnit() to expose the replaced unit from AgentState to the engine reconciler. And make the engine reconciler unschedule the replaced unit, and schedule the current unit. The engine scheduler's decision structure needs to have a helper for the rescheduling case, by simply scheduling the replaced unit to a free machine. --- agent/state.go | 8 ++++++ engine/reconciler.go | 66 ++++++++++++++++++++++++++++++++++++++++++++ engine/scheduler.go | 39 +++++++++++++++++++++++++- 3 files changed, 112 insertions(+), 1 deletion(-) diff --git a/agent/state.go b/agent/state.go index 48d3d7fe5..c254147fa 100644 --- a/agent/state.go +++ b/agent/state.go @@ -156,3 +156,11 @@ func (as *AgentState) AbleToRun(j *job.Job) (jobAction job.JobAction, errstr str 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 958dbf5b8..3d98660cf 100644 --- a/engine/reconciler.go +++ b/engine/reconciler.go @@ -106,6 +106,62 @@ func (r *Reconciler) calculateClusterTasks(clust *clusterState, stopchan chan st 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) @@ -115,14 +171,24 @@ func (r *Reconciler) calculateClusterTasks(clust *clusterState, stopchan chan st } act, reason := decide(j) + if act == job.JobActionReschedule && handle_reschedule(j, reason) { + log.Debugf("Job(%s) is rescheduled: %v", j.Name, reason) + continue + } + 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 401c77ae7..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 act, _ := as.AbleToRun(j); act != job.JobActionSchedule { + 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 { From 3df0ec2b5f29ee9e1dc0788eff78f21766f71e7b Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Fri, 4 Nov 2016 10:54:59 +0100 Subject: [PATCH 3/4] agent: update unit tests in agent reconciler_test Now that the AbleToRun() returns JobAction instead of bool, unit tests also need to be changed. --- agent/reconcile_test.go | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) 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, }, } From f382b135b7f5c100d43154fffc0d4af46eb231ce Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Fri, 4 Nov 2016 10:55:02 +0100 Subject: [PATCH 4/4] functional: make TestScheduleReplace() do correct tests For correct tests, TestScheduleReplace() now does the following tests. * Start 4 units. replace.0 on m0, replace.[12] on m1, and replace-kick0 on m0. Doing that we could trigger a situation of unit replacement. * Ensure that machine of kick0 unit is not the same as that of unit 0, also that machine of kick0 is the same as the original machine m0. * Make use of WaitForState() to avoid races with unit state publisher. --- .../fixtures/units/replace-kick0.service | 8 ++ functional/fixtures/units/replace.1.service | 3 - functional/fixtures/units/replace.2.service | 8 ++ functional/scheduling_test.go | 115 ++++++++++++------ 4 files changed, 92 insertions(+), 42 deletions(-) create mode 100644 functional/fixtures/units/replace-kick0.service create mode 100644 functional/fixtures/units/replace.2.service 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)