mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-21 11:46:37 +00:00
refactor(vm): typed Status accessor, sentinel-aware Stop/Terminate (#191)
This commit is contained in:
@@ -347,8 +347,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
|
||||
var successCount int
|
||||
for _, instance := range instances {
|
||||
// Skip if instance was terminated by a concurrent request
|
||||
var status vm.InstanceState
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
|
||||
status := d.vmMgr.Status(instance)
|
||||
if status != vm.StatePending && status != vm.StateProvisioning {
|
||||
slog.Info("Instance state changed during provisioning, skipping launch",
|
||||
"instanceId", instance.ID, "status", string(status))
|
||||
@@ -429,8 +428,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom
|
||||
slog.Info("Starting instance", "id", command.ID)
|
||||
|
||||
// Validate instance is in stopped state
|
||||
var status vm.InstanceState
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
|
||||
status := d.vmMgr.Status(instance)
|
||||
|
||||
if status != vm.StateStopped {
|
||||
slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status)
|
||||
@@ -451,7 +449,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom
|
||||
// Clear stop attribute before launch so WriteState inside the manager
|
||||
// persists the correct attributes. Without this, a daemon restart after
|
||||
// a stop→start cycle would see StopInstance=true and skip reconnecting QEMU.
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) { v.Attributes = command.Attributes })
|
||||
d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes })
|
||||
|
||||
// Launch the instance infrastructure (QEMU, QMP, NATS subscriptions)
|
||||
if err := d.vmMgr.Start(instance.ID); err != nil {
|
||||
@@ -484,11 +482,7 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I
|
||||
|
||||
slog.Info(action+" instance", "id", command.ID)
|
||||
|
||||
// Check state validity before dispatching so we can return the correct
|
||||
// AWS error code (IncorrectInstanceState) when the instance is already
|
||||
// stopped/terminated/etc. Manager.Stop/Terminate re-validate internally.
|
||||
var currentState vm.InstanceState
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) { currentState = v.Status })
|
||||
currentState := d.vmMgr.Status(instance)
|
||||
|
||||
// Idempotent: a concurrent terminate goroutine is already cleaning up.
|
||||
if isTerminate && currentState == vm.StateShuttingDown {
|
||||
@@ -499,6 +493,10 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I
|
||||
return
|
||||
}
|
||||
|
||||
// Validate the transition synchronously before dispatching so the AWS
|
||||
// SDK sees IncorrectInstanceState (400) instead of a stale 200. The
|
||||
// async Stop/Terminate path re-validates and surfaces vm.ErrInvalidTransition
|
||||
// on a racing transition; we map that into the same AWS error below.
|
||||
if !vm.IsValidTransition(currentState, initialState) {
|
||||
slog.Warn("Instance in incorrect state for "+strings.ToLower(action),
|
||||
"instanceId", instance.ID, "currentState", string(currentState))
|
||||
@@ -524,6 +522,13 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I
|
||||
err = d.vmMgr.Stop(id)
|
||||
}
|
||||
if err != nil {
|
||||
// Race with another transition: log at the right level so it
|
||||
// shows up in dashboards but doesn't trigger paging.
|
||||
if errors.Is(err, vm.ErrInvalidTransition) {
|
||||
slog.Warn("Lifecycle transition raced; ack already sent",
|
||||
"id", id, "action", strings.ToLower(action), "err", err)
|
||||
return
|
||||
}
|
||||
slog.Error("Failed to "+strings.ToLower(action)+" instance", "err", err, "id", id)
|
||||
}
|
||||
}(instance.ID)
|
||||
|
||||
@@ -32,8 +32,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm
|
||||
|
||||
// Surface IncorrectInstanceState before any volume-side lookups so the
|
||||
// AWS-SDK error matches the pre-2d ordering.
|
||||
var status vm.InstanceState
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
|
||||
status := d.vmMgr.Status(instance)
|
||||
if status != vm.StateRunning {
|
||||
slog.Error("AttachVolume: instance not running",
|
||||
"instanceId", command.ID, "status", status)
|
||||
|
||||
@@ -370,7 +370,7 @@ func (d *Daemon) TerminateSystemInstance(instanceID string) error {
|
||||
} else {
|
||||
slog.Info("TerminateSystemInstance: released EIP", "instanceId", instanceID, "ip", instance.PublicIP, "allocationId", instance.PublicIPAllocID)
|
||||
}
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) {
|
||||
d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) {
|
||||
v.PublicIP = ""
|
||||
v.PublicIPPool = ""
|
||||
v.PublicIPAllocID = ""
|
||||
|
||||
@@ -49,7 +49,7 @@ func TestWaitForSystemInstance_TransitionsToRunning(t *testing.T) {
|
||||
// Transition to running after a short delay
|
||||
go func() {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
d.vmMgr.Inspect(inst, func(v *vm.VM) { v.Status = vm.StateRunning })
|
||||
d.vmMgr.UpdateState(inst.ID, func(v *vm.VM) { v.Status = vm.StateRunning })
|
||||
}()
|
||||
|
||||
err := d.WaitForSystemInstance("i-pending", 2*time.Second)
|
||||
|
||||
@@ -1954,8 +1954,9 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("InstanceNotRunning", func(t *testing.T) {
|
||||
// Temporarily set status to stopped
|
||||
instance.Status = vm.StateStopped
|
||||
// Temporarily set status to stopped under the manager lock so -race
|
||||
// reflects production discipline.
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
|
||||
command := types.EC2InstanceCommand{
|
||||
ID: instanceID,
|
||||
Attributes: types.EC2CommandAttributes{
|
||||
@@ -1976,7 +1977,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
|
||||
assert.Contains(t, string(resp.Data), "IncorrectInstanceState")
|
||||
|
||||
// Restore running state
|
||||
instance.Status = vm.StateRunning
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateRunning })
|
||||
})
|
||||
|
||||
t.Run("VolumeNotFound", func(t *testing.T) {
|
||||
@@ -2093,8 +2094,9 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("InstanceNotRunning", func(t *testing.T) {
|
||||
// Temporarily set status to stopped
|
||||
instance.Status = vm.StateStopped
|
||||
// Temporarily set status to stopped under the manager lock so -race
|
||||
// reflects production discipline.
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
|
||||
command := types.EC2InstanceCommand{
|
||||
ID: instanceID,
|
||||
Attributes: types.EC2CommandAttributes{
|
||||
@@ -2115,7 +2117,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
|
||||
assert.Contains(t, string(resp.Data), "IncorrectInstanceState")
|
||||
|
||||
// Restore running state
|
||||
instance.Status = vm.StateRunning
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateRunning })
|
||||
})
|
||||
|
||||
t.Run("VolumeNotAttached", func(t *testing.T) {
|
||||
@@ -3118,9 +3120,7 @@ func TestMarkInstanceFailed(t *testing.T) {
|
||||
assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code)
|
||||
assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message)
|
||||
require.Eventually(t, func() bool {
|
||||
var status vm.InstanceState
|
||||
daemon.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
|
||||
return status == vm.StateTerminated
|
||||
return daemon.vmMgr.Status(instance) == vm.StateTerminated
|
||||
}, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated")
|
||||
}
|
||||
|
||||
@@ -3142,9 +3142,7 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) {
|
||||
daemon.vmMgr.MarkFailed(instance, "test_failure")
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
var status vm.InstanceState
|
||||
daemon.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
|
||||
return status == vm.StateTerminated
|
||||
return daemon.vmMgr.Status(instance) == vm.StateTerminated
|
||||
}, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated")
|
||||
}
|
||||
|
||||
@@ -3292,7 +3290,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
t.Run("StopAlreadyStoppedInstance", func(t *testing.T) {
|
||||
instance.Status = vm.StateStopped
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
|
||||
command := types.EC2InstanceCommand{
|
||||
ID: instanceID,
|
||||
Attributes: types.EC2CommandAttributes{
|
||||
@@ -3312,7 +3310,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("TerminateAlreadyTerminatedInstance", func(t *testing.T) {
|
||||
instance.Status = vm.StateTerminated
|
||||
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateTerminated })
|
||||
command := types.EC2InstanceCommand{
|
||||
ID: instanceID,
|
||||
Attributes: types.EC2CommandAttributes{
|
||||
|
||||
@@ -336,7 +336,7 @@ func TestHandleInstanceCrash_FirstCrashSetsTime(t *testing.T) {
|
||||
assert.Equal(t, 1, instance.Health.CrashCount)
|
||||
|
||||
// Reset to running for a second crash
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) {
|
||||
d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) {
|
||||
v.Status = vm.StateRunning
|
||||
v.Running = true
|
||||
v.PID = 222
|
||||
|
||||
@@ -9,7 +9,9 @@ import (
|
||||
|
||||
// TransitionState validates and applies a state transition on the given instance.
|
||||
// It sets instance.Status, persists via WriteState, and logs.
|
||||
// On validation failure, the instance status is unchanged and an error is returned.
|
||||
// On validation failure, the instance status is unchanged and a wrapped
|
||||
// vm.ErrInvalidTransition is returned so callers can map it to the AWS
|
||||
// IncorrectInstanceState error code via errors.Is.
|
||||
// If WriteState fails, the in-memory status retains the new value (the VM has
|
||||
// physically changed state regardless) and an error is returned.
|
||||
func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error {
|
||||
@@ -17,6 +19,8 @@ func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error
|
||||
current vm.InstanceState
|
||||
invalid bool
|
||||
)
|
||||
// Inspect (not UpdateState): MarkFailed may invoke this for an instance
|
||||
// that is no longer in the running map.
|
||||
d.vmMgr.Inspect(instance, func(v *vm.VM) {
|
||||
current = v.Status
|
||||
if !vm.IsValidTransition(current, target) {
|
||||
@@ -26,7 +30,8 @@ func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error
|
||||
v.Status = target
|
||||
})
|
||||
if invalid {
|
||||
return fmt.Errorf("invalid state transition: %s -> %s for instance %s", current, target, instance.ID)
|
||||
return fmt.Errorf("%w: %s -> %s for instance %s",
|
||||
vm.ErrInvalidTransition, current, target, instance.ID)
|
||||
}
|
||||
|
||||
slog.Info("Instance state transition", "instanceId", instance.ID, "from", string(current), "to", string(target))
|
||||
|
||||
@@ -1006,7 +1006,8 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) {
|
||||
// Run one watchdog tick manually instead of waiting for the ticker
|
||||
var stuck []*vm.VM
|
||||
for _, instance := range daemon.vmMgr.Snapshot() {
|
||||
if (instance.Status == vm.StatePending || instance.Status == vm.StateProvisioning) &&
|
||||
status := daemon.vmMgr.Status(instance)
|
||||
if (status == vm.StatePending || status == vm.StateProvisioning) &&
|
||||
instance.Instance != nil && instance.Instance.LaunchTime != nil &&
|
||||
time.Since(*instance.Instance.LaunchTime) > 5*time.Minute {
|
||||
stuck = append(stuck, instance)
|
||||
@@ -1027,9 +1028,7 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) {
|
||||
assert.Equal(t, "Server.InternalError", *stuckBefore.Instance.StateReason.Code)
|
||||
assert.Equal(t, "launch_timeout", *stuckBefore.Instance.StateReason.Message)
|
||||
require.Eventually(t, func() bool {
|
||||
var status vm.InstanceState
|
||||
daemon.vmMgr.Inspect(stuckBefore, func(v *vm.VM) { status = v.Status })
|
||||
return status == vm.StateTerminated
|
||||
return daemon.vmMgr.Status(stuckBefore) == vm.StateTerminated
|
||||
}, 5*time.Second, 10*time.Millisecond, "MarkFailed cleanup goroutine should reach terminated")
|
||||
|
||||
// Fresh instance should still be pending
|
||||
|
||||
@@ -80,10 +80,7 @@ func ClassifyCrashReason(waitErr error) string {
|
||||
// Wired as Deps.CrashHandler so the launch goroutine in lifecycle.go can
|
||||
// invoke it without importing the manager into a separate goroutine spawner.
|
||||
func (m *Manager) HandleCrash(instance *VM, waitErr error) {
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
|
||||
if status != StateRunning {
|
||||
if status := m.Status(instance); status != StateRunning {
|
||||
slog.Debug("QEMU exited but instance not in running state, skipping crash handler",
|
||||
"instance", instance.ID, "status", status)
|
||||
return
|
||||
@@ -106,7 +103,7 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) {
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
v.Health.CrashCount++
|
||||
v.Health.LastCrashTime = now
|
||||
v.Health.LastCrashReason = reason
|
||||
@@ -158,7 +155,7 @@ func (m *Manager) MaybeRestart(instance *VM) {
|
||||
restartCount int
|
||||
exceeded bool
|
||||
)
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
health := &v.Health
|
||||
if !health.FirstCrashTime.IsZero() && now.Sub(health.FirstCrashTime) > RestartWindow {
|
||||
slog.Info("Crash window expired, resetting counters", "instance", v.ID)
|
||||
@@ -217,7 +214,7 @@ func (m *Manager) MaybeRestart(instance *VM) {
|
||||
func (m *Manager) RestartCrashedInstance(instance *VM) {
|
||||
var skipReason string
|
||||
var restartCount int
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
if v.Status != StateError {
|
||||
skipReason = fmt.Sprintf("not in error state (%s)", v.Status)
|
||||
return
|
||||
|
||||
@@ -177,7 +177,7 @@ func (m *Manager) UpdateGuestDeviceNames(instance *VM) {
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
if v.Instance == nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -47,9 +47,7 @@ func (m *Manager) Reboot(id string) error {
|
||||
if !ok {
|
||||
return ErrInstanceNotFound
|
||||
}
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
if status != StateRunning {
|
||||
if status := m.Status(instance); status != StateRunning {
|
||||
return fmt.Errorf("%w: cannot reboot instance %s in state %s",
|
||||
ErrInvalidTransition, id, status)
|
||||
}
|
||||
@@ -65,8 +63,7 @@ func (m *Manager) Reboot(id string) error {
|
||||
// terminate goroutine owns cleanup and launch must bail without further side
|
||||
// effects.
|
||||
func (m *Manager) launchStillValid(instance *VM) bool {
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
status := m.Status(instance)
|
||||
if status == StatePending || status == StateStopped || status == StateProvisioning {
|
||||
return true
|
||||
}
|
||||
@@ -446,8 +443,7 @@ func (m *Manager) qmpHeartbeat(instance *VM) {
|
||||
for {
|
||||
time.Sleep(30 * time.Second)
|
||||
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
status := m.Status(instance)
|
||||
|
||||
if status == StateStopping || status == StateStopped ||
|
||||
status == StateShuttingDown || status == StateTerminated || status == StateError {
|
||||
|
||||
+13
-1
@@ -220,10 +220,22 @@ func (m *Manager) UpdateState(id string, fn func(*VM)) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// Status returns v.Status under the manager lock. Replaces the dominant
|
||||
// "Inspect to read Status" pattern with a typed accessor. Read-only — no
|
||||
// membership check, so callers may pass a pointer to an instance no longer
|
||||
// in the map (e.g. mid-cleanup) and still get a consistent read.
|
||||
func (m *Manager) Status(v *VM) InstanceState {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return v.Status
|
||||
}
|
||||
|
||||
// Inspect runs fn(v) under the manager lock for an already-resolved VM
|
||||
// pointer. Used by call sites that hold a *VM (e.g. from a NATS handler
|
||||
// dispatch) and need to read or mutate its fields with the same memory-
|
||||
// ordering guarantee as map-keyed access.
|
||||
// ordering guarantee as map-keyed access. Prefer UpdateState (membership-
|
||||
// checked) for mutation and Status for the read-Status pattern; Inspect
|
||||
// remains for closures that mutate fields on a possibly-orphaned VM.
|
||||
func (m *Manager) Inspect(v *VM, fn func(*VM)) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
@@ -119,6 +119,25 @@ func TestManager_Inspect(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_Status(t *testing.T) {
|
||||
m := NewManager()
|
||||
a := mkVM("i-a")
|
||||
a.Status = StatePending
|
||||
m.Insert(a)
|
||||
|
||||
if got := m.Status(a); got != StatePending {
|
||||
t.Fatalf("Status: got %s, want %s", got, StatePending)
|
||||
}
|
||||
|
||||
// Status reads under lock without verifying membership — orphaned VMs
|
||||
// (mid-cleanup, never-inserted) still return a consistent value.
|
||||
loose := mkVM("i-loose")
|
||||
loose.Status = StateRunning
|
||||
if got := m.Status(loose); got != StateRunning {
|
||||
t.Fatalf("Status on loose VM: got %s, want %s", got, StateRunning)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_ForEachAndSnapshot(t *testing.T) {
|
||||
m := NewManager()
|
||||
want := map[string]bool{"i-1": true, "i-2": true, "i-3": true}
|
||||
|
||||
@@ -275,8 +275,7 @@ func (m *Manager) relaunchAll(toLaunch []*VM) {
|
||||
}
|
||||
}()
|
||||
|
||||
var status InstanceState
|
||||
m.Inspect(inst, func(v *VM) { status = v.Status })
|
||||
status := m.Status(inst)
|
||||
if status != StatePending && status != StateProvisioning {
|
||||
slog.Info("Instance state changed during recovery, skipping launch",
|
||||
"instanceId", inst.ID, "status", string(status))
|
||||
|
||||
+9
-12
@@ -38,7 +38,7 @@ func (m *Manager) Stop(id string) error {
|
||||
|
||||
m.stopCleanup(instance)
|
||||
|
||||
m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID })
|
||||
m.UpdateState(instance.ID, func(v *VM) { v.LastNode = m.deps.NodeID })
|
||||
|
||||
if err := m.transitionWithPrecheck(instance, StateStopped); err != nil {
|
||||
slog.Error("Failed to transition to stopped", "instanceId", instance.ID, "err", err)
|
||||
@@ -110,9 +110,7 @@ func (m *Manager) Terminate(id string) error {
|
||||
return ErrInstanceNotFound
|
||||
}
|
||||
|
||||
var current InstanceState
|
||||
m.Inspect(instance, func(v *VM) { current = v.Status })
|
||||
if current == StateShuttingDown {
|
||||
if current := m.Status(instance); current == StateShuttingDown {
|
||||
// Concurrent failed-launch goroutine already owns cleanup.
|
||||
return nil
|
||||
}
|
||||
@@ -161,9 +159,7 @@ func (m *Manager) MarkFailed(instance *VM, reason string) {
|
||||
slog.Error("MarkFailed transition failed", "instanceId", instance.ID, "err", err)
|
||||
// If this was a persistence-only failure, in-memory state is now
|
||||
// shutting-down and we still want to finalize. Otherwise bail.
|
||||
var postErr InstanceState
|
||||
m.Inspect(instance, func(v *VM) { postErr = v.Status })
|
||||
if postErr != StateShuttingDown {
|
||||
if m.Status(instance) != StateShuttingDown {
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -182,6 +178,8 @@ func (m *Manager) MarkFailed(instance *VM, reason string) {
|
||||
// OnInstanceDown, and persists the running set. Shared by Terminate and
|
||||
// MarkFailed.
|
||||
func (m *Manager) finalizeTerminated(instance *VM) error {
|
||||
// Inspect (not UpdateState): MarkFailed may invoke this for an
|
||||
// instance that was never inserted into the local map.
|
||||
m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID })
|
||||
|
||||
if err := m.transitionWithPrecheck(instance, StateTerminated); err != nil {
|
||||
@@ -323,13 +321,14 @@ func (m *Manager) deallocateResources(instance *VM) {
|
||||
// as a persistence failure on a transition whose memory mutation
|
||||
// already succeeded.
|
||||
func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) error {
|
||||
var current InstanceState
|
||||
m.Inspect(instance, func(v *VM) { current = v.Status })
|
||||
current := m.Status(instance)
|
||||
if !IsValidTransition(current, target) {
|
||||
return fmt.Errorf("%w: %s -> %s for instance %s",
|
||||
ErrInvalidTransition, current, target, instance.ID)
|
||||
}
|
||||
if m.deps.TransitionState == nil {
|
||||
// Inspect (not UpdateState): MarkFailed may run this on an instance
|
||||
// that was never inserted into the local map.
|
||||
m.Inspect(instance, func(v *VM) { v.Status = target })
|
||||
return nil
|
||||
}
|
||||
@@ -337,9 +336,7 @@ func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) err
|
||||
// Could be persistence failure (memory state already updated) or a
|
||||
// racing transition that invalidated the precheck. Re-inspect to
|
||||
// distinguish.
|
||||
var observed InstanceState
|
||||
m.Inspect(instance, func(v *VM) { observed = v.Status })
|
||||
if observed != target {
|
||||
if m.Status(instance) != target {
|
||||
return fmt.Errorf("%w: %s -> %s for instance %s (raced)",
|
||||
ErrInvalidTransition, current, target, instance.ID)
|
||||
}
|
||||
|
||||
@@ -44,15 +44,13 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult,
|
||||
return AttachVolumeResult{}, ErrInstanceNotFound
|
||||
}
|
||||
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
if status != StateRunning {
|
||||
if status := m.Status(instance); status != StateRunning {
|
||||
return AttachVolumeResult{}, fmt.Errorf("%w: cannot attach to instance %s in state %s",
|
||||
ErrInvalidTransition, id, status)
|
||||
}
|
||||
|
||||
if device == "" {
|
||||
m.Inspect(instance, func(v *VM) { device = nextAvailableDevice(v) })
|
||||
m.UpdateState(id, func(v *VM) { device = nextAvailableDevice(v) })
|
||||
if device == "" {
|
||||
return AttachVolumeResult{}, ErrAttachmentLimitExceeded
|
||||
}
|
||||
@@ -190,7 +188,7 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult,
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(id, func(v *VM) {
|
||||
if v.Instance == nil {
|
||||
return
|
||||
}
|
||||
@@ -238,9 +236,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
|
||||
return "", ErrInstanceNotFound
|
||||
}
|
||||
|
||||
var status InstanceState
|
||||
m.Inspect(instance, func(v *VM) { status = v.Status })
|
||||
if status != StateRunning {
|
||||
if status := m.Status(instance); status != StateRunning {
|
||||
return "", fmt.Errorf("%w: cannot detach from instance %s in state %s",
|
||||
ErrInvalidTransition, id, status)
|
||||
}
|
||||
@@ -330,7 +326,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
m.Inspect(instance, func(v *VM) {
|
||||
m.UpdateState(id, func(v *VM) {
|
||||
if v.Instance == nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -374,9 +374,7 @@ func TestReboot_DoesNotChangeStatus(t *testing.T) {
|
||||
|
||||
v, ok := m.Get("i-1")
|
||||
require.True(t, ok)
|
||||
var status InstanceState
|
||||
m.Inspect(v, func(v *VM) { status = v.Status })
|
||||
assert.Equal(t, StateRunning, status)
|
||||
assert.Equal(t, StateRunning, m.Status(v))
|
||||
}
|
||||
|
||||
func TestReboot_QMPFailureSurfacesError(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user