diff --git a/internal/worker/vmmanager/base/base.go b/internal/worker/vmmanager/base/base.go index e896edf..3c9f4d0 100644 --- a/internal/worker/vmmanager/base/base.go +++ b/internal/worker/vmmanager/base/base.go @@ -122,6 +122,15 @@ func (vm *VM) ConditionsSet() mapset.Set[v1.ConditionType] { return vm.conditions } +// Running reports whether the VM is running and neither stopping nor suspending. +func (vm *VM) Running() bool { + // Snapshot the flags together so shutdown cannot mix old and new conditions + conditions := vm.conditions.Clone() + + return conditions.ContainsOne(v1.ConditionTypeRunning) && + !conditions.ContainsAny(v1.ConditionTypeStopping, v1.ConditionTypeSuspending) +} + func (vm *VM) Conditions() []v1.Condition { // The worker must observe transitions before applying a new specification. return []v1.Condition{ diff --git a/internal/worker/vmmanager/vmmanager.go b/internal/worker/vmmanager/vmmanager.go index a2ed664..284a5da 100644 --- a/internal/worker/vmmanager/vmmanager.go +++ b/internal/worker/vmmanager/vmmanager.go @@ -24,6 +24,7 @@ type VM interface { StatusMessage() string Err() error Conditions() []v1.Condition + Running() bool Start(eventStreamer *client.EventStreamer) Suspend() <-chan error diff --git a/internal/worker/worker.go b/internal/worker/worker.go index c9fe016..a212648 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -601,6 +601,7 @@ func (worker *Worker) syncVMs( currentVMResource := vm.Resource() if worker.softnetPolicyUpdates.OrElse(false) && + vm.Running() && currentVMResource.SoftnetEnabled() && vmResource.SoftnetEnabled() && currentVMResource.SoftnetPolicyChanged(vmResource.VMSpec) { if err := vm.UpdateSoftnetPolicy(ctx, @@ -687,10 +688,11 @@ func (worker *Worker) monitorRunningVM( // Try to apply the host process updates in-place or recover processes that exited unexpectedly hostProcessesChanged := !currentVMResource.VMSpec.HostProcessesEqual(vmResource.VMSpec) hostProcessesNeedRestart := currentVMResource.Generation == vmResource.Generation && - v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) && !vm.HostProcessSet().Ready() - if vmResource.PowerState == v1.PowerStateRunning && (hostProcessesChanged || hostProcessesNeedRestart) { + if vmResource.PowerState == v1.PowerStateRunning && + vm.Running() && + (hostProcessesChanged || hostProcessesNeedRestart) { if err := vm.HostProcessSet().Replace(ctx, vmResource.HostProcesses); err != nil { worker.logger.Warnf("failed to update host processes for VM %q: %v", vmResource.Name, err) @@ -709,9 +711,7 @@ func (worker *Worker) monitorRunningVM( ) if vmResource.PowerState == v1.PowerStateRunning && - !v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping) && - !v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeSuspending) && - v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) && + vm.Running() && endpointsChanged { currentVMResource.Endpoints = vmResource.Endpoints appliedInPlace = true diff --git a/internal/worker/worker_stop_test.go b/internal/worker/worker_stop_test.go index 58ba2eb..1a0198f 100644 --- a/internal/worker/worker_stop_test.go +++ b/internal/worker/worker_stop_test.go @@ -41,6 +41,12 @@ func (vm *delayedStopVM) Status() v1.VMStatus { return vm.status } func (vm *delayedStopVM) Conditions() []v1.Condition { return vm.conditions } +func (vm *delayedStopVM) Running() bool { + return v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeRunning) && + !v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeStopping) && + !v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeSuspending) +} + func (vm *delayedStopVM) Resource() v1.VM { return vm.resource } func (vm *delayedStopVM) SetResource(resource v1.VM) { @@ -130,6 +136,57 @@ func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) { } } +func TestMonitorPreservesRestartAfterHostProcessUpdate(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, runtime: runtime.NewSynthetic()} + update := func(context.Context, v1.VM) error { return nil } + + // An earlier specification change has already initiated a restart + vm := &delayedStopVM{ + VM: &synthetic.VM{VM: base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, zap.NewNop().Sugar())}, + resource: v1.VM{ + Name: "test-vm", + PowerState: v1.PowerStateRunning, + HostProcesses: []v1.HostProcess{{Name: "removed", Program: "unused"}}, + }, + conditions: []v1.Condition{ + {Type: v1.ConditionTypeRunning, State: v1.ConditionStateFalse}, + {Type: v1.ConditionTypeStopping, State: v1.ConditionStateTrue}, + }, + } + desired := vm.resource + desired.Generation = 2 + desired.HostProcesses = nil + + // Removing the host process must leave the generation pending while stopping + require.NoError(t, worker.monitorRunningVM(t.Context(), &desired, vm, update)) + require.Zero(t, vm.resource.Generation) + require.Zero(t, desired.ObservedGeneration) + require.Zero(t, vm.starts) + + // After stopping, the pending generation must still cause the VM to start + 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) + require.Equal(t, 1, vm.starts) + require.Empty(t, vm.resource.HostProcesses) + select { + case <-events: + case <-time.After(time.Second): + t.Fatal("restart event stream did not close") + } +} + func TestSyncVMsWaitsForVMShutdown(t *testing.T) { tests := []struct { name string diff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go index 74bdb6f..54c5b68 100644 --- a/internal/worker/worker_test.go +++ b/internal/worker/worker_test.go @@ -927,6 +927,8 @@ func (vm *recoveryTestVM) Conditions() []v1.Condition { return nil } +func (vm *recoveryTestVM) Running() bool { return false } + func (vm *recoveryTestVM) Stop() <-chan error { vm.stopped.Store(true)