diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 69e0435fd..1b528d5d9 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1507,7 +1507,10 @@ func (d *Daemon) startCluster() error { // data-safety reaper (ADR-0005 §3) rides the same backstop but only marks + // alarms — it never deletes volume data. if d.jsManager != nil { - reapers := []vm.Reaper{d.vmMgr.NewTerminatedTeardownReaper()} + reapers := []vm.Reaper{ + d.vmMgr.NewTerminatedTeardownReaper(), + d.vmMgr.NewOrphanQEMUReaper(), + } if eniRec := d.newENIReconciler(); eniRec != nil { reapers = append(reapers, eniRec) } diff --git a/spinifex/utils/process.go b/spinifex/utils/process.go index c8cb15513..734754967 100644 --- a/spinifex/utils/process.go +++ b/spinifex/utils/process.go @@ -39,6 +39,39 @@ func StopProcess(serviceName string) error { return nil } +// ProcessAlive reports whether the process is still running, via a +// signal-0 liveness probe. A PID file going missing does NOT imply the +// process exited, so callers that must reap a process check this directly. +func ProcessAlive(pid int) bool { + if pid <= 0 { + return false + } + process, err := os.FindProcess(pid) + if err != nil { + return false + } + return process.Signal(syscall.Signal(0)) == nil +} + +// ForceKillProcess SIGKILLs a process immediately and waits up to timeout for +// it to exit. Used to reap a known-orphan (e.g. a QEMU for an +// already-terminated instance) where the graceful SIGTERM grace period is not +// warranted. SIGKILL cannot be caught, so the process never removes its own PID +// file — callers should RemovePidFile after this returns. +func ForceKillProcess(pid int, timeout time.Duration) error { + if pid <= 0 { + return fmt.Errorf("invalid pid %d", pid) + } + process, err := os.FindProcess(pid) + if err != nil { + return err + } + if err := process.Signal(syscall.SIGKILL); err != nil { + return err + } + return WaitForProcessExit(pid, timeout) +} + func KillProcess(pid int) error { process, err := os.FindProcess(pid) diff --git a/spinifex/utils/utils_test.go b/spinifex/utils/utils_test.go index 0ac10299d..c8c2a522c 100644 --- a/spinifex/utils/utils_test.go +++ b/spinifex/utils/utils_test.go @@ -633,6 +633,39 @@ func TestKillProcess(t *testing.T) { assert.Error(t, err, "Should error when killing non-existent process") } +func TestProcessAlive(t *testing.T) { + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + + var wg sync.WaitGroup + wg.Go(func() { _ = cmd.Wait() }) + + assert.True(t, ProcessAlive(pid), "a running process must report alive") + assert.False(t, ProcessAlive(0), "pid 0 must report not alive") + assert.False(t, ProcessAlive(-1), "negative pid must report not alive") + assert.False(t, ProcessAlive(999999), "a non-existent pid must report not alive") + + require.NoError(t, cmd.Process.Kill()) + wg.Wait() + assert.False(t, ProcessAlive(pid), "a killed process must report not alive") +} + +func TestForceKillProcess(t *testing.T) { + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + + var wg sync.WaitGroup + wg.Go(func() { _ = cmd.Wait() }) + + require.NoError(t, ForceKillProcess(pid, 5*time.Second)) + wg.Wait() + assert.False(t, ProcessAlive(pid), "ForceKillProcess must SIGKILL and confirm exit") + + assert.Error(t, ForceKillProcess(0, time.Second), "invalid pid must error") +} + func TestStopProcess(t *testing.T) { // Create and start a test process cmd := exec.Command("sleep", "60") diff --git a/spinifex/vm/orphan_qemu_reaper.go b/spinifex/vm/orphan_qemu_reaper.go new file mode 100644 index 000000000..a49a9231d --- /dev/null +++ b/spinifex/vm/orphan_qemu_reaper.go @@ -0,0 +1,77 @@ +package vm + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/mulgadc/spinifex/spinifex/utils" +) + +// orphanQEMUKillTimeout bounds how long the reaper waits for a SIGKILL'd +// orphan QEMU to disappear before giving up and retrying on the next sweep. +const orphanQEMUKillTimeout = 10 * time.Second + +// OrphanQEMUReaper kills QEMU processes left running for instances that are +// already terminated. TerminateInstances marks the EC2 record terminated and +// drives node-local teardown, but if that teardown ran on a node other than the +// one hosting the QEMU (e.g. the per-instance subscription was lost and the +// request fell back to the ec2.terminate broadcast), the process survives and +// keeps holding its OVN logical port / IP, blocking rebuilds. +// +// This reaper is node-local and conservative: it only inspects instances in the +// terminated KV bucket and only acts when a live process is found via the +// instance's local PID file. Running, pending, and stopped instances are never +// touched, so a launching or healthy VM can never be reaped. Whichever node +// actually holds the orphan PID file is the one that reaps it. +type OrphanQEMUReaper struct { + m *Manager +} + +var _ Reaper = (*OrphanQEMUReaper)(nil) + +// NewOrphanQEMUReaper builds the reaper bound to this Manager's state store. +func (m *Manager) NewOrphanQEMUReaper() *OrphanQEMUReaper { + return &OrphanQEMUReaper{m: m} +} + +func (r *OrphanQEMUReaper) Class() string { return "orphan-qemu" } +func (r *OrphanQEMUReaper) Scope() ReaperScope { return ScopeNodeLocal } + +// Sweep reaps any live QEMU whose instance is already terminated. A terminated +// instance must have no running process; one that does is an orphan holding +// OVN ports. The PID file is read by instance ID, so only the node hosting the +// process finds it alive — every other node's lookup misses and is a no-op. +func (r *OrphanQEMUReaper) Sweep(context.Context) (int, error) { + if r.m.deps.StateStore == nil { + return 0, nil + } + terminated, err := r.m.deps.StateStore.ListTerminatedInstances() + if err != nil { + return 0, fmt.Errorf("list terminated instances: %w", err) + } + + reaped := 0 + for _, v := range terminated { + pid, err := utils.ReadPidFile(v.ID) + if err != nil { + continue // no PID file on this node: process is not here + } + if !utils.ProcessAlive(pid) { + _ = utils.RemovePidFile(v.ID) // stale PID file for a dead, terminated instance + continue + } + + slog.Warn("vm/gc: reaping orphan QEMU for terminated instance (held OVN ports)", + "instanceId", v.ID, "pid", pid) + if err := utils.ForceKillProcess(pid, orphanQEMUKillTimeout); err != nil { + slog.Error("vm/gc: failed to reap orphan QEMU, will retry next sweep", + "instanceId", v.ID, "pid", pid, "err", err) + continue + } + _ = utils.RemovePidFile(v.ID) + reaped++ + } + return reaped, nil +} diff --git a/spinifex/vm/orphan_qemu_reaper_test.go b/spinifex/vm/orphan_qemu_reaper_test.go new file mode 100644 index 000000000..6da5d14d1 --- /dev/null +++ b/spinifex/vm/orphan_qemu_reaper_test.go @@ -0,0 +1,86 @@ +package vm + +import ( + "context" + "os/exec" + "sync" + "testing" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newOrphanReaperManager(t *testing.T) (*OrphanQEMUReaper, *fakeStateStore) { + t.Helper() + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + store := newFakeStateStore() + m := NewManager() + m.SetDeps(Deps{NodeID: "test-node", StateStore: store}) + return m.NewOrphanQEMUReaper(), store +} + +func TestOrphanQEMUReaper(t *testing.T) { + t.Run("kills live QEMU for a terminated instance and removes its PID file", func(t *testing.T) { + reaper, store := newOrphanReaperManager(t) + + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + var wg sync.WaitGroup + wg.Go(func() { _ = cmd.Wait() }) + + const id = "i-orphan-term" + require.NoError(t, utils.WritePidFile(id, pid)) + store.terminated[id] = &VM{ID: id, Status: StateTerminated} + + reaped, err := reaper.Sweep(context.Background()) + require.NoError(t, err) + wg.Wait() + + assert.Equal(t, 1, reaped, "the orphan QEMU for a terminated instance must be reaped") + assert.False(t, utils.ProcessAlive(pid), + "a terminated instance must have no surviving process — otherwise it holds OVN ports (siv-476)") + _, perr := utils.ReadPidFile(id) + assert.Error(t, perr, "the PID file must be removed after reaping") + }) + + t.Run("no PID file for a terminated instance is a no-op", func(t *testing.T) { + reaper, store := newOrphanReaperManager(t) + store.terminated["i-term-elsewhere"] = &VM{ID: "i-term-elsewhere", Status: StateTerminated} + + reaped, err := reaper.Sweep(context.Background()) + require.NoError(t, err) + assert.Zero(t, reaped, "an instance whose process lives on another node must not be reaped here") + }) + + t.Run("stale PID file (dead process) is cleaned without counting as a reap", func(t *testing.T) { + reaper, store := newOrphanReaperManager(t) + const id = "i-term-stale" + require.NoError(t, utils.WritePidFile(id, 999999)) // dead pid + store.terminated[id] = &VM{ID: id, Status: StateTerminated} + + reaped, err := reaper.Sweep(context.Background()) + require.NoError(t, err) + assert.Zero(t, reaped, "a dead process is not a reap") + _, perr := utils.ReadPidFile(id) + assert.Error(t, perr, "the stale PID file must be cleaned up") + }) + + t.Run("a running instance with a live process is never touched", func(t *testing.T) { + reaper, _ := newOrphanReaperManager(t) + + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + + // PID file exists but the instance is NOT in the terminated bucket. + require.NoError(t, utils.WritePidFile("i-running", pid)) + + reaped, err := reaper.Sweep(context.Background()) + require.NoError(t, err) + assert.Zero(t, reaped, "only terminated instances are candidates") + assert.True(t, utils.ProcessAlive(pid), "a non-terminated instance's process must survive") + }) +} diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index 6edb1c18f..d7511df5d 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -351,12 +351,17 @@ func (m *Manager) shutdownAndUnmount(instance *VM) { } } + // The PID file persisting past the timeout means QEMU did not exit on its + // own; force-kill it then. A PID file that disappears is the clean-exit + // signal — do not kill on that path (the PID may be stale or reused). The + // wrong-node terminate case, where this never runs on the hosting node, is + // the OrphanQEMUReaper's job. if err := utils.WaitForPidFileRemoval(instance.ID, pidFileRemovalTimeout); err != nil { slog.Warn("Timeout waiting for PID file removal", "id", instance.ID, "err", err) pid, readErr := utils.ReadPidFile(instance.ID) if readErr != nil { slog.Debug("No PID file found (VM likely already stopped)", "id", instance.ID) - } else { + } else if utils.ProcessAlive(pid) { slog.Info("Force killing process", "pid", pid, "id", instance.ID) if err := utils.KillProcess(pid); err != nil { slog.Error("Failed to kill process", "pid", pid, "id", instance.ID, "err", err)