From fd6c12f140dd7ed1b842714836e1aa51b77dfa3c Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Fri, 2 Oct 2026 15:34:38 +1000 Subject: [PATCH] refactor: move process supervision helpers to runtime host --- spinifex/daemon/daemon.go | 3 +- spinifex/daemon/daemon_system_instance.go | 9 +- spinifex/daemon/local_api.go | 4 +- spinifex/daemon/local_api_test.go | 8 +- spinifex/daemon/shutdown.go | 16 +- spinifex/daemon/shutdown_test.go | 8 +- .../host/systemd/units_invariants_test.go | 2 +- .../runtime/compute/vm/crash_recovery_test.go | 2 +- spinifex/runtime/compute/vm/lifecycle.go | 19 +- .../runtime/compute/vm/orphan_qemu_reaper.go | 22 +- .../compute/vm/orphan_qemu_reaper_test.go | 18 +- spinifex/runtime/compute/vm/orphan_scan.go | 6 +- spinifex/runtime/compute/vm/restore.go | 9 +- spinifex/runtime/compute/vm/restore_test.go | 20 +- spinifex/runtime/compute/vm/shutdown.go | 18 +- spinifex/runtime/compute/vm/shutdown_test.go | 18 +- .../compute/vm/stuck_terminate_reaper_test.go | 6 +- spinifex/runtime/compute/vm/telemetry_meta.go | 6 +- .../runtime/compute/vm/telemetry_meta_test.go | 14 +- spinifex/runtime/compute/vm/volumes_test.go | 12 +- .../host/process}/pidfile.go | 13 +- spinifex/runtime/host/process/pidfile_test.go | 472 ++++++++++++++++++ .../host/process}/process.go | 2 +- .../host/process}/process_terminate_test.go | 14 +- .../host/process}/process_test.go | 2 +- spinifex/runtime/roles/nats/nats.go | 4 +- spinifex/runtime/roles/qemunbdd/service.go | 3 +- .../roles/qmpcollector/qmpcollector.go | 7 +- spinifex/runtime/roles/spinifex/spinifex.go | 4 +- .../runtime/roles/spinifexui/spinifexui.go | 4 +- spinifex/services/awsgw/awsgw.go | 3 +- spinifex/services/northstar/northstar.go | 3 +- spinifex/services/predastore/predastore.go | 4 +- .../viperblockd/nbdkit_recovery_test.go | 28 +- .../services/viperblockd/provider_handlers.go | 9 +- spinifex/services/viperblockd/viperblockd.go | 11 +- .../viperblockd/volume_abandon_test.go | 4 +- spinifex/services/viperblockd/volume_fence.go | 6 +- .../services/viperblockd/volume_fence_test.go | 4 +- spinifex/utils/nbd.go | 8 +- spinifex/utils/utils_test.go | 457 ----------------- spinifex/vpcd/vpcd.go | 3 +- 42 files changed, 662 insertions(+), 623 deletions(-) rename spinifex/{utils => runtime/host/process}/pidfile.go (96%) create mode 100644 spinifex/runtime/host/process/pidfile_test.go rename spinifex/{utils => runtime/host/process}/process.go (99%) rename spinifex/{utils => runtime/host/process}/process_terminate_test.go (82%) rename spinifex/{utils => runtime/host/process}/process_test.go (99%) diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 633a925b4..4158bfcc0 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -78,6 +78,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/providers/objectstore" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -1383,7 +1384,7 @@ func (d *Daemon) startLocal() error { } // Protect daemon from OOM killer (prefer killing QEMU VMs instead). - if err := utils.SetOOMScore(os.Getpid(), -500); err != nil { + if err := hostprocess.SetOOMScore(os.Getpid(), -500); err != nil { slog.Warn("Failed to set daemon OOM score", "err", err) } diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index a18162c93..c070cac27 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -21,6 +21,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -668,7 +669,7 @@ func (d *Daemon) prepareInstanceRelaunch(inst *vm.VM) error { } // refreshSystemInstanceState regenerates the tmpfs-backed fw_cfg blobs that -// QEMU loads at boot. The blobs live under utils.RuntimeDir() (tmpfs on +// QEMU loads at boot. The blobs live under hostprocess.RuntimeDir() (tmpfs on // production hosts) and are wiped on host reboot while the persisted // vm.Config still references the same paths. Customer VMs use only paths // under /var/lib/spinifex/ and are a no-op. @@ -778,7 +779,7 @@ func (d *Daemon) WaitForSystemInstance(instanceID string, timeout time.Duration) // (PID file, console log, serial socket) are filled in later by startQEMU. // // Network topology from input.NICs is serialised to three fw_cfg tmpfiles -// (netcfg, lb-agent-env, ca-cert) under utils.RuntimeDir(). +// (netcfg, lb-agent-env, ca-cert) under hostprocess.RuntimeDir(). func (d *Daemon) buildDirectBootConfig(instanceID string, input *handlers_elbv2.SystemInstanceInput) (vm.Config, error) { it := d.resourceMgr.instanceTypes[input.InstanceType] architecture := "x86_64" @@ -923,11 +924,11 @@ func tapNameForNIC(idx int, _ handlers_elbv2.NICConfig, instanceID string, input } // writeFwCfgBlobs serialises NIC configuration, lb-agent env, and CA cert to -// per-VM tmpfiles under utils.RuntimeDir(). Returns the fw_cfg entries for the +// per-VM tmpfiles under hostprocess.RuntimeDir(). Returns the fw_cfg entries for the // three blobs and an error if any write fails or the NIC default invariant is // violated. func (d *Daemon) writeFwCfgBlobs(instanceID string, input *handlers_elbv2.SystemInstanceInput) ([]vm.FwCfgEntry, error) { - runtimeDir := utils.RuntimeDir() + runtimeDir := hostprocess.RuntimeDir() netcfgPath := filepath.Join(runtimeDir, fmt.Sprintf("fwcfg-%s-netcfg.tmp", instanceID)) lbenvPath := filepath.Join(runtimeDir, fmt.Sprintf("fwcfg-%s-lbenv.tmp", instanceID)) diff --git a/spinifex/daemon/local_api.go b/spinifex/daemon/local_api.go index 4a5af2e80..2c062b816 100644 --- a/spinifex/daemon/local_api.go +++ b/spinifex/daemon/local_api.go @@ -9,7 +9,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // NATS connectivity strings reported by /local/status. @@ -109,7 +109,7 @@ func vmToLocalInstance(v *vm.VM) LocalInstance { // PID is read from the QEMU pidfile each request; missing file or unreadable // value just omits the field (json:"omitempty"). Read failures are not // surfaced — /local/instances must keep working even when QEMU is gone. - if pid, err := utils.ReadPidFile(v.ID); err == nil { + if pid, err := hostprocess.ReadPidFile(v.ID); err == nil { li.PID = pid } return li diff --git a/spinifex/daemon/local_api_test.go b/spinifex/daemon/local_api_test.go index 3f3c128a2..4a7b98b9a 100644 --- a/spinifex/daemon/local_api_test.go +++ b/spinifex/daemon/local_api_test.go @@ -11,7 +11,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/mulgadc/spinifex/spinifex/bootstrap/config" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -19,7 +19,7 @@ import ( // newLocalAPITestDaemon returns a Daemon with just enough wiring to exercise // the read-only /local/* handlers: a vm.Manager, a node name, and a Config so // any path-resolving code (revision bump via WriteState) has a temp DataDir. -// XDG_RUNTIME_DIR is pinned to a temp dir so utils.WritePidFile / ReadPidFile +// XDG_RUNTIME_DIR is pinned to a temp dir so hostprocess.WritePidFile / ReadPidFile // (used by vmToLocalInstance to surface PIDs) stay scoped to the test. func newLocalAPITestDaemon(t *testing.T) *Daemon { t.Helper() @@ -63,7 +63,7 @@ func TestLocalAPI_Instances_ListsVMs(t *testing.T) { d := newLocalAPITestDaemon(t) d.vmMgr.Insert(&vm.VM{ID: "i-b", Status: vm.StateRunning}) d.vmMgr.Insert(&vm.VM{ID: "i-a", Status: vm.StateStopped}) - require.NoError(t, utils.WritePidFile("i-b", 4242)) + require.NoError(t, hostprocess.WritePidFile("i-b", 4242)) rec := doGET(t, newLocalAPIRouter(d), "/local/instances") require.Equal(t, http.StatusOK, rec.Code) @@ -83,7 +83,7 @@ func TestLocalAPI_Instances_ListsVMs(t *testing.T) { func TestLocalAPI_Instance_Found(t *testing.T) { d := newLocalAPITestDaemon(t) d.vmMgr.Insert(&vm.VM{ID: "i-find", Status: vm.StateRunning}) - require.NoError(t, utils.WritePidFile("i-find", 7)) + require.NoError(t, hostprocess.WritePidFile("i-find", 7)) rec := doGET(t, newLocalAPIRouter(d), "/local/instances/i-find") require.Equal(t, http.StatusOK, rec.Code) diff --git a/spinifex/daemon/shutdown.go b/spinifex/daemon/shutdown.go index 441d67245..e487e69ce 100644 --- a/spinifex/daemon/shutdown.go +++ b/spinifex/daemon/shutdown.go @@ -11,7 +11,7 @@ import ( "strings" "syscall" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/nats-io/nats.go" ) @@ -79,7 +79,7 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) string { // Stop AWSGW if d.config.HasService("awsgw") { - if err := utils.StopProcessAt(d.pidDir(), "awsgw"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "awsgw"); err != nil { slog.Warn("Failed to stop awsgw", "error", err) } else { stopped = append(stopped, "awsgw") @@ -88,7 +88,7 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) string { // Stop UI if d.config.HasService("ui") { - if err := utils.StopProcessAt(d.pidDir(), "spinifex-ui"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "spinifex-ui"); err != nil { slog.Warn("Failed to stop spinifex-ui", "error", err) } else { stopped = append(stopped, "spinifex-ui") @@ -97,7 +97,7 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) string { // Stop vpcd (VPC daemon) if d.config.HasService("vpcd") { - if err := utils.StopProcessAt(d.pidDir(), "vpcd"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "vpcd"); err != nil { slog.Warn("Failed to stop vpcd", "error", err) } else { stopped = append(stopped, "vpcd") @@ -196,14 +196,14 @@ func (d *Daemon) handleShutdownStorage(msg *nats.Msg) string { var stopped []string if d.config.HasService("viperblock") { - if err := utils.StopProcessAt(d.pidDir(), "viperblock"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "viperblock"); err != nil { slog.Warn("Failed to stop viperblock", "error", err) } else { stopped = append(stopped, "viperblock") } } - cleanupOrphanNBDKit(utils.RuntimeDir()) + cleanupOrphanNBDKit(hostprocess.RuntimeDir()) ack := ShutdownACK{ Node: d.node, @@ -232,7 +232,7 @@ func (d *Daemon) handleShutdownPersist(msg *nats.Msg) string { var stopped []string if d.config.HasService("predastore") { - if err := utils.StopProcessAt(d.pidDir(), "predastore"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "predastore"); err != nil { slog.Warn("Failed to stop predastore", "error", err) } else { stopped = append(stopped, "predastore") @@ -287,7 +287,7 @@ func (d *Daemon) handleShutdownInfra(msg *nats.Msg) string { // Stop NATS server if this node runs it if d.config.HasService("nats") { - if err := utils.StopProcessAt(d.pidDir(), "nats"); err != nil { + if err := hostprocess.StopProcessAt(d.pidDir(), "nats"); err != nil { slog.Warn("Failed to stop nats", "error", err) } else { slog.Info("NATS server stopped", "node", d.node) diff --git a/spinifex/daemon/shutdown_test.go b/spinifex/daemon/shutdown_test.go index 152fac8eb..b4c7e132f 100644 --- a/spinifex/daemon/shutdown_test.go +++ b/spinifex/daemon/shutdown_test.go @@ -14,7 +14,7 @@ import ( viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" ebspolicy "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/policy" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -231,7 +231,7 @@ func TestHandleShutdownGate(t *testing.T) { pidDir := configurePidDir(t, daemon) pid := startSleepProcess(t) - require.NoError(t, utils.WritePidFileTo(pidDir, "awsgw", pid)) + require.NoError(t, hostprocess.WritePidFileTo(pidDir, "awsgw", pid)) subject := "spinifex.cluster.shutdown.gate" sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleShutdownGate)) @@ -287,7 +287,7 @@ func TestHandleShutdownGate(t *testing.T) { pidDir := configurePidDir(t, daemon) pid := startSleepProcess(t) - require.NoError(t, utils.WritePidFileTo(pidDir, "awsgw", pid)) + require.NoError(t, hostprocess.WritePidFileTo(pidDir, "awsgw", pid)) subject := "spinifex.cluster.shutdown.gate.partial" sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleShutdownGate)) @@ -321,7 +321,7 @@ func TestShutdownRequestTarget(t *testing.T) { pidDir := configurePidDir(t, daemon) pid := startSleepProcess(t) - require.NoError(t, utils.WritePidFileTo(pidDir, "awsgw", pid)) + require.NoError(t, hostprocess.WritePidFileTo(pidDir, "awsgw", pid)) subject := "spinifex.cluster.shutdown.gate.targeted-elsewhere" sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleShutdownGate)) diff --git a/spinifex/operator/host/systemd/units_invariants_test.go b/spinifex/operator/host/systemd/units_invariants_test.go index b383519b0..65b9560cb 100644 --- a/spinifex/operator/host/systemd/units_invariants_test.go +++ b/spinifex/operator/host/systemd/units_invariants_test.go @@ -283,7 +283,7 @@ func TestGracefulDrainOrdering(t *testing.T) { } // shutdownVolumes reaps idle nbdkit concurrently, so the unit only has to - // outlive the single slowest utils.KillProcess grace (120s), not that + // outlive the single slowest hostprocess.KillProcess grace (120s), not that // grace summed across every mounted volume. const killProcessGracePeriod = 120 * time.Second if got := timeoutStopSec(t, viperblock); got <= killProcessGracePeriod { diff --git a/spinifex/runtime/compute/vm/crash_recovery_test.go b/spinifex/runtime/compute/vm/crash_recovery_test.go index c8767ceee..189fee76b 100644 --- a/spinifex/runtime/compute/vm/crash_recovery_test.go +++ b/spinifex/runtime/compute/vm/crash_recovery_test.go @@ -113,7 +113,7 @@ func TestRestartBackoff_Exponential(t *testing.T) { // shutdown signal. // // Sets XDG_RUNTIME_DIR to a per-test tempdir so PID-file paths -// (utils.WaitForPidFileRemoval, ReadPidFile) cannot collide between +// (hostprocess.WaitForPidFileRemoval, ReadPidFile) cannot collide between // tests sharing the host's real runtime dir. func crashTestManager(t *testing.T) (m *Manager, rc *fakeResourceController, rt *recordedTransitions, shuttingDown *atomic.Bool) { t.Helper() diff --git a/spinifex/runtime/compute/vm/lifecycle.go b/spinifex/runtime/compute/vm/lifecycle.go index f2ea1e7b9..69ad0d8ec 100644 --- a/spinifex/runtime/compute/vm/lifecycle.go +++ b/spinifex/runtime/compute/vm/lifecycle.go @@ -22,6 +22,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -271,7 +272,7 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) { } }) - pid, _ := utils.ReadPidFile(instance.ID) + pid, _ := hostprocess.ReadPidFile(instance.ID) if pid > 0 { process, err := os.FindProcess(pid) if err != nil { @@ -327,11 +328,11 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) { // VFIO device is released before the caller frees the GPU pool entry; // otherwise the next Claim gets the same PCI address while QEMU still // holds /dev/vfio/, causing "device or resource busy". - if pid, pidErr := utils.ReadPidFile(instance.ID); pidErr == nil && pid > 0 { + if pid, pidErr := hostprocess.ReadPidFile(instance.ID); pidErr == nil && pid > 0 { if proc, procErr := os.FindProcess(pid); procErr == nil { _ = proc.Signal(syscall.SIGKILL) } - _ = utils.WaitForProcessExit(pid, 10*time.Second) + _ = hostprocess.WaitForProcessExit(pid, 10*time.Second) } return err } @@ -402,7 +403,7 @@ const ( // startQEMU launches the QEMU process for instance and waits for startup to // confirm. func (m *Manager) startQEMU(instance *VM) error { - pidFile, err := utils.GeneratePidFile(instance.ID) + pidFile, err := hostprocess.GeneratePidFile(instance.ID) if err != nil { slog.Error("Failed to generate PID file", "err", err) return err @@ -416,7 +417,7 @@ func (m *Manager) startQEMU(instance *VM) error { return fmt.Errorf("instance type %s not found", instance.InstanceType) } - runtimeDir := utils.RuntimeDir() + runtimeDir := hostprocess.RuntimeDir() consoleLogPath := filepath.Join(runtimeDir, fmt.Sprintf("console-%s.log", instance.ID)) serialSocket := filepath.Join(runtimeDir, fmt.Sprintf("serial-%s.sock", instance.ID)) @@ -653,7 +654,7 @@ func (m *Manager) startQEMU(instance *VM) error { slog.Info("VM started successfully", "pid", cmd.Process.Pid) oomScore := guestOOMScore(instance.ManagedBy) - if err := utils.SetOOMScore(cmd.Process.Pid, oomScore); err != nil { + if err := hostprocess.SetOOMScore(cmd.Process.Pid, oomScore); err != nil { slog.Warn("Failed to set QEMU OOM score", "pid", cmd.Process.Pid, "score", oomScore, "err", err) } @@ -734,7 +735,7 @@ func (m *Manager) startQEMU(instance *VM) error { // Wait for the pidfile so we don't tear down the tap before QEMU finishes // attaching to it (can lag under post-reboot recovery load). - if _, err := utils.WaitForPidFile(instance.ID, 3*time.Second); err != nil { + if _, err := hostprocess.WaitForPidFile(instance.ID, 3*time.Second); err != nil { slog.Error("Failed to read PID file", "err", err) return err } @@ -953,7 +954,7 @@ func isTransientDialError(err error) bool { func newQMPClientWithHandshake(ctx context.Context, v *VM) (*qmp.QMPClient, error) { // QMP socket bind lags the pidfile under recovery load; wait for the // socket inode to exist before dialling to avoid an ENOENT race. - if err := utils.WaitForUnixSocket(v.Config.QMPSocket, qmpSocketWaitTimeout); err != nil { + if err := hostprocess.WaitForUnixSocket(v.Config.QMPSocket, qmpSocketWaitTimeout); err != nil { return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err) } client, err := dialQMPWithRetry(v.Config.QMPSocket, qmpGreetingTimeout(v)) @@ -1355,7 +1356,7 @@ func (m *Manager) appendSystemNetcfgFwCfg(instance *VM) error { // Management NIC: static, off br-mgmt, never the default route. fmt.Fprintf(&b, "NIC%d_MAC=%s\nNIC%d_CIDR=%s/24\nNIC%d_DEFAULT=0\n", n, instance.MgmtMAC, n, instance.MgmtIP, n) - path := filepath.Join(utils.RuntimeDir(), fmt.Sprintf("fwcfg-%s-netcfg.tmp", instance.ID)) + path := filepath.Join(hostprocess.RuntimeDir(), fmt.Sprintf("fwcfg-%s-netcfg.tmp", instance.ID)) if err := os.WriteFile(path, []byte(b.String()), 0o600); err != nil { return fmt.Errorf("write system netcfg blob: %w", err) } diff --git a/spinifex/runtime/compute/vm/orphan_qemu_reaper.go b/spinifex/runtime/compute/vm/orphan_qemu_reaper.go index 8379ffda6..1c00c29b6 100644 --- a/spinifex/runtime/compute/vm/orphan_qemu_reaper.go +++ b/spinifex/runtime/compute/vm/orphan_qemu_reaper.go @@ -6,7 +6,7 @@ import ( "log/slog" "time" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // orphanQEMUKillTimeout bounds how long the reaper waits for a SIGKILL'd @@ -63,23 +63,23 @@ func (r *OrphanQEMUReaper) Sweep(context.Context) (int, error) { reaped := 0 for _, v := range terminated { - pid, err := utils.ReadPidFile(v.ID) + pid, err := hostprocess.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 + if !hostprocess.ProcessAlive(pid) { + _ = hostprocess.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 { + if err := hostprocess.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) + _ = hostprocess.RemovePidFile(v.ID) reaped++ } @@ -94,7 +94,7 @@ func (r *OrphanQEMUReaper) Sweep(context.Context) (int, error) { if qemuProcessAlive(v.ID) { continue } - _ = utils.RemovePidFile(v.ID) // clear any stale file for the dead process + _ = hostprocess.RemovePidFile(v.ID) // clear any stale file for the dead process slog.Warn("vm/gc: reconciling wedged shutting-down instance, QEMU vanished", "instanceId", v.ID) @@ -114,11 +114,11 @@ func (r *OrphanQEMUReaper) Sweep(context.Context) (int, error) { // the reaper, where a terminated/shutting-down instance's absent PID file // legitimately means the process is gone. func qemuProcessAlive(instanceID string) bool { - pid, err := utils.ReadPidFile(instanceID) + pid, err := hostprocess.ReadPidFile(instanceID) if err != nil { return false } - return utils.ProcessAlive(pid) + return hostprocess.ProcessAlive(pid) } // qemuConfirmedDead reports whether the instance's QEMU is provably gone: its @@ -127,9 +127,9 @@ func qemuProcessAlive(instanceID string) bool { // still-running instance (DetachVolume) falls through to its normal QMP path // rather than short-circuiting on an absent file. func qemuConfirmedDead(instanceID string) bool { - pid, err := utils.ReadPidFile(instanceID) + pid, err := hostprocess.ReadPidFile(instanceID) if err != nil { return false } - return !utils.ProcessAlive(pid) + return !hostprocess.ProcessAlive(pid) } diff --git a/spinifex/runtime/compute/vm/orphan_qemu_reaper_test.go b/spinifex/runtime/compute/vm/orphan_qemu_reaper_test.go index 1ec6f391c..dec098e18 100644 --- a/spinifex/runtime/compute/vm/orphan_qemu_reaper_test.go +++ b/spinifex/runtime/compute/vm/orphan_qemu_reaper_test.go @@ -6,7 +6,7 @@ import ( "sync" "testing" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -31,7 +31,7 @@ func TestOrphanQEMUReaper(t *testing.T) { wg.Go(func() { _ = cmd.Wait() }) const id = "i-orphan-term" - require.NoError(t, utils.WritePidFile(id, pid)) + require.NoError(t, hostprocess.WritePidFile(id, pid)) store.terminated[id] = &VM{ID: id, Status: StateTerminated} reaped, err := reaper.Sweep(context.Background()) @@ -39,9 +39,9 @@ func TestOrphanQEMUReaper(t *testing.T) { wg.Wait() assert.Equal(t, 1, reaped, "the orphan QEMU for a terminated instance must be reaped") - assert.False(t, utils.ProcessAlive(pid), + assert.False(t, hostprocess.ProcessAlive(pid), "a terminated instance must have no surviving process — otherwise it holds OVN ports ") - _, perr := utils.ReadPidFile(id) + _, perr := hostprocess.ReadPidFile(id) assert.Error(t, perr, "the PID file must be removed after reaping") }) @@ -57,13 +57,13 @@ func TestOrphanQEMUReaper(t *testing.T) { 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 + require.NoError(t, hostprocess.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) + _, perr := hostprocess.ReadPidFile(id) assert.Error(t, perr, "the stale PID file must be cleaned up") }) @@ -76,12 +76,12 @@ func TestOrphanQEMUReaper(t *testing.T) { 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)) + require.NoError(t, hostprocess.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") + assert.True(t, hostprocess.ProcessAlive(pid), "a non-terminated instance's process must survive") }) t.Run("reconciles a wedged shutting-down instance whose QEMU vanished", func(t *testing.T) { @@ -122,7 +122,7 @@ func TestOrphanQEMUReaper(t *testing.T) { const id = "i-terminating" m.InsertIfAbsent(&VM{ID: id, Status: StateShuttingDown}) - require.NoError(t, utils.WritePidFile(id, pid)) + require.NoError(t, hostprocess.WritePidFile(id, pid)) reaped, err := reaper.Sweep(context.Background()) require.NoError(t, err) diff --git a/spinifex/runtime/compute/vm/orphan_scan.go b/spinifex/runtime/compute/vm/orphan_scan.go index 6545410cb..4f4c087a6 100644 --- a/spinifex/runtime/compute/vm/orphan_scan.go +++ b/spinifex/runtime/compute/vm/orphan_scan.go @@ -7,7 +7,7 @@ import ( "strconv" "strings" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // qemuProcessPrefix matches both qemu-system-x86_64 and qemu-system-aarch64; @@ -60,7 +60,7 @@ func pidFileOwners(runtimeDir string) map[int]string { if !ok { continue } - if pid, err := utils.ReadPidFileFrom(runtimeDir, id); err == nil { + if pid, err := hostprocess.ReadPidFileFrom(runtimeDir, id); err == nil { owners[pid] = id } } @@ -116,7 +116,7 @@ func (m *Manager) reportRecordlessQEMUOrphans() { known[instance.ID] = true } - orphans, err := classifyQEMUOrphans(qemuProcRoot, utils.RuntimeDir(), known) + orphans, err := classifyQEMUOrphans(qemuProcRoot, hostprocess.RuntimeDir(), known) if err != nil { slog.Warn("recordless QEMU orphan scan failed", "error", err) return diff --git a/spinifex/runtime/compute/vm/restore.go b/spinifex/runtime/compute/vm/restore.go index 7b20dd71a..c5886fc3b 100644 --- a/spinifex/runtime/compute/vm/restore.go +++ b/spinifex/runtime/compute/vm/restore.go @@ -14,6 +14,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -278,7 +279,7 @@ func markUnschedulable(instance *VM, reason string) { // Returns true when the process is gone and classification can proceed. // SIGKILL is used directly to avoid the 120s SIGTERM timeout blocking startup. func killOrphanedQEMU(instance *VM) bool { - pid, pidErr := utils.ReadPidFile(instance.ID) + pid, pidErr := hostprocess.ReadPidFile(instance.ID) if pidErr != nil || pid <= 0 { slog.Error("Cannot read PID for orphaned QEMU, skipping relaunch", "instanceId", instance.ID, "err", pidErr) @@ -289,12 +290,12 @@ func killOrphanedQEMU(instance *VM) bool { } // SIGKILL cannot be caught; QEMU never runs its cleanup so the PID // file stays on disk. Wait for the process to die, then remove it. - if err := utils.WaitForProcessExit(pid, 10*time.Second); err != nil { + if err := hostprocess.WaitForProcessExit(pid, 10*time.Second); err != nil { slog.Error("Orphaned QEMU did not exit after SIGKILL, skipping relaunch", "instanceId", instance.ID, "pid", pid, "err", err) return false } - _ = utils.RemovePidFile(instance.ID) + _ = hostprocess.RemovePidFile(instance.ID) return true } @@ -501,7 +502,7 @@ func (m *Manager) reconnectInstance(instance *VM) error { // isInstanceProcessRunning reports whether the QEMU process in the PID file // is still alive. Returns false on any failure (missing file, dead PID). func isInstanceProcessRunning(instance *VM) bool { - pid, err := utils.ReadPidFile(instance.ID) + pid, err := hostprocess.ReadPidFile(instance.ID) if err != nil || pid <= 0 { return false } diff --git a/spinifex/runtime/compute/vm/restore_test.go b/spinifex/runtime/compute/vm/restore_test.go index d402d1660..2d6980892 100644 --- a/spinifex/runtime/compute/vm/restore_test.go +++ b/spinifex/runtime/compute/vm/restore_test.go @@ -18,7 +18,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -277,7 +277,7 @@ func TestClassifyRestoredInstances_ShuttingDownWithLiveQEMUTerminates(t *testing m, store, _ := classifyTestManager(t) t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) v := &VM{ID: "i-shutdown-live", Status: StateShuttingDown, InstanceType: "t3.micro"} - require.NoError(t, utils.WritePidFile(v.ID, spawnLiveChild(t))) + require.NoError(t, hostprocess.WritePidFile(v.ID, spawnLiveChild(t))) m.Replace(map[string]*VM{v.ID: v}) toLaunch := m.classifyRestoredInstances() @@ -295,7 +295,7 @@ func TestClassifyRestoredInstances_StoppingWithLiveQEMUStops(t *testing.T) { m, store, _ := classifyTestManager(t) t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) v := &VM{ID: "i-stopping-live", Status: StateStopping, InstanceType: "t3.micro"} - require.NoError(t, utils.WritePidFile(v.ID, spawnLiveChild(t))) + require.NoError(t, hostprocess.WritePidFile(v.ID, spawnLiveChild(t))) m.Replace(map[string]*VM{v.ID: v}) toLaunch := m.classifyRestoredInstances() @@ -530,7 +530,7 @@ func TestIsInstanceProcessRunning(t *testing.T) { t.Run("signal nil returns true", func(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-alive", os.Getpid())) + require.NoError(t, hostprocess.WritePidFile("i-alive", os.Getpid())) instance := &VM{ID: "i-alive"} assert.True(t, isInstanceProcessRunning(instance), @@ -540,7 +540,7 @@ func TestIsInstanceProcessRunning(t *testing.T) { t.Run("signal ESRCH returns false", func(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) dead := findDeadPID(t) - require.NoError(t, utils.WritePidFile("i-esrch", dead)) + require.NoError(t, hostprocess.WritePidFile("i-esrch", dead)) instance := &VM{ID: "i-esrch"} assert.False(t, isInstanceProcessRunning(instance), @@ -552,7 +552,7 @@ func TestIsInstanceProcessRunning(t *testing.T) { t.Skip("running as root: cannot trigger EPERM by signalling PID 1") } t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-eperm", 1)) + require.NoError(t, hostprocess.WritePidFile("i-eperm", 1)) instance := &VM{ID: "i-eperm"} assert.False(t, isInstanceProcessRunning(instance), @@ -590,12 +590,12 @@ func TestKillOrphanedQEMU(t *testing.T) { t.Run("dead pid: signal error swallowed, wait succeeds, pid file removed", func(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) dead := findDeadPID(t) - require.NoError(t, utils.WritePidFile("i-dead", dead)) + require.NoError(t, hostprocess.WritePidFile("i-dead", dead)) ok := killOrphanedQEMU(&VM{ID: "i-dead"}) assert.True(t, ok, "WaitForProcessExit returns nil for an already-dead pid; caller proceeds with relaunch") - _, readErr := utils.ReadPidFile("i-dead") + _, readErr := hostprocess.ReadPidFile("i-dead") assert.Error(t, readErr, "PID file must be removed after a successful wait") }) @@ -616,12 +616,12 @@ func TestKillOrphanedQEMU(t *testing.T) { _ = cmd.Process.Kill() <-reaped }) - require.NoError(t, utils.WritePidFile("i-real", cmd.Process.Pid)) + require.NoError(t, hostprocess.WritePidFile("i-real", cmd.Process.Pid)) ok := killOrphanedQEMU(&VM{ID: "i-real"}) assert.True(t, ok, "SIGKILLing the live subprocess must report success") - _, readErr := utils.ReadPidFile("i-real") + _, readErr := hostprocess.ReadPidFile("i-real") assert.Error(t, readErr, "PID file must be removed on the success path") }) } diff --git a/spinifex/runtime/compute/vm/shutdown.go b/spinifex/runtime/compute/vm/shutdown.go index adf7c0949..3ef0065e6 100644 --- a/spinifex/runtime/compute/vm/shutdown.go +++ b/spinifex/runtime/compute/vm/shutdown.go @@ -14,7 +14,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // pidFileRemovalTimeout is how long Stop/Terminate wait for the PID file to @@ -43,7 +43,7 @@ var errPowerdownTimedOut = errors.New("guest did not power down within its budge // exits on powerdown instead, and a reboot would have nothing left to reset. // Indirected so a test can describe a guest without one running behind it. var qemuPausesOnShutdown = func(instance *VM) bool { - pid, err := utils.ReadPidFile(instance.ID) + pid, err := hostprocess.ReadPidFile(instance.ID) if err != nil || pid <= 0 { return false } @@ -453,14 +453,14 @@ func (m *Manager) reconcileVanishedQEMU(instance *VM) error { // teardown failed for TerminatedTeardownReaper, and drives the record to // terminated. The cleaner is idempotent, so racing the wedged goroutine is safe. func (m *Manager) forceFinalizeStuckTerminate(instance *VM) error { - if pid, err := utils.ReadPidFile(instance.ID); err == nil && utils.ProcessAlive(pid) { + if pid, err := hostprocess.ReadPidFile(instance.ID); err == nil && hostprocess.ProcessAlive(pid) { slog.Warn("Force-killing wedged QEMU for stuck terminate", "instanceId", instance.ID, "pid", pid) - if err := utils.ForceKillProcess(pid, orphanQEMUKillTimeout); err != nil { + if err := hostprocess.ForceKillProcess(pid, orphanQEMUKillTimeout); err != nil { slog.Error("Failed to kill wedged QEMU, continuing finalize", "instanceId", instance.ID, "pid", pid, "err", err) } - _ = utils.RemovePidFile(instance.ID) + _ = hostprocess.RemovePidFile(instance.ID) } m.markTeardown(instance, TeardownQEMU, TeardownDone) @@ -710,14 +710,14 @@ func (m *Manager) shutdownQEMU(instance *VM) { // 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 { + if err := hostprocess.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) + pid, readErr := hostprocess.ReadPidFile(instance.ID) if readErr != nil { slog.Debug("No PID file found (VM likely already stopped)", "id", instance.ID) - } else if utils.ProcessAlive(pid) { + } else if hostprocess.ProcessAlive(pid) { slog.Info("Force killing process", "pid", pid, "id", instance.ID) - if err := utils.KillProcess(pid); err != nil { + if err := hostprocess.KillProcess(pid); err != nil { slog.Error("Failed to kill process", "pid", pid, "id", instance.ID, "err", err) } } diff --git a/spinifex/runtime/compute/vm/shutdown_test.go b/spinifex/runtime/compute/vm/shutdown_test.go index c3fcbfc04..0513a103b 100644 --- a/spinifex/runtime/compute/vm/shutdown_test.go +++ b/spinifex/runtime/compute/vm/shutdown_test.go @@ -13,7 +13,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/goleak" @@ -31,7 +31,7 @@ const markFailedDeadline = 10 * time.Second // what cleanup ran (and what didn't). // // Sets XDG_RUNTIME_DIR to a per-test tempdir so PID-file paths -// (utils.WaitForPidFileRemoval, ReadPidFile) cannot collide between +// (hostprocess.WaitForPidFileRemoval, ReadPidFile) cannot collide between // concurrent or sequential tests sharing the host's real runtime dir. func shutdownTestManager(t *testing.T) (m *Manager, store *fakeStateStore, mounter *fakeVolumeMounter, cleaner *recordingInstanceCleaner, rt *recordedTransitions) { t.Helper() @@ -889,7 +889,7 @@ func (f *failingSaveRunningStore) SaveRunningState(string, map[string]*VM) error // recording cleaner. Returns everything callers may need to assert on. // // Sets XDG_RUNTIME_DIR to a per-test tempdir so PID-file paths -// (utils.WaitForPidFileRemoval, ReadPidFile invoked from +// (hostprocess.WaitForPidFileRemoval, ReadPidFile invoked from // shutdownAndUnmount) cannot collide between tests. func terminateTestManager(t *testing.T, store StateStore) (m *Manager, cleaner *recordingInstanceCleaner, rt *recordedTransitions, downCount *atomic.Int64, downIDs *[]string) { t.Helper() @@ -1467,14 +1467,14 @@ func TestShutdownAndUnmount_PowerdownSent_NoForceKill(t *testing.T) { // runs and the unmount step still fires. func TestShutdownAndUnmount_PIDFileRemovedBeforeTimeout_NoForceKill(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-pid-removed", os.Getpid())) + require.NoError(t, hostprocess.WritePidFile("i-pid-removed", os.Getpid())) mounter := &fakeVolumeMounter{} m := NewManagerWithDeps(Deps{VolumeMounter: mounter}) go func() { time.Sleep(150 * time.Millisecond) - _ = utils.RemovePidFile("i-pid-removed") + _ = hostprocess.RemovePidFile("i-pid-removed") }() instance := &VM{ID: "i-pid-removed", QMPClient: nil} @@ -1651,8 +1651,8 @@ func TestTerminate_SealFailure_Tolerated(t *testing.T) { // guest that was only busy booting, so the button is pressed again. func TestGracefulPowerdown_ResendReachesALateGuest(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-late", os.Getpid())) - t.Cleanup(func() { _ = utils.RemovePidFile("i-late") }) + require.NoError(t, hostprocess.WritePidFile("i-late", os.Getpid())) + t.Cleanup(func() { _ = hostprocess.RemovePidFile("i-late") }) var presses int recorder := &qmpRecorder{} @@ -1686,8 +1686,8 @@ func TestGracefulPowerdown_ResendReachesALateGuest(t *testing.T) { // to SIGKILL on and Reboot escalates to a hard reset on. func TestGracefulPowerdown_WedgedGuestTimesOut(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-wedged", os.Getpid())) - t.Cleanup(func() { _ = utils.RemovePidFile("i-wedged") }) + require.NoError(t, hostprocess.WritePidFile("i-wedged", os.Getpid())) + t.Cleanup(func() { _ = hostprocess.RemovePidFile("i-wedged") }) qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { if cmd.Execute == "query-status" { diff --git a/spinifex/runtime/compute/vm/stuck_terminate_reaper_test.go b/spinifex/runtime/compute/vm/stuck_terminate_reaper_test.go index 77845e294..2ed660271 100644 --- a/spinifex/runtime/compute/vm/stuck_terminate_reaper_test.go +++ b/spinifex/runtime/compute/vm/stuck_terminate_reaper_test.go @@ -8,7 +8,7 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -38,7 +38,7 @@ func TestStuckTerminateReaper(t *testing.T) { wg.Go(func() { _ = cmd.Wait() }) const id = "i-wedged-live" - require.NoError(t, utils.WritePidFile(id, pid)) + require.NoError(t, hostprocess.WritePidFile(id, pid)) m.InsertIfAbsent(&VM{ ID: id, Status: StateShuttingDown, @@ -51,7 +51,7 @@ func TestStuckTerminateReaper(t *testing.T) { wg.Wait() assert.Equal(t, 1, reaped, "a terminate wedged past the timeout must be force-completed") - assert.False(t, utils.ProcessAlive(pid), "the wedged QEMU must be force-killed") + assert.False(t, hostprocess.ProcessAlive(pid), "the wedged QEMU must be force-killed") _, ok := m.Get(id) assert.False(t, ok, "the finalized instance must leave the local running map") diff --git a/spinifex/runtime/compute/vm/telemetry_meta.go b/spinifex/runtime/compute/vm/telemetry_meta.go index 55be3d642..672b11b6b 100644 --- a/spinifex/runtime/compute/vm/telemetry_meta.go +++ b/spinifex/runtime/compute/vm/telemetry_meta.go @@ -8,7 +8,7 @@ import ( "github.com/aws/aws-sdk-go/aws" telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // telemetryPeriodSeconds maps the instance's EC2 monitoring tier to the @@ -69,7 +69,7 @@ func writeTelemetryMeta(v *VM) error { if err != nil { return err } - return os.WriteFile(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID), data, 0o600) + return os.WriteFile(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID), data, 0o600) } // refreshTelemetryMeta rewrites the discovery file after an ENI change; safe @@ -93,5 +93,5 @@ func removeTelemetryArtifacts(v *VM) { if v.Config.TelemetryQMPSocket != "" { _ = os.Remove(v.Config.TelemetryQMPSocket) } - _ = os.Remove(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID)) + _ = os.Remove(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID)) } diff --git a/spinifex/runtime/compute/vm/telemetry_meta_test.go b/spinifex/runtime/compute/vm/telemetry_meta_test.go index c8b601714..acac78a6f 100644 --- a/spinifex/runtime/compute/vm/telemetry_meta_test.go +++ b/spinifex/runtime/compute/vm/telemetry_meta_test.go @@ -9,7 +9,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) func TestTelemetryPeriodSeconds(t *testing.T) { @@ -63,15 +63,15 @@ func TestTelemetryMetaRoundtrip(t *testing.T) { if err := writeTelemetryMeta(v); err != nil { t.Fatalf("no-op write: %v", err) } - if _, err := os.Stat(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID)); !os.IsNotExist(err) { + if _, err := os.Stat(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID)); !os.IsNotExist(err) { t.Fatal("no file expected without a telemetry socket") } - v.Config.TelemetryQMPSocket = telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID) + ".sock" + v.Config.TelemetryQMPSocket = telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID) + ".sock" if err := writeTelemetryMeta(v); err != nil { t.Fatalf("write: %v", err) } - data, err := os.ReadFile(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID)) + data, err := os.ReadFile(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID)) if err != nil { t.Fatalf("read back: %v", err) } @@ -86,7 +86,7 @@ func TestTelemetryMetaRoundtrip(t *testing.T) { } removeTelemetryArtifacts(v) - if _, err := os.Stat(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID)); !os.IsNotExist(err) { + if _, err := os.Stat(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID)); !os.IsNotExist(err) { t.Error("metadata must be removed") } } @@ -97,11 +97,11 @@ func TestRefreshTelemetryMetaCarriesTierChange(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) v := &VM{ID: "i-0abc", AccountID: "123456789012", RunInstancesInput: &ec2.RunInstancesInput{}} - v.Config.TelemetryQMPSocket = telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID) + ".sock" + v.Config.TelemetryQMPSocket = telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID) + ".sock" readPeriod := func() int { t.Helper() - data, err := os.ReadFile(telemetryv1.GuestTelemetryMetaPath(utils.RuntimeDir(), v.ID)) + data, err := os.ReadFile(telemetryv1.GuestTelemetryMetaPath(hostprocess.RuntimeDir(), v.ID)) if err != nil { t.Fatalf("read back: %v", err) } diff --git a/spinifex/runtime/compute/vm/volumes_test.go b/spinifex/runtime/compute/vm/volumes_test.go index c6ae12765..29049d4c4 100644 --- a/spinifex/runtime/compute/vm/volumes_test.go +++ b/spinifex/runtime/compute/vm/volumes_test.go @@ -18,7 +18,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -863,8 +863,8 @@ func TestQMPHeartbeatPoll_UsesRunstate(t *testing.T) { m := NewManager() instance := &VM{ID: "i-heartbeat-runstate", Status: StateRunning, QMPClient: qmpClient} m.Insert(instance) - require.NoError(t, utils.WritePidFile(instance.ID, os.Getpid())) - t.Cleanup(func() { _ = utils.RemovePidFile(instance.ID) }) + require.NoError(t, hostprocess.WritePidFile(instance.ID, os.Getpid())) + t.Cleanup(func() { _ = hostprocess.RemovePidFile(instance.ID) }) for range QMPMaxConsecutiveFailures { assert.True(t, m.qmpHeartbeatPoll(instance)) @@ -1570,7 +1570,7 @@ func TestDetachVolume_ConfirmedDeadQEMU_SkipsQMPAndSeals(t *testing.T) { defer cancel() const id = "i-1" - require.NoError(t, utils.WritePidFile(id, 999999)) // PID file present but process dead + require.NoError(t, hostprocess.WritePidFile(id, 999999)) // PID file present but process dead mounter := &fakeVolumeMounter{} updater := &fakeVolumeStateUpdater{} @@ -1634,8 +1634,8 @@ func TestDetachVolume_BlockdevDelAlreadyRemoved_IdempotentSuccess(t *testing.T) func TestReboot_WedgedGuestFallsBackToHardReset(t *testing.T) { pausingGuest(t) t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) - require.NoError(t, utils.WritePidFile("i-1", os.Getpid())) - t.Cleanup(func() { _ = utils.RemovePidFile("i-1") }) + require.NoError(t, hostprocess.WritePidFile("i-1", os.Getpid())) + t.Cleanup(func() { _ = hostprocess.RemovePidFile("i-1") }) previous := rebootPowerdownTimeout rebootPowerdownTimeout = 30 * time.Millisecond diff --git a/spinifex/utils/pidfile.go b/spinifex/runtime/host/process/pidfile.go similarity index 96% rename from spinifex/utils/pidfile.go rename to spinifex/runtime/host/process/pidfile.go index 28c902acd..c20bb81c0 100644 --- a/spinifex/utils/pidfile.go +++ b/spinifex/runtime/host/process/pidfile.go @@ -1,4 +1,4 @@ -package utils +package process import ( "bytes" @@ -221,3 +221,14 @@ func WaitForPidFileRemoval(instanceID string, timeout time.Duration) error { } } } + +func dirExists(path string) bool { + info, err := os.Stat(path) + if os.IsNotExist(err) { + return false + } + if err != nil { + return false + } + return info.IsDir() +} diff --git a/spinifex/runtime/host/process/pidfile_test.go b/spinifex/runtime/host/process/pidfile_test.go new file mode 100644 index 000000000..7aa7ab72c --- /dev/null +++ b/spinifex/runtime/host/process/pidfile_test.go @@ -0,0 +1,472 @@ +package process + +import ( + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGeneratePidFile(t *testing.T) { + // Simulate a sample process running (e.g cat) + cmd := exec.Command("cat") + cmd.Start() + + err := WritePidFile("utilsunittest", cmd.Process.Pid) + + assert.NoError(t, err) + + // Read the PID file and verify contents + pid, err := ReadPidFile("utilsunittest") + + assert.NoError(t, err) + assert.Equal(t, cmd.Process.Pid, pid) + + // Test attempt to read a PID file that doesn't exist + _, err = ReadPidFile("nonexistentpidfile") + assert.Error(t, err) + + // Cleanup + err = RemovePidFile("utilsunittest") + assert.NoError(t, err) + + // Give some time before killing the process + //time.Sleep(2 * time.Second) + + // Simulate process ending +} + +func TestExecProcessAndKill(t *testing.T) { + // Simulate a sample process running (e.g sleep, 30 secs) + cmd := exec.Command("sleep", "30") + + // Detach: new process group, no controlling terminal. + cmd.SysProcAttr = &syscall.SysProcAttr{ + Setpgid: true, // put child in new process group + } + + // Make it fully background-friendly: + // - close stdio so parent doesn't block on pipes + cmd.Stdin = nil + cmd.Stdout = nil + cmd.Stderr = nil + + // Start (non-blocking). If command is missing, error. + if err := cmd.Start(); err != nil { + assert.Fail(t, "Failed to start command", err) + } + + // IMPORTANT: reap the child to avoid zombies. + // Since we're "backgrounding" it, do Wait() in a goroutine. + go func(c *exec.Cmd) { + t.Log("Waiting for command to finish...") + _ = c.Wait() // ignore error; ensures kernel reaps the process + t.Log("Command finished.") + }(cmd) + + err := WritePidFile("utilsunittest", cmd.Process.Pid) + + t.Log("Started process with PID:", cmd.Process.Pid) + + assert.NoError(t, err) + + // Test PID file removed + err = WaitForPidFileRemoval("utilsunittest", 100*time.Millisecond) + assert.Error(t, err) // Should timeout since file should still exist + + time.Sleep(20 * time.Millisecond) + + // Kill the process + err = StopProcessAt("", "utilsunittest") + assert.NoError(t, err) + + // Test PID file removed + err = WaitForPidFileRemoval("utilsunittest", 1*time.Second) + assert.NoError(t, err) // Should timeout since file should still exist + + // Verify process is killed + err = cmd.Process.Signal(syscall.Signal(0)) + assert.Error(t, err) // Should return an error since process is killed +} + +func TestWaitForPidFile(t *testing.T) { + t.Run("returns pid when file appears within timeout", func(t *testing.T) { + const name = "wait-pidfile-appears" + _ = RemovePidFile(name) + t.Cleanup(func() { _ = RemovePidFile(name) }) + + go func() { + time.Sleep(150 * time.Millisecond) + _ = WritePidFile(name, 4242) + }() + + pid, err := WaitForPidFile(name, time.Second) + require.NoError(t, err) + assert.Equal(t, 4242, pid) + }) + + t.Run("returns error when timeout expires", func(t *testing.T) { + const name = "wait-pidfile-missing" + _ = RemovePidFile(name) + + _, err := WaitForPidFile(name, 100*time.Millisecond) + assert.Error(t, err) + }) + + t.Run("returns immediately when file already present", func(t *testing.T) { + const name = "wait-pidfile-present" + require.NoError(t, WritePidFile(name, 7777)) + t.Cleanup(func() { _ = RemovePidFile(name) }) + + start := time.Now() + pid, err := WaitForPidFile(name, time.Second) + require.NoError(t, err) + assert.Equal(t, 7777, pid) + assert.Less(t, time.Since(start), 50*time.Millisecond) + }) +} + +func TestWaitForUnixSocket(t *testing.T) { + t.Run("returns nil when socket appears within timeout", func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "appears.sock") + + ln, err := net.Listen("unix", path) + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + + require.NoError(t, WaitForUnixSocket(path, time.Second)) + }) + + t.Run("returns error when timeout expires", func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "missing.sock") + + err := WaitForUnixSocket(path, 100*time.Millisecond) + require.Error(t, err) + }) + + t.Run("waits for socket created after a delay", func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "delayed.sock") + + go func() { + time.Sleep(150 * time.Millisecond) + ln, err := net.Listen("unix", path) + if err == nil { + t.Cleanup(func() { _ = ln.Close() }) + } + }() + + require.NoError(t, WaitForUnixSocket(path, time.Second)) + }) +} + +func TestKillProcess(t *testing.T) { + // Create a test process + cmd := exec.Command("sleep", "60") + err := cmd.Start() + require.NoError(t, err) + + pid := cmd.Process.Pid + + // Reap in background so KillProcess can detect termination + var wg sync.WaitGroup + wg.Go(func() { + _ = cmd.Wait() + }) + + err = KillProcess(pid) + assert.NoError(t, err) + wg.Wait() + + // Test killing non-existent process + err = KillProcess(999999) + 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") +} + +// TestForceKillProcess_AlreadyExitedIsSuccess pins the contract callers rely on. +// This is asked "make sure this process is gone", and one that died on its own +// is as gone as one it killed — reporting a failure there makes a caller treat +// a stopped writer as a live one. +func TestForceKillProcess_AlreadyExitedIsSuccess(t *testing.T) { + cmd := exec.Command("true") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + require.NoError(t, cmd.Wait()) + + assert.NoError(t, ForceKillProcess(pid, time.Second), + "an already-exited process satisfies the request, it does not fail it") +} + +func TestGeneratePidFile_EmptyName(t *testing.T) { + _, err := GeneratePidFile("") + assert.Error(t, err) +} + +func TestWritePidFileTo(t *testing.T) { + dir := t.TempDir() + + cmd := exec.Command("cat") + require.NoError(t, cmd.Start()) + defer cmd.Process.Kill() + + // Write PID to custom directory + err := WritePidFileTo(dir, "testservice", cmd.Process.Pid) + require.NoError(t, err) + + // Read it back from the same directory + pid, err := ReadPidFileFrom(dir, "testservice") + require.NoError(t, err) + assert.Equal(t, cmd.Process.Pid, pid) + + // Clean up + err = RemovePidFileAt(dir, "testservice") + assert.NoError(t, err) + + // Verify it's gone + _, err = ReadPidFileFrom(dir, "testservice") + assert.Error(t, err) +} + +func TestWritePidFileTo_EmptyDir(t *testing.T) { + // With empty dir, should fall back to default RuntimeDir() + cmd := exec.Command("cat") + require.NoError(t, cmd.Start()) + defer cmd.Process.Kill() + + err := WritePidFileTo("", "pidto-fallback", cmd.Process.Pid) + require.NoError(t, err) + + // Should be readable via the default ReadPidFile + pid, err := ReadPidFile("pidto-fallback") + require.NoError(t, err) + assert.Equal(t, cmd.Process.Pid, pid) + + // Clean up + RemovePidFile("pidto-fallback") +} + +func TestStopProcessAt(t *testing.T) { + dir := t.TempDir() + + // Start a process we can kill + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + + // Write PID file + err := WritePidFileTo(dir, "stopat-test", cmd.Process.Pid) + require.NoError(t, err) + + // Reap in background so StopProcessAt can detect termination + var wg sync.WaitGroup + wg.Go(func() { + _ = cmd.Wait() + }) + + err = StopProcessAt(dir, "stopat-test") + assert.NoError(t, err) + wg.Wait() + + // Verify PID file was removed + _, err = ReadPidFileFrom(dir, "stopat-test") + assert.Error(t, err, "PID file should be removed") +} + +func TestStopProcessAt_StaleProcess(t *testing.T) { + dir := t.TempDir() + + // Start a process and let it exit, leaving a stale PID file + cmd := exec.Command("true") + require.NoError(t, cmd.Start()) + stalePid := cmd.Process.Pid + require.NoError(t, cmd.Wait()) + + // Write stale PID file + err := WritePidFileTo(dir, "stale-test", stalePid) + require.NoError(t, err) + + // StopProcessAt should return an error (process is dead) but still + // clean up the PID file + err = StopProcessAt(dir, "stale-test") + assert.Error(t, err, "should error because process is already dead") + + // PID file must be removed despite the kill error + _, err = ReadPidFileFrom(dir, "stale-test") + assert.Error(t, err, "PID file should be removed even when process is already dead") +} + +func TestStopProcessAt_NoPidFile(t *testing.T) { + dir := t.TempDir() + err := StopProcessAt(dir, "nonexistent") + assert.Error(t, err, "should error when PID file does not exist") +} + +func TestSetOOMScore(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("OOM score adjustment only supported on Linux") + } + + // Read current value so we can set something higher (unprivileged processes + // can only increase their OOM score, not decrease it) + pid := os.Getpid() + current, err := os.ReadFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid)) + if err != nil { + t.Skipf("Cannot read OOM score: %v", err) + } + + // Set a positive score (always allowed for unprivileged processes) + err = SetOOMScore(pid, 100) + if err != nil { + t.Skipf("Insufficient permissions to set OOM score: %v", err) + } + + // Read back and verify + data, err := os.ReadFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid)) + assert.NoError(t, err) + assert.Equal(t, "100", strings.TrimSpace(string(data))) + + // Best-effort restore (may fail without privileges if original was lower) + _ = os.WriteFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid), current, 0644) +} + +func TestSetOOMScore_InvalidPID(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("OOM score adjustment only supported on Linux") + } + + err := SetOOMScore(999999999, 100) + assert.Error(t, err) +} + +func TestRuntimeDir(t *testing.T) { + dir := RuntimeDir() + assert.NotEmpty(t, dir) +} + +func TestPidPath_XDG(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", "/tmp/test-xdg-runtime") + assert.Equal(t, "/tmp/test-xdg-runtime", RuntimeDir()) +} + +func TestPidPath_HomeSpinifexFallback(t *testing.T) { + tmpHome := t.TempDir() + spinifexDir := fmt.Sprintf("%s/spinifex", tmpHome) + require.NoError(t, os.Mkdir(spinifexDir, 0755)) + + t.Setenv("XDG_RUNTIME_DIR", "") + t.Setenv("HOME", tmpHome) + + assert.Equal(t, spinifexDir, RuntimeDir()) +} + +func TestPidPath_TempDirFallback(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", "") + t.Setenv("HOME", "/nonexistent-home-dir-utils-test") + + assert.Equal(t, os.TempDir(), RuntimeDir()) +} + +func TestWritePidFile_CreateError(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", "/nonexistent-dir-utils-test") + err := WritePidFile("testservice", 12345) + assert.Error(t, err) +} + +func TestRemovePidFileAt_EmptyDir(t *testing.T) { + err := RemovePidFileAt("", fmt.Sprintf("nonexistent-service-%d", time.Now().UnixNano())) + assert.Error(t, err) +} + +func TestGeneratePidFile_InvalidPath(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", "/nonexistent-dir-utils-test") + path, err := GeneratePidFile("test") + // GeneratePidFile doesn't check if path exists, just builds it + assert.NoError(t, err) + assert.Contains(t, path, "test.pid") +} + +func TestReadPidFileFrom_EmptyDir(t *testing.T) { + _, err := ReadPidFileFrom("", fmt.Sprintf("nonexistent-service-%d", time.Now().UnixNano())) + assert.Error(t, err) +} + +func TestWaitForProcessExit_ProcessAlreadyDead(t *testing.T) { + cmd := exec.Command("true") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + require.NoError(t, cmd.Wait()) + + // `true` has exited and been reaped — WaitForProcessExit should return + // nil immediately on the first tick. + err := WaitForProcessExit(pid, 2*time.Second) + assert.NoError(t, err) +} + +func TestWaitForProcessExit_ProcessExitsBeforeTimeout(t *testing.T) { + cmd := exec.Command("sleep", "0.1") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + + // Reap in the background so the kernel releases the PID once sleep exits. + go func() { _ = cmd.Wait() }() + + err := WaitForProcessExit(pid, 5*time.Second) + assert.NoError(t, err) +} + +func TestWaitForProcessExit_Timeout(t *testing.T) { + cmd := exec.Command("sleep", "30") + require.NoError(t, cmd.Start()) + defer func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + }() + + err := WaitForProcessExit(cmd.Process.Pid, 200*time.Millisecond) + require.Error(t, err) + assert.Contains(t, err.Error(), "timeout") +} diff --git a/spinifex/utils/process.go b/spinifex/runtime/host/process/process.go similarity index 99% rename from spinifex/utils/process.go rename to spinifex/runtime/host/process/process.go index ade890d6b..0a6b83ac6 100644 --- a/spinifex/utils/process.go +++ b/spinifex/runtime/host/process/process.go @@ -1,4 +1,4 @@ -package utils +package process import ( "errors" diff --git a/spinifex/utils/process_terminate_test.go b/spinifex/runtime/host/process/process_terminate_test.go similarity index 82% rename from spinifex/utils/process_terminate_test.go rename to spinifex/runtime/host/process/process_terminate_test.go index 57c7a75dd..ea643599a 100644 --- a/spinifex/utils/process_terminate_test.go +++ b/spinifex/runtime/host/process/process_terminate_test.go @@ -1,4 +1,4 @@ -package utils_test +package process_test import ( "bufio" @@ -9,7 +9,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // The unmount path calls this on an nbdkit that is already sealing, so the @@ -23,7 +23,7 @@ func TestTerminateProcessReturnsAsSoonAsTheProcessExits(t *testing.T) { go func() { _ = cmd.Wait(); close(done) }() start := time.Now() - exited := utils.TerminateProcess(pid, 10*time.Second) + exited := hostprocess.TerminateProcess(pid, 10*time.Second) elapsed := time.Since(start) assert.True(t, exited, "a process that dies on SIGTERM must be reported as exited") @@ -56,13 +56,13 @@ func TestTerminateProcessGivesUpAtTheGraceAndDoesNotKill(t *testing.T) { const grace = 200 * time.Millisecond start := time.Now() - exited := utils.TerminateProcess(pid, grace) + exited := hostprocess.TerminateProcess(pid, grace) elapsed := time.Since(start) assert.False(t, exited) assert.GreaterOrEqual(t, elapsed, grace, "must wait out the grace before giving up") assert.Less(t, elapsed, grace+2*time.Second, "must be bounded by the grace, not hang") - assert.True(t, utils.ProcessAlive(pid), "escalation belongs to the caller, not here") + assert.True(t, hostprocess.ProcessAlive(pid), "escalation belongs to the caller, not here") } func TestTerminateProcessAlreadyExitedIsSuccess(t *testing.T) { @@ -71,6 +71,6 @@ func TestTerminateProcessAlreadyExitedIsSuccess(t *testing.T) { pid := cmd.Process.Pid require.NoError(t, cmd.Wait()) - assert.True(t, utils.TerminateProcess(pid, time.Second)) - assert.False(t, utils.TerminateProcess(0, time.Second), "an invalid pid is not an exit") + assert.True(t, hostprocess.TerminateProcess(pid, time.Second)) + assert.False(t, hostprocess.TerminateProcess(0, time.Second), "an invalid pid is not an exit") } diff --git a/spinifex/utils/process_test.go b/spinifex/runtime/host/process/process_test.go similarity index 99% rename from spinifex/utils/process_test.go rename to spinifex/runtime/host/process/process_test.go index f5c121e75..098a4253a 100644 --- a/spinifex/utils/process_test.go +++ b/spinifex/runtime/host/process/process_test.go @@ -1,4 +1,4 @@ -package utils +package process import ( "bufio" diff --git a/spinifex/runtime/roles/nats/nats.go b/spinifex/runtime/roles/nats/nats.go index 2acd5abe9..911994ee0 100644 --- a/spinifex/runtime/roles/nats/nats.go +++ b/spinifex/runtime/roles/nats/nats.go @@ -5,7 +5,7 @@ import ( "log/slog" "os" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/nats-io/nats-server/v2/server" ) @@ -37,7 +37,7 @@ func New(config any) (svc *Service, err error) { } func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.DataDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.DataDir, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } err := launchService(svc.Config) diff --git a/spinifex/runtime/roles/qemunbdd/service.go b/spinifex/runtime/roles/qemunbdd/service.go index 885ad60f4..cfe4773db 100644 --- a/spinifex/runtime/roles/qemunbdd/service.go +++ b/spinifex/runtime/roles/qemunbdd/service.go @@ -10,6 +10,7 @@ import ( "syscall" "github.com/mulgadc/spinifex/spinifex/providers/ebs/natsserve" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -44,7 +45,7 @@ func New(config any) (svc *Service, err error) { } func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } if err := launchService(svc.Config); err != nil { diff --git a/spinifex/runtime/roles/qmpcollector/qmpcollector.go b/spinifex/runtime/roles/qmpcollector/qmpcollector.go index 318c0939c..3cdaec6b5 100644 --- a/spinifex/runtime/roles/qmpcollector/qmpcollector.go +++ b/spinifex/runtime/roles/qmpcollector/qmpcollector.go @@ -17,6 +17,7 @@ import ( "syscall" "time" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -36,7 +37,7 @@ type Config struct { NodeName string // RuntimeDir overrides where qmp-telemetry-* sockets and metadata live - // (default utils.RuntimeDir()); ProcRoot / SysRoot override /proc and + // (default hostprocess.RuntimeDir()); ProcRoot / SysRoot override /proc and // /sys. Test injection points. RuntimeDir string ProcRoot string @@ -61,7 +62,7 @@ func New(config any) (*Service, error) { // Start runs the collector until SIGINT/SIGTERM. func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } @@ -74,7 +75,7 @@ func (svc *Service) Start() (int, error) { func launchService(cfg *Config) error { if cfg.RuntimeDir == "" { - cfg.RuntimeDir = utils.RuntimeDir() + cfg.RuntimeDir = hostprocess.RuntimeDir() } if cfg.ProcRoot == "" { cfg.ProcRoot = "/proc" diff --git a/spinifex/runtime/roles/spinifex/spinifex.go b/spinifex/runtime/roles/spinifex/spinifex.go index 06ba5b031..f4b1c6992 100644 --- a/spinifex/runtime/roles/spinifex/spinifex.go +++ b/spinifex/runtime/roles/spinifex/spinifex.go @@ -7,7 +7,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/bootstrap/config" "github.com/mulgadc/spinifex/spinifex/daemon" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) var serviceName = "spinifex" @@ -33,7 +33,7 @@ func (svc *Service) SetConfigPath(path string) { } func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.NodeBaseDir(), serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.NodeBaseDir(), serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } err := launchService(svc.Config, svc.ConfigPath) diff --git a/spinifex/runtime/roles/spinifexui/spinifexui.go b/spinifex/runtime/roles/spinifexui/spinifexui.go index 39628cad7..ed0be828a 100644 --- a/spinifex/runtime/roles/spinifexui/spinifexui.go +++ b/spinifex/runtime/roles/spinifexui/spinifexui.go @@ -21,7 +21,7 @@ import ( "github.com/go-chi/chi/v5/middleware" "github.com/mulgadc/bluebottle/pkg/tlsconfig" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) var serviceName = "spinifex-ui" @@ -126,7 +126,7 @@ func New(config any) (*Service, error) { // Start starts the spinifex-ui service. func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { slog.Error("Failed to write pid file", "err", err) } diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 173b107fe..b1f87dd4c 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -42,6 +42,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/providers/objectstore" "github.com/mulgadc/spinifex/spinifex/runtime/compute/cache" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -80,7 +81,7 @@ func New(cfg any) (svc *Service, err error) { } func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.NodeBaseDir(), serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.NodeBaseDir(), serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } err := launchService(svc.Config) diff --git a/spinifex/services/northstar/northstar.go b/spinifex/services/northstar/northstar.go index 4d866ceac..8cafa5762 100644 --- a/spinifex/services/northstar/northstar.go +++ b/spinifex/services/northstar/northstar.go @@ -14,6 +14,7 @@ import ( nsconfig "github.com/mulgadc/northstar/pkg/config" nsserver "github.com/mulgadc/northstar/pkg/server" handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -65,7 +66,7 @@ func (svc *Service) Start() (int, error) { } svc.server = server - if err := utils.WritePidFileTo(svc.Config.BasePath, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BasePath, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } diff --git a/spinifex/services/predastore/predastore.go b/spinifex/services/predastore/predastore.go index 2f2d4663f..88fa8469a 100644 --- a/spinifex/services/predastore/predastore.go +++ b/spinifex/services/predastore/predastore.go @@ -11,7 +11,7 @@ import ( "github.com/mulgadc/bluebottle/pkg/masterkey" pds "github.com/mulgadc/predastore" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) var serviceName = "predastore" @@ -67,7 +67,7 @@ func (svc *Service) Start() (int, error) { return 0, fmt.Errorf("predastore encryption key file is required (set EncryptionKeyFile)") } - if err := utils.WritePidFileTo(svc.Config.BasePath, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BasePath, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } diff --git a/spinifex/services/viperblockd/nbdkit_recovery_test.go b/spinifex/services/viperblockd/nbdkit_recovery_test.go index adee3159c..bb7778915 100644 --- a/spinifex/services/viperblockd/nbdkit_recovery_test.go +++ b/spinifex/services/viperblockd/nbdkit_recovery_test.go @@ -19,7 +19,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/providers/ebs" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/viperblock/viperblock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -377,7 +377,7 @@ func TestReapOrphanedNbdkit_NoReferencer_SignalsAndRemovesSocket(t *testing.T) { disc := discoveredNbdkit{PID: pid, Volume: "vol-reap-unreferenced1", Socket: socket} reapOrphanedNbdkit(procRoot, disc, reapReasonUnadoptable) - require.Eventually(t, func() bool { return !utils.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, + require.Eventually(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, "an unreferenced orphan must be signalled and exit") _, err := os.Stat(socket) assert.True(t, os.IsNotExist(err), "the reaped orphan's socket file must be removed") @@ -406,7 +406,7 @@ func TestReapOrphanedNbdkit_LiveReferencer_LeftRunningAndLogged(t *testing.T) { // Give a real signal every chance to have landed before asserting it did not. time.Sleep(100 * time.Millisecond) - assert.True(t, utils.ProcessAlive(pid), "a referenced orphan must never be signalled") + assert.True(t, hostprocess.ProcessAlive(pid), "a referenced orphan must never be signalled") _, err := os.Stat(socket) assert.NoError(t, err, "a live orphan's socket must not be removed") assert.Contains(t, logs.String(), "left running deliberately") @@ -430,7 +430,7 @@ func TestReapOrphanedNbdkit_TCPEndpointReferencedAcrossSplitBlockdevArgs(t *test reapOrphanedNbdkit(procRoot, disc, reapReasonUnadoptable) time.Sleep(100 * time.Millisecond) - assert.True(t, utils.ProcessAlive(pid), "a TCP endpoint split across server.host=/server.port= must still be recognized as referenced") + assert.True(t, hostprocess.ProcessAlive(pid), "a TCP endpoint split across server.host=/server.port= must still be recognized as referenced") } // unreadableProcDir creates procRoot/ with a cmdline file inside, then @@ -503,7 +503,7 @@ func TestReapOrphanedNbdkit_UnreadableCandidate_LeftRunningAndLogged(t *testing. reapOrphanedNbdkit(procRoot, disc, reapReasonUnadoptable) time.Sleep(100 * time.Millisecond) - assert.True(t, utils.ProcessAlive(pid), "an inconclusive scan must never result in a signal") + assert.True(t, hostprocess.ProcessAlive(pid), "an inconclusive scan must never result in a signal") _, err := os.Stat(socket) assert.NoError(t, err, "an orphan left running because the scan was inconclusive must keep its socket") assert.Contains(t, logs.String(), "could not scan for processes") @@ -577,7 +577,7 @@ func TestReapOrphanedNbdkit_HidepidInvisible_DeclinesEvenOnACleanScan(t *testing reapOrphanedNbdkit(procRoot, disc, reapReasonUnadoptable) time.Sleep(100 * time.Millisecond) - assert.True(t, utils.ProcessAlive(pid), "hidepid=invisible must never be read as a clean 'no referencer' scan") + assert.True(t, hostprocess.ProcessAlive(pid), "hidepid=invisible must never be read as a clean 'no referencer' scan") _, err := os.Stat(socket) assert.NoError(t, err, "an orphan left running because entries could be hidden must keep its socket") assert.Contains(t, logs.String(), "hides other users' process entries") @@ -629,9 +629,9 @@ func TestRecoverMountedVolumes_DuplicateVolume_UnusedDuplicateReaped(t *testing. } } - require.Eventually(t, func() bool { return !utils.ProcessAlive(duplicatePID) }, 2*time.Second, 20*time.Millisecond, + require.Eventually(t, func() bool { return !hostprocess.ProcessAlive(duplicatePID) }, 2*time.Second, 20*time.Millisecond, "the unused duplicate must be reaped") - assert.True(t, utils.ProcessAlive(mv.PID), "the adopted process must never be signalled") + assert.True(t, hostprocess.ProcessAlive(mv.PID), "the adopted process must never be signalled") _, err := os.Stat(duplicateSocket) assert.True(t, os.IsNotExist(err), "the reaped duplicate's socket file must be removed") } @@ -661,7 +661,7 @@ func TestRecoverMountedVolumes_UnadoptableReapedAcrossRestarts(t *testing.T) { logs := captureLogs(t) recoverMountedVolumes(context.Background(), cfg, nc, procRoot) - require.Eventually(t, func() bool { return !utils.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, + require.Eventually(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, "an unadoptable backend with no referencer must be reaped on its first recovery pass") assert.Contains(t, logs.String(), "reaped unclaimed nbdkit process") _, ok := findMountedVolume(cfg, volumeName) @@ -779,7 +779,7 @@ func TestReapUnregisteredNbdkit_ReapsAnExportTheRegistryNeverClaimed(t *testing. assert.True(t, reapUnregisteredNbdkit(context.Background(), cfg, volumeName), "an export serving a volume with no registry entry must be reported as reaped") - require.Eventually(t, func() bool { return !utils.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, + require.Eventually(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, "the unregistered export must be signalled and exit") _, err := os.Stat(socket) assert.True(t, os.IsNotExist(err), "the reaped export's socket file must be removed") @@ -796,7 +796,7 @@ func TestReapUnregisteredNbdkit_LeavesAnotherVolumeAlone(t *testing.T) { assert.False(t, reapUnregisteredNbdkit(context.Background(), cfg, "vol-unreg0002"), "the boot volume's unmount must not claim to have reaped the EFI volume's export") - assert.True(t, utils.ProcessAlive(pid), "an export for a different volume must be left running") + assert.True(t, hostprocess.ProcessAlive(pid), "an export for a different volume must be left running") _, err := os.Stat(socket) assert.NoError(t, err, "an export for a different volume must keep its socket") } @@ -811,7 +811,7 @@ func TestReapUnregisteredNbdkit_LeavesAForeignExportAlone(t *testing.T) { assert.False(t, reapUnregisteredNbdkit(context.Background(), cfg, volumeName), "an export for a different data dir belongs to a different daemon") - assert.True(t, utils.ProcessAlive(pid), "a foreign export must be left running") + assert.True(t, hostprocess.ProcessAlive(pid), "a foreign export must be left running") _, err := os.Stat(socket) assert.NoError(t, err, "a foreign export must keep its socket") } @@ -827,7 +827,7 @@ func TestReapUnregisteredNbdkit_LeavesAReferencedExportAlone(t *testing.T) { logs := captureLogs(t) assert.False(t, reapUnregisteredNbdkit(context.Background(), cfg, volumeName), "an export something still references must not be reported as reaped") - assert.True(t, utils.ProcessAlive(pid), "an export a live process references must be left running") + assert.True(t, hostprocess.ProcessAlive(pid), "an export a live process references must be left running") assert.Contains(t, logs.String(), "still referenced by a live process") } @@ -844,6 +844,6 @@ func TestUnmountVolume_ReapsAnUnregisteredExport(t *testing.T) { require.Error(t, err, "the volume is genuinely not mounted here, reaped or not") assert.True(t, resp.NotFound, "an unregistered volume has no entry to seal") assert.True(t, resp.Reaped, "the export the unmount reaped must be reported") - require.Eventually(t, func() bool { return !utils.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, + require.Eventually(t, func() bool { return !hostprocess.ProcessAlive(pid) }, 2*time.Second, 20*time.Millisecond, "unmount must leave no nbdkit serving the volume it just removed") } diff --git a/spinifex/services/viperblockd/provider_handlers.go b/spinifex/services/viperblockd/provider_handlers.go index 3e8558fe2..10a1ec723 100644 --- a/spinifex/services/viperblockd/provider_handlers.go +++ b/spinifex/services/viperblockd/provider_handlers.go @@ -21,6 +21,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/providers/ebs" "github.com/mulgadc/spinifex/spinifex/providers/objectstore" "github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" vbtypes "github.com/mulgadc/viperblock/types" "github.com/mulgadc/viperblock/viperblock" @@ -1440,7 +1441,7 @@ func mountVolume(ctx context.Context, cfg *Config, nc *nats.Conn, volumeName str } // Generate PID file for nbdkit process - nbdPidFile, err := utils.GeneratePidFile(fmt.Sprintf("nbdkit-vol-%s", volumeName)) + nbdPidFile, err := hostprocess.GeneratePidFile(fmt.Sprintf("nbdkit-vol-%s", volumeName)) if err != nil { slog.ErrorContext(ctx, "Failed to generate nbdkit pid file", "err", err) ebsResponse.Error = fmt.Sprintf("failed to generate pid file: %v", err) @@ -1631,13 +1632,13 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (viperbl // drains to the backend and leaves a receipt. SIGKILLing into that // discards the drain and leaves the WAL for the fallback seal below to // replay, which costs far more than the caller's deadline allows. - utils.TerminateProcess(matched.PID, pluginSealGrace) + hostprocess.TerminateProcess(matched.PID, pluginSealGrace) // The seal below rewrites the directory nbdkit writes, so a kill that // did not take makes it a concurrent write to that directory. Fail the // unmount instead: the entry stays mounted and a retry re-attempts // both, which is what a failed seal already does. - if err := utils.ForceKillProcess(matched.PID, fenceKillTimeout); err != nil { + if err := hostprocess.ForceKillProcess(matched.PID, fenceKillTimeout); err != nil { slog.ErrorContext(ctx, "ebs.unmount: nbdkit did not exit, refusing to seal underneath a live writer", "volume", matched.Name, "pid", matched.PID, "err", err) ebsResponse.Error = fmt.Sprintf("kill nbdkit for %s: %v", matched.Name, err) @@ -1851,7 +1852,7 @@ func mountEntryIsStale(mv MountedVolume) (bool, string) { return true, "socket is gone" } } - if mv.PID > 0 && !utils.ProcessAlive(mv.PID) { + if mv.PID > 0 && !hostprocess.ProcessAlive(mv.PID) { return true, "nbdkit process is gone" } return false, "" diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 7a956ba08..d96432d8d 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -20,6 +20,7 @@ import ( "github.com/mulgadc/bluebottle/pkg/masterkey" viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/viperblock/viperblock" @@ -205,7 +206,7 @@ type Config struct { constructVB func(ctx context.Context, volumeName string) (*viperblock.VB, int, error) // fenceWatchEvery and processAlive drive the watch a failed fence keeps on - // its writer. Zero means the production interval and utils.ProcessAlive: a + // its writer. Zero means the production interval and hostprocess.ProcessAlive: a // SIGKILL that does not take effect cannot be staged with a real process. fenceWatchEvery time.Duration processAlive func(pid int) bool @@ -662,7 +663,7 @@ func respondAndPublish(msg *nats.Msg, nc *nats.Conn, topic string, data any) { } func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) } err := launchService(svc.Config) @@ -768,7 +769,7 @@ func launchService(cfg *Config) (err error) { if matched.VB != nil { matched.VB.Detach() } - if err := utils.KillProcess(matched.PID); err != nil { + if err := hostprocess.KillProcess(matched.PID); err != nil { slog.ErrorContext(ctx, "Failed to kill nbdkit process", "pid", matched.PID, "err", err) } @@ -1083,7 +1084,7 @@ func launchService(cfg *Config) (err error) { // drain (or unmount) path owns reaping in-use nbdkit after the guest is gone. // // The reap itself fans out one goroutine per idle volume so the wall-clock -// cost is bounded by the slowest single nbdkit's utils.KillProcess grace, not +// cost is bounded by the slowest single nbdkit's hostprocess.KillProcess grace, not // the sum across every mounted volume — the caller returns (and the process // exits) only once every goroutine below has finished. func shutdownVolumes(volumes []MountedVolume, inUse func(MountedVolume) bool) { @@ -1101,7 +1102,7 @@ func shutdownVolumes(volumes []MountedVolume, inUse func(MountedVolume) bool) { go func(volume MountedVolume) { defer wg.Done() slog.Info("Killing idle nbdkit process", "pid", volume.PID, "name", volume.Name) - if err := utils.KillProcess(volume.PID); err != nil { + if err := hostprocess.KillProcess(volume.PID); err != nil { slog.Error("Failed to kill nbdkit process", "pid", volume.PID, "err", err) } }(volume) diff --git a/spinifex/services/viperblockd/volume_abandon_test.go b/spinifex/services/viperblockd/volume_abandon_test.go index 672a66580..4d86fee67 100644 --- a/spinifex/services/viperblockd/volume_abandon_test.go +++ b/spinifex/services/viperblockd/volume_abandon_test.go @@ -7,7 +7,7 @@ import ( "context" "testing" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -31,7 +31,7 @@ func TestVolumeAbandon_GivesUpTheExportAndTheLease(t *testing.T) { mounted := len(cfg.MountedVolumes) cfg.mu.Unlock() assert.Zero(t, mounted, "an export left up is a second writer, whoever decided to stop") - assert.False(t, utils.ProcessAlive(pid), "the export process has to be gone, not merely forgotten") + assert.False(t, hostprocess.ProcessAlive(pid), "the export process has to be gone, not merely forgotten") } // The difference from a fence, and the reason this path is worth having. A diff --git a/spinifex/services/viperblockd/volume_fence.go b/spinifex/services/viperblockd/volume_fence.go index 4f342e024..a2a3fd238 100644 --- a/spinifex/services/viperblockd/volume_fence.go +++ b/spinifex/services/viperblockd/volume_fence.go @@ -10,8 +10,8 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire" - "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" ) @@ -119,7 +119,7 @@ func (cfg *Config) writerAlive(pid int) bool { if cfg.processAlive != nil { return cfg.processAlive(pid) } - return utils.ProcessAlive(pid) + return hostprocess.ProcessAlive(pid) } // fenceWatchInterval is how often a failed fence re-checks its writer. @@ -252,7 +252,7 @@ func (cfg *Config) tearDownExport(ctx context.Context, matched MountedVolume, wh matched.VB.Detach() } - if err := utils.ForceKillProcess(matched.PID, fenceKillTimeout); err != nil { + if err := hostprocess.ForceKillProcess(matched.PID, fenceKillTimeout); err != nil { return err } diff --git a/spinifex/services/viperblockd/volume_fence_test.go b/spinifex/services/viperblockd/volume_fence_test.go index 05413da03..29bc50ac7 100644 --- a/spinifex/services/viperblockd/volume_fence_test.go +++ b/spinifex/services/viperblockd/volume_fence_test.go @@ -10,7 +10,7 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/spinifex/utils" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -81,7 +81,7 @@ func TestVolumeFence_TearsDownWhenAnotherNodeHoldsTheLease(t *testing.T) { mounted := len(cfg.MountedVolumes) cfg.mu.Unlock() assert.Zero(t, mounted, "a fenced volume must not be left exported: the export is the second writer") - assert.False(t, utils.ProcessAlive(pid), "the export process has to be gone, not merely forgotten") + assert.False(t, hostprocess.ProcessAlive(pid), "the export process has to be gone, not merely forgotten") } // TestVolumeFence_FencesAnEntryNobodyHolds covers the lapsed case. An entry that diff --git a/spinifex/utils/nbd.go b/spinifex/utils/nbd.go index 0aa658d20..77aa1a2bc 100644 --- a/spinifex/utils/nbd.go +++ b/spinifex/utils/nbd.go @@ -9,6 +9,8 @@ import ( "strconv" "strings" "time" + + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" ) // GenerateSocketFile generates a socket file path for the given name. @@ -18,7 +20,7 @@ func GenerateSocketFile(name string) (string, error) { return "", errors.New("name is required") } - pidPath := RuntimeDir() + pidPath := hostprocess.RuntimeDir() if pidPath == "" { return "", errors.New("pid path is empty") @@ -49,7 +51,7 @@ func NBDSocketDir() string { if dirExists(systemdNBDDir) { return systemdNBDDir } - return RuntimeDir() + return hostprocess.RuntimeDir() } // IsSocketURI reports whether the NBD URI refers to a Unix socket (ends with ".sock" or contains "unix:"). @@ -78,7 +80,7 @@ func WaitForNBDReady(uri string, timeout time.Duration) error { } switch serverType { case "unix": - return WaitForUnixSocket(path, timeout) + return hostprocess.WaitForUnixSocket(path, timeout) case "inet": return waitForTCPListener(net.JoinHostPort(host, strconv.Itoa(port)), timeout) default: diff --git a/spinifex/utils/utils_test.go b/spinifex/utils/utils_test.go index 75ca75ffe..7743adf52 100644 --- a/spinifex/utils/utils_test.go +++ b/spinifex/utils/utils_test.go @@ -12,12 +12,9 @@ import ( "os" "os/exec" "path/filepath" - "runtime" "strconv" "strings" - "sync" "sync/atomic" - "syscall" "testing" "time" @@ -52,35 +49,6 @@ func TestGenerateResourceID(t *testing.T) { } } -func TestGeneratePidFile(t *testing.T) { - // Simulate a sample process running (e.g cat) - cmd := exec.Command("cat") - cmd.Start() - - err := WritePidFile("utilsunittest", cmd.Process.Pid) - - assert.NoError(t, err) - - // Read the PID file and verify contents - pid, err := ReadPidFile("utilsunittest") - - assert.NoError(t, err) - assert.Equal(t, cmd.Process.Pid, pid) - - // Test attempt to read a PID file that doesn't exist - _, err = ReadPidFile("nonexistentpidfile") - assert.Error(t, err) - - // Cleanup - err = RemovePidFile("utilsunittest") - assert.NoError(t, err) - - // Give some time before killing the process - //time.Sleep(2 * time.Second) - - // Simulate process ending -} - func TestGenerateSocketFile(t *testing.T) { socketPath := fmt.Sprintf("%s/%s", t.TempDir(), "utilsunittest") @@ -96,132 +64,6 @@ func TestGenerateSocketFile(t *testing.T) { assert.Error(t, err) } -func TestExecProcessAndKill(t *testing.T) { - // Simulate a sample process running (e.g sleep, 30 secs) - cmd := exec.Command("sleep", "30") - - // Detach: new process group, no controlling terminal. - cmd.SysProcAttr = &syscall.SysProcAttr{ - Setpgid: true, // put child in new process group - } - - // Make it fully background-friendly: - // - close stdio so parent doesn't block on pipes - cmd.Stdin = nil - cmd.Stdout = nil - cmd.Stderr = nil - - // Start (non-blocking). If command is missing, error. - if err := cmd.Start(); err != nil { - assert.Fail(t, "Failed to start command", err) - } - - // IMPORTANT: reap the child to avoid zombies. - // Since we're "backgrounding" it, do Wait() in a goroutine. - go func(c *exec.Cmd) { - t.Log("Waiting for command to finish...") - _ = c.Wait() // ignore error; ensures kernel reaps the process - t.Log("Command finished.") - }(cmd) - - err := WritePidFile("utilsunittest", cmd.Process.Pid) - - t.Log("Started process with PID:", cmd.Process.Pid) - - assert.NoError(t, err) - - // Test PID file removed - err = WaitForPidFileRemoval("utilsunittest", 100*time.Millisecond) - assert.Error(t, err) // Should timeout since file should still exist - - time.Sleep(20 * time.Millisecond) - - // Kill the process - err = StopProcessAt("", "utilsunittest") - assert.NoError(t, err) - - // Test PID file removed - err = WaitForPidFileRemoval("utilsunittest", 1*time.Second) - assert.NoError(t, err) // Should timeout since file should still exist - - // Verify process is killed - err = cmd.Process.Signal(syscall.Signal(0)) - assert.Error(t, err) // Should return an error since process is killed -} - -func TestWaitForPidFile(t *testing.T) { - t.Run("returns pid when file appears within timeout", func(t *testing.T) { - const name = "wait-pidfile-appears" - _ = RemovePidFile(name) - t.Cleanup(func() { _ = RemovePidFile(name) }) - - go func() { - time.Sleep(150 * time.Millisecond) - _ = WritePidFile(name, 4242) - }() - - pid, err := WaitForPidFile(name, time.Second) - require.NoError(t, err) - assert.Equal(t, 4242, pid) - }) - - t.Run("returns error when timeout expires", func(t *testing.T) { - const name = "wait-pidfile-missing" - _ = RemovePidFile(name) - - _, err := WaitForPidFile(name, 100*time.Millisecond) - assert.Error(t, err) - }) - - t.Run("returns immediately when file already present", func(t *testing.T) { - const name = "wait-pidfile-present" - require.NoError(t, WritePidFile(name, 7777)) - t.Cleanup(func() { _ = RemovePidFile(name) }) - - start := time.Now() - pid, err := WaitForPidFile(name, time.Second) - require.NoError(t, err) - assert.Equal(t, 7777, pid) - assert.Less(t, time.Since(start), 50*time.Millisecond) - }) -} - -func TestWaitForUnixSocket(t *testing.T) { - t.Run("returns nil when socket appears within timeout", func(t *testing.T) { - dir := t.TempDir() - path := filepath.Join(dir, "appears.sock") - - ln, err := net.Listen("unix", path) - require.NoError(t, err) - t.Cleanup(func() { _ = ln.Close() }) - - require.NoError(t, WaitForUnixSocket(path, time.Second)) - }) - - t.Run("returns error when timeout expires", func(t *testing.T) { - dir := t.TempDir() - path := filepath.Join(dir, "missing.sock") - - err := WaitForUnixSocket(path, 100*time.Millisecond) - require.Error(t, err) - }) - - t.Run("waits for socket created after a delay", func(t *testing.T) { - dir := t.TempDir() - path := filepath.Join(dir, "delayed.sock") - - go func() { - time.Sleep(150 * time.Millisecond) - ln, err := net.Listen("unix", path) - if err == nil { - t.Cleanup(func() { _ = ln.Close() }) - } - }() - - require.NoError(t, WaitForUnixSocket(path, time.Second)) - }) -} - func TestWaitForNBDReady(t *testing.T) { t.Run("unix socket ready", func(t *testing.T) { dir := t.TempDir() @@ -509,76 +351,6 @@ func TestParseNBDURI(t *testing.T) { } } -func TestKillProcess(t *testing.T) { - // Create a test process - cmd := exec.Command("sleep", "60") - err := cmd.Start() - require.NoError(t, err) - - pid := cmd.Process.Pid - - // Reap in background so KillProcess can detect termination - var wg sync.WaitGroup - wg.Go(func() { - _ = cmd.Wait() - }) - - err = KillProcess(pid) - assert.NoError(t, err) - wg.Wait() - - // Test killing non-existent process - err = KillProcess(999999) - 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") -} - -// TestForceKillProcess_AlreadyExitedIsSuccess pins the contract callers rely on. -// This is asked "make sure this process is gone", and one that died on its own -// is as gone as one it killed — reporting a failure there makes a caller treat -// a stopped writer as a live one. -func TestForceKillProcess_AlreadyExitedIsSuccess(t *testing.T) { - cmd := exec.Command("true") - require.NoError(t, cmd.Start()) - pid := cmd.Process.Pid - require.NoError(t, cmd.Wait()) - - assert.NoError(t, ForceKillProcess(pid, time.Second), - "an already-exited process satisfies the request, it does not fail it") -} - // Test file extraction process func TestExtractDiskImageFromFile(t *testing.T) { @@ -736,179 +508,11 @@ func TestDirExists(t *testing.T) { assert.False(t, dirExists(tmpFile.Name())) } -func TestGeneratePidFile_EmptyName(t *testing.T) { - _, err := GeneratePidFile("") - assert.Error(t, err) -} - -func TestWritePidFileTo(t *testing.T) { - dir := t.TempDir() - - cmd := exec.Command("cat") - require.NoError(t, cmd.Start()) - defer cmd.Process.Kill() - - // Write PID to custom directory - err := WritePidFileTo(dir, "testservice", cmd.Process.Pid) - require.NoError(t, err) - - // Read it back from the same directory - pid, err := ReadPidFileFrom(dir, "testservice") - require.NoError(t, err) - assert.Equal(t, cmd.Process.Pid, pid) - - // Clean up - err = RemovePidFileAt(dir, "testservice") - assert.NoError(t, err) - - // Verify it's gone - _, err = ReadPidFileFrom(dir, "testservice") - assert.Error(t, err) -} - -func TestWritePidFileTo_EmptyDir(t *testing.T) { - // With empty dir, should fall back to default RuntimeDir() - cmd := exec.Command("cat") - require.NoError(t, cmd.Start()) - defer cmd.Process.Kill() - - err := WritePidFileTo("", "pidto-fallback", cmd.Process.Pid) - require.NoError(t, err) - - // Should be readable via the default ReadPidFile - pid, err := ReadPidFile("pidto-fallback") - require.NoError(t, err) - assert.Equal(t, cmd.Process.Pid, pid) - - // Clean up - RemovePidFile("pidto-fallback") -} - -func TestStopProcessAt(t *testing.T) { - dir := t.TempDir() - - // Start a process we can kill - cmd := exec.Command("sleep", "60") - require.NoError(t, cmd.Start()) - - // Write PID file - err := WritePidFileTo(dir, "stopat-test", cmd.Process.Pid) - require.NoError(t, err) - - // Reap in background so StopProcessAt can detect termination - var wg sync.WaitGroup - wg.Go(func() { - _ = cmd.Wait() - }) - - err = StopProcessAt(dir, "stopat-test") - assert.NoError(t, err) - wg.Wait() - - // Verify PID file was removed - _, err = ReadPidFileFrom(dir, "stopat-test") - assert.Error(t, err, "PID file should be removed") -} - -func TestStopProcessAt_StaleProcess(t *testing.T) { - dir := t.TempDir() - - // Start a process and let it exit, leaving a stale PID file - cmd := exec.Command("true") - require.NoError(t, cmd.Start()) - stalePid := cmd.Process.Pid - require.NoError(t, cmd.Wait()) - - // Write stale PID file - err := WritePidFileTo(dir, "stale-test", stalePid) - require.NoError(t, err) - - // StopProcessAt should return an error (process is dead) but still - // clean up the PID file - err = StopProcessAt(dir, "stale-test") - assert.Error(t, err, "should error because process is already dead") - - // PID file must be removed despite the kill error - _, err = ReadPidFileFrom(dir, "stale-test") - assert.Error(t, err, "PID file should be removed even when process is already dead") -} - -func TestStopProcessAt_NoPidFile(t *testing.T) { - dir := t.TempDir() - err := StopProcessAt(dir, "nonexistent") - assert.Error(t, err, "should error when PID file does not exist") -} - func TestGenerateSocketFile_EmptyName(t *testing.T) { _, err := GenerateSocketFile("") assert.Error(t, err) } -func TestSetOOMScore(t *testing.T) { - if runtime.GOOS != "linux" { - t.Skip("OOM score adjustment only supported on Linux") - } - - // Read current value so we can set something higher (unprivileged processes - // can only increase their OOM score, not decrease it) - pid := os.Getpid() - current, err := os.ReadFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid)) - if err != nil { - t.Skipf("Cannot read OOM score: %v", err) - } - - // Set a positive score (always allowed for unprivileged processes) - err = SetOOMScore(pid, 100) - if err != nil { - t.Skipf("Insufficient permissions to set OOM score: %v", err) - } - - // Read back and verify - data, err := os.ReadFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid)) - assert.NoError(t, err) - assert.Equal(t, "100", strings.TrimSpace(string(data))) - - // Best-effort restore (may fail without privileges if original was lower) - _ = os.WriteFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid), current, 0644) -} - -func TestSetOOMScore_InvalidPID(t *testing.T) { - if runtime.GOOS != "linux" { - t.Skip("OOM score adjustment only supported on Linux") - } - - err := SetOOMScore(999999999, 100) - assert.Error(t, err) -} - -func TestRuntimeDir(t *testing.T) { - dir := RuntimeDir() - assert.NotEmpty(t, dir) -} - -func TestPidPath_XDG(t *testing.T) { - t.Setenv("XDG_RUNTIME_DIR", "/tmp/test-xdg-runtime") - assert.Equal(t, "/tmp/test-xdg-runtime", RuntimeDir()) -} - -func TestPidPath_HomeSpinifexFallback(t *testing.T) { - tmpHome := t.TempDir() - spinifexDir := fmt.Sprintf("%s/spinifex", tmpHome) - require.NoError(t, os.Mkdir(spinifexDir, 0755)) - - t.Setenv("XDG_RUNTIME_DIR", "") - t.Setenv("HOME", tmpHome) - - assert.Equal(t, spinifexDir, RuntimeDir()) -} - -func TestPidPath_TempDirFallback(t *testing.T) { - t.Setenv("XDG_RUNTIME_DIR", "") - t.Setenv("HOME", "/nonexistent-home-dir-utils-test") - - assert.Equal(t, os.TempDir(), RuntimeDir()) -} - func TestExtractDiskImagePath_NoMatch(t *testing.T) { output := []byte("somefile.txt\nanotherfile.conf\n") diskimage, err := extractDiskImagePath(t.TempDir(), output) @@ -922,67 +526,6 @@ func TestExtractDiskImagePath_EmptyOutput(t *testing.T) { assert.NoError(t, err) } -func TestWritePidFile_CreateError(t *testing.T) { - t.Setenv("XDG_RUNTIME_DIR", "/nonexistent-dir-utils-test") - err := WritePidFile("testservice", 12345) - assert.Error(t, err) -} - -func TestRemovePidFileAt_EmptyDir(t *testing.T) { - err := RemovePidFileAt("", fmt.Sprintf("nonexistent-service-%d", time.Now().UnixNano())) - assert.Error(t, err) -} - -func TestGeneratePidFile_InvalidPath(t *testing.T) { - t.Setenv("XDG_RUNTIME_DIR", "/nonexistent-dir-utils-test") - path, err := GeneratePidFile("test") - // GeneratePidFile doesn't check if path exists, just builds it - assert.NoError(t, err) - assert.Contains(t, path, "test.pid") -} - -func TestReadPidFileFrom_EmptyDir(t *testing.T) { - _, err := ReadPidFileFrom("", fmt.Sprintf("nonexistent-service-%d", time.Now().UnixNano())) - assert.Error(t, err) -} - -func TestWaitForProcessExit_ProcessAlreadyDead(t *testing.T) { - cmd := exec.Command("true") - require.NoError(t, cmd.Start()) - pid := cmd.Process.Pid - require.NoError(t, cmd.Wait()) - - // `true` has exited and been reaped — WaitForProcessExit should return - // nil immediately on the first tick. - err := WaitForProcessExit(pid, 2*time.Second) - assert.NoError(t, err) -} - -func TestWaitForProcessExit_ProcessExitsBeforeTimeout(t *testing.T) { - cmd := exec.Command("sleep", "0.1") - require.NoError(t, cmd.Start()) - pid := cmd.Process.Pid - - // Reap in the background so the kernel releases the PID once sleep exits. - go func() { _ = cmd.Wait() }() - - err := WaitForProcessExit(pid, 5*time.Second) - assert.NoError(t, err) -} - -func TestWaitForProcessExit_Timeout(t *testing.T) { - cmd := exec.Command("sleep", "30") - require.NoError(t, cmd.Start()) - defer func() { - _ = cmd.Process.Kill() - _ = cmd.Wait() - }() - - err := WaitForProcessExit(cmd.Process.Pid, 200*time.Millisecond) - require.Error(t, err) - assert.Contains(t, err.Error(), "timeout") -} - // writeTempImage writes content to a file named name under t.TempDir() and // returns the absolute path. func writeTempImage(t *testing.T, name string, content []byte) string { diff --git a/spinifex/vpcd/vpcd.go b/spinifex/vpcd/vpcd.go index 318bf4949..7727dbba0 100644 --- a/spinifex/vpcd/vpcd.go +++ b/spinifex/vpcd/vpcd.go @@ -31,6 +31,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/network/subscribers" "github.com/mulgadc/spinifex/spinifex/network/topology" hostcommand "github.com/mulgadc/spinifex/spinifex/runtime/host/command" + hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -201,7 +202,7 @@ func New(config any) (*Service, error) { // Start starts the vpcd service. func (svc *Service) Start() (int, error) { - if err := utils.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { + if err := hostprocess.WritePidFileTo(svc.Config.BaseDir, serviceName, os.Getpid()); err != nil { return 0, fmt.Errorf("write pid file: %w", err) }