diff --git a/build/systemd/spinifex-shutdown.service b/build/systemd/spinifex-shutdown.service new file mode 100644 index 000000000..f4a3f2197 --- /dev/null +++ b/build/systemd/spinifex-shutdown.service @@ -0,0 +1,18 @@ +[Unit] +Description=Spinifex graceful guest drain +PartOf=spinifex.target +# After= all services so it stops FIRST on target/host shutdown: the drain runs +# while nats/predastore/viperblock/daemon are still up (QMP, unmount, WAL flush). +After=spinifex-nats.service spinifex-predastore.service spinifex-viperblock.service spinifex-daemon.service + +[Service] +Type=oneshot +RemainAfterExit=yes +ExecStart=/bin/true +# ExecStop only runs when the target/host goes down, not on a single-service +# restart — so a daemon restart still leaves guests running (DDIL). +ExecStop=/usr/local/bin/spx admin node drain --local --timeout=120s +TimeoutStopSec=180 + +[Install] +WantedBy=spinifex.target diff --git a/build/systemd/spinifex.target b/build/systemd/spinifex.target index 594f39fa0..ba6d3d67c 100644 --- a/build/systemd/spinifex.target +++ b/build/systemd/spinifex.target @@ -2,7 +2,7 @@ Description=Spinifex Infrastructure Platform After=network-online.target Wants=network-online.target -Wants=spinifex-nats.service spinifex-predastore.service spinifex-viperblock.service spinifex-daemon.service spinifex-awsgw.service spinifex-vpcd.service spinifex-ui.service openvswitch-switch.service ovn-controller.service +Wants=spinifex-nats.service spinifex-predastore.service spinifex-viperblock.service spinifex-daemon.service spinifex-awsgw.service spinifex-vpcd.service spinifex-ui.service spinifex-shutdown.service openvswitch-switch.service ovn-controller.service [Install] WantedBy=multi-user.target diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index f85ab293c..b17b9b802 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -115,6 +115,23 @@ Each phase waits for all nodes to ACK before proceeding to the next.`, Run: runClusterShutdown, } +var nodeCmd = &cobra.Command{ + Use: "node", + Short: "Node-local operations", + Long: `Node-local administrative operations such as a graceful local guest drain.`, +} + +var nodeDrainCmd = &cobra.Command{ + Use: "drain", + Short: "Gracefully drain guests on the local node", + Long: `Run the GATE and DRAIN shutdown phases against the local node only: power down +its guests via QMP and unmount their volumes (flushing the viperblock WAL) while +every service is still running. STORAGE/PERSIST/INFRA are left to systemd's +ordered unit teardown. This is the ExecStop of spinifex-shutdown.service, so a +systemctl stop or host reboot drains guests before any storage service stops.`, + Run: runNodeDrainLocal, +} + var clusterDrainDHCPCmd = &cobra.Command{ Use: "drain-dhcp", Short: "Release all upstream DHCP leases held by vpcd", @@ -258,6 +275,11 @@ func init() { clusterCmd.AddCommand(clusterDrainDHCPCmd) clusterDrainDHCPCmd.Flags().Duration("timeout", 30*time.Second, "Reply-collection window for vpcd drain responders") + adminCmd.AddCommand(nodeCmd) + nodeCmd.AddCommand(nodeDrainCmd) + nodeDrainCmd.Flags().Bool("local", false, "Drain the local node only (required)") + nodeDrainCmd.Flags().Duration("timeout", 120*time.Second, "Maximum time to wait per phase") + adminCmd.AddCommand(imagesCmd) imagesCmd.AddCommand(imagesImportCmd) imagesCmd.AddCommand(imagesListCmd) diff --git a/cmd/spinifex/cmd/admin_test.go b/cmd/spinifex/cmd/admin_test.go index fb355e9a1..48f3ab25c 100644 --- a/cmd/spinifex/cmd/admin_test.go +++ b/cmd/spinifex/cmd/admin_test.go @@ -9,7 +9,6 @@ import ( "strings" "testing" - s3 "github.com/mulgadc/predastore/s3" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/formation" "github.com/mulgadc/spinifex/spinifex/utils" @@ -503,7 +502,7 @@ func TestPredastoreTemplates_SetCompactionEmitsParseableBlock(t *testing.T) { AccessKey: "AK", SecretKey: "SK", NatsToken: "tok", CompactionIntervalSeconds: interval, }) assert.Contains(t, single, "[compaction]") - var singleCfg s3.Config + var singleCfg compactionConfig require.NoError(t, toml.Unmarshal([]byte(single), &singleCfg)) assert.Equal(t, interval, singleCfg.Compaction.IntervalSeconds) @@ -511,11 +510,20 @@ func TestPredastoreTemplates_SetCompactionEmitsParseableBlock(t *testing.T) { predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", interval, ) require.NoError(t, err) - var multiCfg s3.Config + var multiCfg compactionConfig require.NoError(t, toml.Unmarshal([]byte(multi), &multiCfg)) assert.Equal(t, interval, multiCfg.Compaction.IntervalSeconds) } +// compactionConfig parses just the rendered [compaction] block. Predastore's +// s3.Config dropped its Compaction field, so this asserts the spinifex template +// still emits a valid, parseable block carrying the interval. +type compactionConfig struct { + Compaction struct { + IntervalSeconds int `toml:"interval_seconds"` + } `toml:"compaction"` +} + // The init/join flag must register with an unset (0) default and flow into the // rendered config via ConfigSettings. func TestAdminInitFlag_CompactionIntervalReachesRenderedConfig(t *testing.T) { diff --git a/cmd/spinifex/cmd/cluster.go b/cmd/spinifex/cmd/cluster.go index 5185343f2..53c60cffa 100644 --- a/cmd/spinifex/cmd/cluster.go +++ b/cmd/spinifex/cmd/cluster.go @@ -161,6 +161,93 @@ func runClusterShutdown(cmd *cobra.Command, args []string) { fmt.Printf("Cluster shutdown complete (%s)\n", time.Since(start).Round(time.Millisecond)) } +// runNodeDrainLocal runs the GATE and DRAIN shutdown phases against the local +// node only, leaving STORAGE/PERSIST/INFRA to systemd's ordered unit teardown. +// It is the ExecStop of spinifex-shutdown.service: guests are powered down and +// their volumes unmounted while every service is still up, so the subsequent +// service stops never tear storage out from under a running guest. +func runNodeDrainLocal(cmd *cobra.Command, args []string) { + local, _ := cmd.Flags().GetBool("local") + timeout, _ := cmd.Flags().GetDuration("timeout") + if !local { + fmt.Fprintln(os.Stderr, "Error: node drain currently supports only --local") + os.Exit(1) + } + + cfg, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + defer nc.Close() + + node := cfg.Node + fmt.Printf("Draining local node %q (gate -> drain), timeout %s/phase\n", node, timeout) + + for _, phase := range []string{"gate", "drain"} { + topic := "spinifex.cluster.shutdown." + phase + req := daemon.ShutdownRequest{Phase: phase, Timeout: int(timeout.Seconds())} + reqData, err := json.Marshal(req) + if err != nil { + fmt.Fprintf(os.Stderr, "Error marshaling request: %v\n", err) + os.Exit(1) + } + + ack, err := collectLocalShutdownACK(nc, topic, reqData, node, timeout) + if err != nil { + fmt.Fprintf(os.Stderr, "[%s] %v\n", strings.ToUpper(phase), err) + os.Exit(1) + } + if ack.Error != "" { + fmt.Fprintf(os.Stderr, "[%s] %s: ERROR - %s\n", strings.ToUpper(phase), ack.Node, ack.Error) + os.Exit(1) + } + if len(ack.Stopped) > 0 { + fmt.Printf("[%s] %s: stopped %s\n", strings.ToUpper(phase), ack.Node, strings.Join(ack.Stopped, ", ")) + } else { + fmt.Printf("[%s] %s: OK\n", strings.ToUpper(phase), ack.Node) + } + } + + fmt.Printf("Local node %q drained; systemd may now stop storage services\n", node) +} + +// collectLocalShutdownACK publishes a shutdown-phase request and returns the ACK +// from the local node, ignoring ACKs from any other node. Returns an error if +// the local node does not respond within timeout. +func collectLocalShutdownACK(nc *nats.Conn, topic string, reqData []byte, node string, timeout time.Duration) (daemon.ShutdownACK, error) { + inbox := nats.NewInbox() + sub, err := nc.SubscribeSync(inbox) + if err != nil { + return daemon.ShutdownACK{}, fmt.Errorf("failed to subscribe to inbox: %w", err) + } + defer sub.Unsubscribe() + + if err := nc.PublishRequest(topic, inbox, reqData); err != nil { + return daemon.ShutdownACK{}, fmt.Errorf("failed to publish request: %w", err) + } + nc.Flush() + + deadline := time.Now().Add(timeout) + for { + remaining := time.Until(deadline) + if remaining <= 0 { + return daemon.ShutdownACK{}, fmt.Errorf("timeout waiting for local node %q ACK", node) + } + msg, err := sub.NextMsg(remaining) + if err != nil { + return daemon.ShutdownACK{}, fmt.Errorf("timeout waiting for local node %q ACK", node) + } + var ack daemon.ShutdownACK + if err := json.Unmarshal(msg.Data, &ack); err != nil { + continue + } + if ack.Node == node { + return ack, nil + } + } +} + // runClusterDrainDHCP asks every vpcd to DHCPRELEASE all external-pool leases // it currently holds, returning them to the upstream DHCP server. Intended for // the teardown path: an env reset otherwise strands held leases until TTL diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index acff9532b..3a07b7ab6 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -66,6 +66,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t |---------|-------|-------------| | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. | | `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. | +| `spx admin node drain --local` | `--local` (drain the local node only, required), `--timeout` (max wait per phase, default 120s) | Runs the GATE and DRAIN phases against the local node only: powers down its guests via QMP and unmounts their volumes (flushing the viperblock WAL) while every service is still up. STORAGE/PERSIST/INFRA are left to systemd's ordered teardown. Wired as the `ExecStop` of `spinifex-shutdown.service`, so `systemctl stop spinifex.target` and host reboot/poweroff drain guests before any storage service stops. | ### Certificate Management diff --git a/spinifex/services/viperblockd/shutdown_test.go b/spinifex/services/viperblockd/shutdown_test.go new file mode 100644 index 000000000..508260008 --- /dev/null +++ b/spinifex/services/viperblockd/shutdown_test.go @@ -0,0 +1,107 @@ +package viperblockd + +import ( + "net" + "os/exec" + "path/filepath" + "syscall" + "testing" +) + +// alive reports whether pid is still a live process. +func alive(pid int) bool { + return syscall.Kill(pid, 0) == nil +} + +// TestShutdownVolumes_InUseNotKilled asserts the SIGTERM handler leaves an +// nbdkit serving an attached guest running — killing it would corrupt the guest. +func TestShutdownVolumes_InUseNotKilled(t *testing.T) { + cmd := exec.Command("sleep", "30") + if err := cmd.Start(); err != nil { + t.Fatalf("start sleep: %v", err) + } + pid := cmd.Process.Pid + defer func() { _ = cmd.Process.Kill() }() + + shutdownVolumes([]MountedVolume{{Name: "vol-inuse", PID: pid}}, func(MountedVolume) bool { return true }) + + if !alive(pid) { + t.Fatal("in-use nbdkit was killed on SIGTERM — guest corruption risk") + } +} + +// TestShutdownVolumes_IdleKilled asserts an nbdkit with no attached guest is +// reaped on SIGTERM, so a clean stop leaves no orphan. +func TestShutdownVolumes_IdleKilled(t *testing.T) { + cmd := exec.Command("sleep", "30") + if err := cmd.Start(); err != nil { + t.Fatalf("start sleep: %v", err) + } + pid := cmd.Process.Pid + defer func() { _ = cmd.Process.Kill() }() + + // Reap concurrently: KillProcess SIGTERMs then polls Signal(0); without a + // Wait the exited child lingers as a zombie and the poll runs full timeout. + go func() { _, _ = cmd.Process.Wait() }() + + shutdownVolumes([]MountedVolume{{Name: "vol-idle", PID: pid}}, func(MountedVolume) bool { return false }) + + if alive(pid) { + t.Fatal("idle nbdkit was not reaped on SIGTERM") + } +} + +// TestNbdkitInUse_ConnectedClientDetected proves nbdkitInUse parses ss output +// and detects a live client on the unix socket — a guest still attached. The +// state column is field[1] (field[0] is the netid), so a wrong index here +// would silently report "idle" and let the SIGTERM path kill an in-use nbdkit. +func TestNbdkitInUse_ConnectedClientDetected(t *testing.T) { + if _, err := exec.LookPath("ss"); err != nil { + t.Skip("ss not available") + } + sock := filepath.Join(t.TempDir(), "nbd.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + accepted := make(chan net.Conn, 1) + go func() { + c, err := ln.Accept() + if err == nil { + accepted <- c + } + }() + client, err := net.Dial("unix", sock) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer client.Close() + srvConn := <-accepted + defer srvConn.Close() + + if !nbdkitInUse(MountedVolume{Name: "vol-attached", Socket: sock}) { + t.Fatal("nbdkitInUse returned false for a socket with a connected client") + } +} + +// TestNbdkitInUse_NoSocketNotInUse: a path with no socket has no ESTAB row, so +// an idle nbdkit is reapable. +func TestNbdkitInUse_NoSocketNotInUse(t *testing.T) { + if _, err := exec.LookPath("ss"); err != nil { + t.Skip("ss not available") + } + sock := filepath.Join(t.TempDir(), "absent.sock") + if nbdkitInUse(MountedVolume{Name: "vol-idle", Socket: sock}) { + t.Fatal("nbdkitInUse returned true for a socket with no connections") + } +} + +// TestNbdkitInUse_TCPAssumedInUse: TCP transport can't be cheaply confirmed +// idle, so it defaults to in-use (safe — never kill under a guest). +func TestNbdkitInUse_TCPAssumedInUse(t *testing.T) { + if !nbdkitInUse(MountedVolume{Name: "vol-tcp", Port: 10809}) { + t.Fatal("nbdkitInUse must assume TCP transport is in use") + } +} diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 53cbd43d1..f068fb9ff 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "os" + "os/exec" "os/signal" "path/filepath" "strconv" @@ -945,15 +946,54 @@ func launchService(cfg *Config) (err error) { cfg.MountedVolumes = nil cfg.mu.Unlock() + shutdownVolumes(volumes, nbdkitInUse) + + return nil +} + +// shutdownVolumes flushes each mounted volume's WAL on SIGTERM but only reaps +// nbdkit for volumes with no attached guest (inUse false). Killing an nbdkit a +// guest is still writing through corrupts that guest's filesystem; the graceful +// drain (or unmount) path owns reaping in-use nbdkit after the guest is gone. +func shutdownVolumes(volumes []MountedVolume, inUse func(MountedVolume) bool) { for _, volume := range volumes { if volume.VB != nil { volume.VB.StopWALSyncer() } - slog.Info("Killing nbdkit process", "pid", volume.PID) + if inUse(volume) { + slog.Warn("nbdkit still serving a guest; leaving it for the drain/unmount path", + "pid", volume.PID, "name", volume.Name, "socket", volume.Socket) + continue + } + slog.Info("Killing idle nbdkit process", "pid", volume.PID, "name", volume.Name) if err := utils.KillProcess(volume.PID); err != nil { slog.Error("Failed to kill nbdkit process", "pid", volume.PID, "err", err) } } - - return nil +} + +// nbdkitInUse best-effort reports whether nbdkit's NBD endpoint still has a +// connected client (a guest). On any uncertainty it returns true so the +// shutdown path never tears a backing store out from under a running guest. +func nbdkitInUse(vol MountedVolume) bool { + if vol.Socket == "" { + // TCP transport: cannot cheaply confirm idle — assume in use. + return true + } + out, err := exec.Command("ss", "-H", "-x", "-a").Output() + if err != nil { + return true + } + // ss -H rows are: ... + // LISTEN is the idle server socket; ESTAB means a client is attached. + for line := range strings.SplitSeq(string(out), "\n") { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + if fields[1] == "ESTAB" && strings.Contains(line, vol.Socket) { + return true + } + } + return false } diff --git a/spinifex/systemd/units_invariants_test.go b/spinifex/systemd/units_invariants_test.go index 90c4c555f..0163d3c3d 100644 --- a/spinifex/systemd/units_invariants_test.go +++ b/spinifex/systemd/units_invariants_test.go @@ -136,6 +136,49 @@ func TestRG9_TierConfinement(t *testing.T) { } } +// TestGracefulDrainOrdering asserts the graceful-shutdown contract: the drain +// oneshot orders After= the storage/daemon units (so a target/host stop runs its +// ExecStop drain first, while those services are still up), the daemon keeps +// KillMode=process (guests survive a daemon restart — DDIL reattach), and the +// drain is wired into the target so a target/host stop triggers it. +func TestGracefulDrainOrdering(t *testing.T) { + dir := unitsDir(t) + + drain := readUnit(t, dir, "spinifex-shutdown.service") + var afterLine string + for l := range strings.SplitSeq(drain, "\n") { + if strings.HasPrefix(strings.TrimSpace(l), "After=") { + afterLine = strings.TrimSpace(l) + } + } + if afterLine == "" { + t.Fatal("spinifex-shutdown.service must declare After= the storage/daemon units") + } + for _, dep := range []string{ + "spinifex-nats.service", + "spinifex-predastore.service", + "spinifex-viperblock.service", + "spinifex-daemon.service", + } { + if !strings.Contains(afterLine, dep) { + t.Errorf("spinifex-shutdown.service After= must include %s so the drain stops before it", dep) + } + } + if !hasDirective(drain, "ExecStop=/usr/local/bin/spx admin node drain --local --timeout=120s") { + t.Error("spinifex-shutdown.service must drain the local node on stop via ExecStop") + } + + daemon := readUnit(t, dir, "spinifex-daemon.service") + if !hasDirective(daemon, "KillMode=process") { + t.Error("spinifex-daemon.service must keep KillMode=process — guests survive daemon restart (DDIL)") + } + + target := readUnit(t, dir, "spinifex.target") + if !strings.Contains(target, "spinifex-shutdown.service") { + t.Error("spinifex.target Wants= must include spinifex-shutdown.service") + } +} + // TestRG11_LeanUnits asserts the RG-11 contract: unit/slice files carry settings // plus terse # RG-n references, not paragraphs of rationale, and never reference // a plan doc, bead, or CI run (project policy — reasoning lives in the ADR).