mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: reap orphan QEMU for terminated instances via node-local reaper (#442)
TerminateInstances marks the EC2 record terminated and drives node-local teardown, but when that teardown runs on a node other than the one hosting QEMU (per-instance subscription lost, request falls back to the ec2.terminate broadcast), the process survives and keeps holding its OVN logical port / IP, blocking rebuilds. Add OrphanQEMUReaper (node-local, ScopeNodeLocal): it inspects only the terminated KV bucket and reaps a live process found via the instance's local PID file. Running, pending, and stopped instances are never touched. Whichever node holds the orphan PID file is the one that reaps it; every other node's lookup misses and is a no-op. Add utils.ProcessAlive (signal-0 liveness) and utils.ForceKillProcess (SIGKILL + bounded wait), guard shutdownAndUnmount's force-kill on liveness, and register the reaper in the daemon GC loop.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
})
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user