refactor: move process supervision helpers to runtime host

This commit is contained in:
joshsiv-mulga
2026-10-02 15:34:38 +10:00
parent 3f6f7a5e53
commit fd6c12f140
42 changed files with 662 additions and 623 deletions
+2 -1
View File
@@ -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)
}
+5 -4
View File
@@ -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))
+2 -2
View File
@@ -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
+4 -4
View File
@@ -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)
+8 -8
View File
@@ -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)
+4 -4
View File
@@ -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))
@@ -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 {
@@ -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()
+10 -9
View File
@@ -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/<group>, 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)
}
@@ -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)
}
@@ -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)
+3 -3
View File
@@ -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
+5 -4
View File
@@ -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
}
+10 -10
View File
@@ -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")
})
}
+9 -9
View File
@@ -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)
}
}
+9 -9
View File
@@ -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" {
@@ -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")
@@ -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))
}
@@ -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)
}
+6 -6
View File
@@ -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
@@ -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()
}
@@ -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")
}
@@ -1,4 +1,4 @@
package utils
package process
import (
"errors"
@@ -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")
}
@@ -1,4 +1,4 @@
package utils
package process
import (
"bufio"
+2 -2
View File
@@ -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)
+2 -1
View File
@@ -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 {
@@ -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"
+2 -2
View File
@@ -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)
@@ -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)
}
+2 -1
View File
@@ -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)
+2 -1
View File
@@ -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)
}
+2 -2
View File
@@ -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)
}
@@ -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/<pid> 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")
}
@@ -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, ""
+6 -5
View File
@@ -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)
@@ -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
@@ -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
}
@@ -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
+5 -3
View File
@@ -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:
-457
View File
@@ -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 {
+2 -1
View File
@@ -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)
}