diff --git a/internal/worker/vmmanager/base/base.go b/internal/worker/vmmanager/base/base.go index 46f4f89..5ad39fb 100644 --- a/internal/worker/vmmanager/base/base.go +++ b/internal/worker/vmmanager/base/base.go @@ -103,10 +103,11 @@ func (vm *VM) ConditionsSet() mapset.Set[v1.ConditionType] { } func (vm *VM) Conditions() []v1.Condition { - // Only expose a minimum amount of conditions necessary - // for the Orchard Controller to make decisions + // The worker must observe transitions before applying a new specification. return []v1.Condition{ vm.conditionTypeToCondition(v1.ConditionTypeRunning), + vm.conditionTypeToCondition(v1.ConditionTypeSuspending), + vm.conditionTypeToCondition(v1.ConditionTypeStopping), } } diff --git a/internal/worker/vmmanager/stop_test.go b/internal/worker/vmmanager/stop_test.go new file mode 100644 index 0000000..bfb825d --- /dev/null +++ b/internal/worker/vmmanager/stop_test.go @@ -0,0 +1,199 @@ +package vmmanager_test + +import ( + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/cirruslabs/orchard/internal/worker/vmmanager" + "github.com/cirruslabs/orchard/internal/worker/vmmanager/tart" + "github.com/cirruslabs/orchard/internal/worker/vmmanager/vetu" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +const stopTestTimeout = 15 * time.Second + +func TestStopWaitsForCommandAndRun(t *testing.T) { + runtimes := []struct { + name string + new func(v1.VM) vmmanager.VM + }{ + {name: "tart", new: func(resource v1.VM) vmmanager.VM { + return tart.NewVM(resource, nil, nil, nil, zap.NewNop().Sugar()) + }}, + {name: "vetu", new: func(resource v1.VM) vmmanager.VM { + return vetu.NewVM(resource, nil, nil, nil, zap.NewNop().Sugar()) + }}, + } + + for _, runtime := range runtimes { + for _, commandFirst := range []bool{true, false} { + order := "run finishes first" + if commandFirst { + order = "stop command finishes first" + } + t.Run(runtime.name+"/"+order, func(t *testing.T) { + checkStopCompletion(t, runtime.name, runtime.new, commandFirst) + }) + } + } +} + +//nolint:exhaustruct_v5 // VM settings unrelated to lifecycle transitions use their defaults. +func checkStopCompletion(t *testing.T, commandName string, newVM func(v1.VM) vmmanager.VM, commandFirst bool) { + t.Helper() + + dir := installStopTestCommand(t, commandName) + runGate := filepath.Join(dir, "run.release") + stopGate := filepath.Join(dir, "stop.release") + release := func(path string) { require.NoError(t, os.WriteFile(path, nil, 0o600)) } + waitForFile := func(path string) { + require.Eventually(t, func() bool { + _, err := os.Stat(path) + return err == nil + }, stopTestTimeout, 5*time.Millisecond, "command did not reach %s", path) + } + waitForStop := func(done <-chan error) { + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(stopTestTimeout): + t.Fatal("Stop did not complete after the commands finished") + } + } + + resource := v1.VM{Name: "stop-test", UID: "test-uid", Image: "source-image"} + vm := newVM(resource) + var first <-chan error + t.Cleanup(func() { + if t.Failed() { + t.Logf("VM status before fixture cleanup: %s; error: %v", vm.StatusMessage(), vm.Err()) + } + release(runGate) + release(stopGate) + if first != nil { + waitForStop(first) + } + waitForStop(vm.Stop()) + }) + + // A restart must create a new Stop completion for the new run. + for run := range 2 { + if run != 0 { + for _, name := range []string{"run.release", "stop.release", "run.started", "stop.started", "stop.finished"} { + require.NoError(t, os.Remove(filepath.Join(dir, name))) + } + vm.Start(nil) + } + waitForFile(filepath.Join(dir, "run.started")) + first = vm.Stop() + waitForFile(filepath.Join(dir, "stop.started")) + + if commandFirst { + release(stopGate) + waitForFile(filepath.Join(dir, "stop.finished")) + pidText, err := os.ReadFile(filepath.Join(dir, "run.started")) //nolint:gosec // Read a t.TempDir fixture. + require.NoError(t, err) + pid, err := strconv.Atoi(strings.TrimSpace(string(pidText))) + require.NoError(t, err) + // Cancellation kills the run process, but its child still holds + // stdout open, so Cmd.Run and the VM goroutine cannot finish yet. + require.Eventually(t, func() bool { + return errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) + }, stopTestTimeout, 5*time.Millisecond, "Stop did not cancel the run process") + } else { + release(runGate) + require.Eventually(t, func() bool { + return v1.ConditionIsFalse(vm.Conditions(), v1.ConditionTypeRunning) + }, stopTestTimeout, 5*time.Millisecond, "the run did not finish") + } + + callers := make(chan (<-chan error), 8) + for range cap(callers) { + go func() { callers <- vm.Stop() }() + } + stops := []<-chan error{first} + for range cap(callers) { + select { + case done := <-callers: + stops = append(stops, done) + case <-time.After(stopTestTimeout): + t.Fatal("concurrent Stop call did not return its completion channel") + } + } + for _, done := range stops { + select { + case <-done: + t.Fatal("Stop completed while shutdown was still in progress") + default: + } + } + // Reconciliation must still see Stopping when the run has ended + // but its stop command has not returned. + require.True(t, v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping)) + + release(runGate) + release(stopGate) + for _, done := range stops { + waitForStop(done) + } + waitForStop(vm.Stop()) + require.False(t, v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning)) + require.False(t, v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping)) + require.NoError(t, vm.Err()) + } + commands, err := os.ReadFile(filepath.Join(dir, "commands.log")) //nolint:gosec // Read a t.TempDir fixture. + require.NoError(t, err) + require.Equal(t, 2, strings.Count(string(commands), "stop --timeout 5 "), + "concurrent callers must share one stop command per run") +} + +func installStopTestCommand(t *testing.T, name string) string { + t.Helper() + + dir := t.TempDir() + const script = `#!/bin/sh +set -eu +wait_for_file() { + attempts=0 + while [ ! -f "$1" ]; do + attempts=$((attempts + 1)) + if [ "$attempts" -ge "$ORCHARD_STOP_TEST_WAIT_ATTEMPTS" ]; then exit 1; fi + /bin/sleep 0.01 + done +} +printf '%s\n' "$*" >> "$ORCHARD_STOP_TEST_DIR/commands.log" +case "$1" in + get) + printf '%s\n' '{"Running":false,"State":"stopped"}' + ;; + run) + # Keep the child's output pipes open after cancellation kills this shell. + wait_for_file "$ORCHARD_STOP_TEST_DIR/run.release" & + printf '%s\n' "$$" > "$ORCHARD_STOP_TEST_DIR/run.started" + wait + ;; + stop) + : > "$ORCHARD_STOP_TEST_DIR/stop.started" + wait_for_file "$ORCHARD_STOP_TEST_DIR/stop.release" + : > "$ORCHARD_STOP_TEST_DIR/stop.finished" + ;; +esac +` + commandPath := filepath.Join(dir, name) + require.NoError(t, os.WriteFile(commandPath, []byte(script), 0o600)) + require.NoError(t, os.Chmod(commandPath, 0o700)) //nolint:gosec // Fake commands must be executable. + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("ORCHARD_STOP_TEST_DIR", dir) + // A command can wait through several Go-side milestones before its gate opens. + // Keep it bounded, with enough time for every milestone under process load. + t.Setenv("ORCHARD_STOP_TEST_WAIT_ATTEMPTS", strconv.Itoa(int(8*stopTestTimeout/(10*time.Millisecond)))) + return dir +} diff --git a/internal/worker/vmmanager/tart/tart.go b/internal/worker/vmmanager/tart/tart.go index 15d0c31..d2d6783 100644 --- a/internal/worker/vmmanager/tart/tart.go +++ b/internal/worker/vmmanager/tart/tart.go @@ -32,6 +32,9 @@ type VM struct { wg *sync.WaitGroup + stopMtx sync.Mutex + stopDone chan error + dialer dialer.Dialer *base.VM @@ -267,7 +270,8 @@ func (vm *VM) cloneAndConfigure(ctx context.Context) error { } func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { - defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending, v1.ConditionTypeStopping) + // Stop owns Stopping until both its command and this goroutine finish. + defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending) // Launch the startup script goroutine as close as possible // to the VM startup (below) to avoid "tart ip" timing out @@ -398,37 +402,44 @@ func (vm *VM) Suspend() <-chan error { } func (vm *VM) Stop() <-chan error { - errCh := make(chan error, 1) - - select { - case <-vm.ctx.Done(): - // VM is already suspended/stopped - errCh <- nil - - return errCh - default: - // VM is still running + vm.stopMtx.Lock() + defer vm.stopMtx.Unlock() + if vm.stopDone != nil { + return vm.stopDone } + done := make(chan error) + vm.stopDone = done + ctx, cancel, wg := vm.ctx, vm.cancel, vm.wg vm.SetStatusMessage("Stopping VM") vm.ConditionsSet().Add(v1.ConditionTypeStopping) go func() { - // Try to gracefully terminate the VM - _, _, _ = Tart(context.Background(), zap.NewNop().Sugar(), "stop", "--timeout", "5", vm.id()) + if ctx.Err() == nil { + // Try to gracefully terminate the VM. + _, _, _ = Tart(context.WithoutCancel(ctx), zap.NewNop().Sugar(), "stop", "--timeout", "5", vm.id()) + } - // Terminate the VM goroutine ("tart pull", "tart clone", "tart run", etc.) via the context - vm.cancel() - vm.wg.Wait() + // Cancellation requests shutdown; it does not establish completion. + cancel() + wg.Wait() - // We don't return an error because we always terminate a VM - errCh <- nil + vm.stopMtx.Lock() + vm.ConditionsSet().Remove(v1.ConditionTypeStopping) + // Closing broadcasts successful completion to every Stop caller. + close(done) + vm.stopMtx.Unlock() }() - return errCh + return done } func (vm *VM) Start(eventStreamer *client.EventStreamer) { + // The worker defers Start while Stopping is true. + vm.stopMtx.Lock() + defer vm.stopMtx.Unlock() + vm.stopDone = nil + vm.SetStatusMessage("Starting VM") vm.ConditionsSet().Add(v1.ConditionTypeRunning) diff --git a/internal/worker/vmmanager/vetu/vetu.go b/internal/worker/vmmanager/vetu/vetu.go index f22e3e6..d5f79cf 100644 --- a/internal/worker/vmmanager/vetu/vetu.go +++ b/internal/worker/vmmanager/vetu/vetu.go @@ -32,6 +32,9 @@ type VM struct { wg *sync.WaitGroup + stopMtx sync.Mutex + stopDone chan error + dialer dialer.Dialer *base.VM @@ -197,7 +200,8 @@ func (vm *VM) cloneAndConfigure(ctx context.Context) error { } func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { - defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending, v1.ConditionTypeStopping) + // Stop owns Stopping until both its command and this goroutine finish. + defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending) // Launch the startup script goroutine as close as possible // to the VM startup (below) to avoid "vetu ip" timing out @@ -257,37 +261,44 @@ func (vm *VM) Suspend() <-chan error { } func (vm *VM) Stop() <-chan error { - errCh := make(chan error, 1) - - select { - case <-vm.ctx.Done(): - // VM is already suspended/stopped - errCh <- nil - - return errCh - default: - // VM is still running + vm.stopMtx.Lock() + defer vm.stopMtx.Unlock() + if vm.stopDone != nil { + return vm.stopDone } + done := make(chan error) + vm.stopDone = done + ctx, cancel, wg := vm.ctx, vm.cancel, vm.wg vm.SetStatusMessage("Stopping VM") vm.ConditionsSet().Add(v1.ConditionTypeStopping) go func() { - // Try to gracefully terminate the VM - _, _, _ = Vetu(context.Background(), zap.NewNop().Sugar(), "stop", "--timeout", "5", vm.id()) + if ctx.Err() == nil { + // Try to gracefully terminate the VM. + _, _, _ = Vetu(context.WithoutCancel(ctx), zap.NewNop().Sugar(), "stop", "--timeout", "5", vm.id()) + } - // Terminate the VM goroutine ("vetu pull", "vetu clone", "vetu run", etc.) via the context - vm.cancel() - vm.wg.Wait() + // Cancellation requests shutdown; it does not establish completion. + cancel() + wg.Wait() - // We don't return an error because we always terminate a VM - errCh <- nil + vm.stopMtx.Lock() + vm.ConditionsSet().Remove(v1.ConditionTypeStopping) + // Closing broadcasts successful completion to every Stop caller. + close(done) + vm.stopMtx.Unlock() }() - return errCh + return done } func (vm *VM) Start(eventStreamer *client.EventStreamer) { + // The worker defers Start while Stopping is true. + vm.stopMtx.Lock() + defer vm.stopMtx.Unlock() + vm.stopDone = nil + vm.SetStatusMessage("Starting VM") vm.ConditionsSet().Add(v1.ConditionTypeRunning) diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 7a9a8a3..fda971e 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -673,6 +673,7 @@ func (worker *Worker) reconcileRunningVM(vmResource *v1.VM, vm vmmanager.VM) { if shouldStop { vm.Stop() + return } else { vm.Suspend() } diff --git a/internal/worker/worker_stop_test.go b/internal/worker/worker_stop_test.go index 671a922..9619c95 100644 --- a/internal/worker/worker_stop_test.go +++ b/internal/worker/worker_stop_test.go @@ -26,6 +26,8 @@ type delayedStopVM struct { status v1.VMStatus stopStarted chan struct{} stopResult chan error + conditions []v1.Condition + starts int } func (vm *delayedStopVM) OnDiskName() ondiskname.OnDiskName { @@ -34,15 +36,96 @@ func (vm *delayedStopVM) OnDiskName() ondiskname.OnDiskName { func (vm *delayedStopVM) Status() v1.VMStatus { return vm.status } -func (vm *delayedStopVM) Conditions() []v1.Condition { return nil } +func (vm *delayedStopVM) Conditions() []v1.Condition { return vm.conditions } + +func (vm *delayedStopVM) Resource() v1.VM { return vm.resource } + +func (vm *delayedStopVM) SetResource(resource v1.VM) { + vm.resource = resource + vm.resource.ObservedGeneration = resource.Generation +} + +func (vm *delayedStopVM) StatusMessage() string { return "" } + +func (vm *delayedStopVM) Start(streamer *client.EventStreamer) { + vm.starts++ + v1.ConditionsSet(&vm.conditions, v1.Condition{ + Type: v1.ConditionTypeRunning, State: v1.ConditionStateTrue, + }) + _ = streamer.Close() +} func (vm *delayedStopVM) Err() error { return errLocalVMFailed } func (vm *delayedStopVM) Stop() <-chan error { close(vm.stopStarted) + if vm.conditions != nil { + v1.ConditionsSet(&vm.conditions, v1.Condition{ + Type: v1.ConditionTypeRunning, State: v1.ConditionStateFalse, + }) + v1.ConditionsSet(&vm.conditions, v1.Condition{ + Type: v1.ConditionTypeStopping, State: v1.ConditionStateTrue, + }) + } return vm.stopResult } +//nolint:exhaustruct_v5 // Fixture fields unrelated to lifecycle transitions use their zero values. +func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) { + for _, powerState := range []v1.PowerState{v1.PowerStateStopped, v1.PowerStateRunning} { + t.Run(string(powerState), func(t *testing.T) { + events := make(chan struct{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.WriteHeader(http.StatusOK) + events <- struct{}{} + })) + defer server.Close() + apiClient, err := client.New(client.WithAddress(server.URL)) + require.NoError(t, err) + worker := &Worker{client: apiClient} + vm := &delayedStopVM{ + resource: v1.VM{Meta: v1.Meta{Name: "test-vm"}}, + stopStarted: make(chan struct{}), + stopResult: make(chan error), + conditions: []v1.Condition{{ + Type: v1.ConditionTypeRunning, State: v1.ConditionStateTrue, + }}, + } + desired := vm.resource + desired.Generation = 1 + desired.PowerState = powerState + update := func(context.Context, v1.VM) error { return nil } + + // Stop clears Running immediately, before the command has finished. + // Neither this reconciliation nor later ones may acknowledge or restart it. + for range 2 { + require.NoError(t, worker.monitorRunningVM(t.Context(), &desired, vm, update)) + require.Zero(t, vm.resource.Generation, "the specification changed before Stop completed") + require.Zero(t, desired.ObservedGeneration, "shutdown was acknowledged before it completed") + require.Zero(t, vm.starts, "the VM restarted before Stop completed") + } + + close(vm.stopResult) + v1.ConditionsSet(&vm.conditions, v1.Condition{ + Type: v1.ConditionTypeStopping, State: v1.ConditionStateFalse, + }) + require.NoError(t, worker.monitorRunningVM(t.Context(), &desired, vm, update)) + require.Equal(t, desired.Generation, vm.resource.Generation) + require.Equal(t, desired.Generation, desired.ObservedGeneration) + if powerState == v1.PowerStateRunning { + require.Equal(t, 1, vm.starts) + select { + case <-events: + case <-time.After(time.Second): + t.Fatal("restart event stream did not close") + } + } else { + require.Zero(t, vm.starts) + } + }) + } +} + func TestSyncVMsWaitsForVMShutdown(t *testing.T) { tests := []struct { name string