From 344a490a70b1e285ea789a4594a9cd5e0d4b3fda Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 22 Jul 2026 20:38:44 +1000 Subject: [PATCH] fix(vm): give relaunched data volumes named block nodes so detach can complete (#575) buildDrives emitted every non-boot data volume as a bare -drive with no id= and no node-name=, so a stop/start (or reboot/resize) relaunch left the volume anonymous: device_del/blockdev-del had nothing to address, QEMU kept the NBD socket open, and DetachVolume stalled through nbdkit's full kill grace before giving up. Add -blockdev support to the QEMU config builder and use it for non-boot, non-EFI volumes, sharing the node-name/device-id/iothread-id naming and virtio-blk-pci device arguments with AttachVolume's hot-plug path so cold-boot and hot-attach can no longer drift apart. A data volume with no persisted HotplugPort now gets one allocated and written back during buildDrives instead of landing on an unplugged bus. Also raise unmountSealTimeout past viperblockd's kill grace (it runs the seal after the SIGKILL, so matching the grace meant an unmount that used the whole grace could never be observed as successful), and bound the e2e harness's forced teardown detach by the teardown deadline so a slow detach is reported as slow rather than as an unexplained leak. --- spinifex/daemon/vm_adapters.go | 9 +- spinifex/vm/devices.go | 114 +++++++++++++++++++++++ spinifex/vm/lifecycle.go | 139 +++++++++++++++++++++-------- spinifex/vm/lifecycle_test.go | 103 +++++++++++++++++++-- spinifex/vm/vm.go | 20 +++++ spinifex/vm/vm_test.go | 85 ++++++++++++++++++ spinifex/vm/volumes.go | 45 ++++------ tests/e2e/harness/ec2helpers.go | 16 +++- tests/e2e/harness/fixtures_test.go | 7 ++ 9 files changed, 464 insertions(+), 74 deletions(-) diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index cf28e6a72..13e9638bb 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -91,9 +91,12 @@ type volumeMounterAdapter struct { var _ vm.VolumeMounter = (*volumeMounterAdapter)(nil) // unmountSealTimeout bounds the ebs.unmount NATS request. The handler now drives -// a synchronous block-map seal to predastore (S3 I/O), so it matches -// viperblockd's 120s KillProcess wait rather than a bare RPC budget. -const unmountSealTimeout = 120 * time.Second +// a synchronous block-map seal to predastore (S3 I/O), which runs AFTER +// viperblockd's SIGKILL once its 120s KillProcess grace elapses — so this +// must strictly exceed that grace plus a seal allowance, never merely match +// it: an unmount that consumes the whole grace could otherwise never be +// observed as successful. +const unmountSealTimeout = 180 * time.Second func newVolumeMounterAdapter(nc *nats.Conn, node string, volState vm.VolumeStateUpdater) *volumeMounterAdapter { return &volumeMounterAdapter{nc: nc, node: node, volState: volState} diff --git a/spinifex/vm/devices.go b/spinifex/vm/devices.go index f0fe5a473..80360189e 100644 --- a/spinifex/vm/devices.go +++ b/spinifex/vm/devices.go @@ -2,6 +2,7 @@ package vm import ( "fmt" + "strconv" "strings" ) @@ -58,3 +59,116 @@ func RngDevice(machineType string) Device { } return Device{Value: "virtio-rng-pci"} } + +// The volume ID/bus/serial helpers below are shared by the hot-attach path +// (AttachVolume/DetachVolume, over QMP) and the cold-boot path (buildDrives, +// on the QEMU command line) so a data volume's block-graph names cannot +// drift between the two: a name minted here for a hot-plugged volume must +// still resolve after a stop/start relaunches it with the same volume ID. + +// VolumeNodeName returns the block-node name (blockdev-add/-blockdev +// node-name, and the -device drive= reference) for a data volume. +func VolumeNodeName(volumeID string) string { + return fmt.Sprintf("nbd-%s", volumeID) +} + +// VolumeDeviceID returns the virtio-blk-pci device id (device_add/-device id=) +// for a data volume. DetachVolume's device_del addresses this exact id, so a +// cold-booted volume must be given the same one an AttachVolume would use. +func VolumeDeviceID(volumeID string) string { + return fmt.Sprintf("vdisk-%s", volumeID) +} + +// VolumeIOThreadID returns the iothread object id (object-add/-object id=) +// backing a data volume's virtio-blk-pci device. +func VolumeIOThreadID(volumeID string) string { + return fmt.Sprintf("ioth-%s", volumeID) +} + +// VolumeSerial returns the virtio-blk serial QEMU exposes in-guest: the +// volume ID with dashes stripped ("vol" + 17 hex = 20 bytes, the virtio-blk +// serial limit). The EBS CSI node plugin matches this via +// /dev/disk/by-id/** to locate the device. +func VolumeSerial(volumeID string) string { + return strings.ReplaceAll(volumeID, "-", "") +} + +// HotplugEBSBus returns the PCIe hot-plug root-port bus name for hot-plug +// port. A device on pcie.0 cannot be unplugged, so every data volume — +// hot-attached or cold-booted — must land on one of these ports. +func HotplugEBSBus(port int) string { + return fmt.Sprintf("hotplug-ebs%d", port) +} + +// VolumeBlkDeviceArgs returns the virtio-blk-pci device's "key=value" options +// (everything after the leading driver name) as an ordered slice, shared by +// AttachVolume's QMP device_add (rendered as a map) and buildDrives' -device +// (rendered as a joined command-line string) so the two block-graph shapes +// cannot diverge. +func VolumeBlkDeviceArgs(volumeID, nodeName, iothreadID, bus string) []string { + return []string{ + fmt.Sprintf("id=%s", VolumeDeviceID(volumeID)), + fmt.Sprintf("drive=%s", nodeName), + fmt.Sprintf("iothread=%s", iothreadID), + fmt.Sprintf("serial=%s", VolumeSerial(volumeID)), + fmt.Sprintf("bus=%s", bus), + } +} + +// VolumeBlkDevice renders a QEMU -device argument for buildDrives' cold-boot +// path: the virtio-blk-pci driver name followed by VolumeBlkDeviceArgs. +func VolumeBlkDevice(volumeID, nodeName, iothreadID, bus string) Device { + args := append([]string{"virtio-blk-pci"}, VolumeBlkDeviceArgs(volumeID, nodeName, iothreadID, bus)...) + return Device{Value: strings.Join(args, ",")} +} + +// VolumeBlkDeviceQMPArgs renders the same virtio-blk-pci arguments as a +// device_add argument map for AttachVolume's QMP hot-attach path. +func VolumeBlkDeviceQMPArgs(volumeID, nodeName, iothreadID, bus string) map[string]any { + return map[string]any{ + "driver": "virtio-blk-pci", + "id": VolumeDeviceID(volumeID), + "drive": nodeName, + "iothread": iothreadID, + "serial": VolumeSerial(volumeID), + "bus": bus, + } +} + +// NBDServerOpts holds the server.* fields QEMU's nbd blockdev driver needs, +// broken out of a parsed NBD URI (see utils.ParseNBDURI) so both the QMP +// blockdev-add nested "server" object and the command-line -blockdev +// "server.*" options can be built from the same parse. +type NBDServerOpts struct { + Type string // "unix" or "inet" + Path string // set when Type == "unix" + Host string // set when Type == "inet" + Port int // set when Type == "inet" +} + +// QMPArg renders the server options as the nested map blockdev-add expects. +func (o NBDServerOpts) QMPArg() map[string]any { + if o.Type == "unix" { + return map[string]any{"type": "unix", "path": o.Path} + } + return map[string]any{"type": "inet", "host": o.Host, "port": strconv.Itoa(o.Port)} +} + +// CommandLineArgs renders the server options as "server.*" command-line +// key=value pairs for -blockdev. +func (o NBDServerOpts) CommandLineArgs() []string { + if o.Type == "unix" { + return []string{"server.type=unix", fmt.Sprintf("server.path=%s", o.Path)} + } + return []string{"server.type=inet", fmt.Sprintf("server.host=%s", o.Host), fmt.Sprintf("server.port=%d", o.Port)} +} + +// VolumeBlockdev renders a -blockdev argument for a data volume's NBD block +// node: the command-line equivalent of blockdev-add, producing a +// monitor-owned node that blockdev-del can later remove (unlike -drive, +// which only creates a BlockBackend with an auto-generated node-name). +func VolumeBlockdev(nodeName string, server NBDServerOpts) Blockdev { + opts := append([]string{"driver=nbd", fmt.Sprintf("node-name=%s", nodeName)}, server.CommandLineArgs()...) + opts = append(opts, "export=") + return Blockdev{Value: strings.Join(opts, ",")} +} diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 3883db517..5999c62ab 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -290,14 +290,15 @@ func (m *Manager) startQEMU(instance *VM) error { m.initENIRequests(instance) instance.EBSRequests.Mu.Lock() - drives, iothreads, devices, err := buildDrives(instance.EBSRequests.Requests, spec.VCPUs, instance.Config.MachineType) + driveCfg, err := buildDrives(instance.EBSRequests.Requests, spec.VCPUs, instance.Config.MachineType) instance.EBSRequests.Mu.Unlock() if err != nil { return err } - instance.Config.Drives = append(instance.Config.Drives, drives...) - instance.Config.IOThreads = append(instance.Config.IOThreads, iothreads...) - instance.Config.Devices = append(instance.Config.Devices, devices...) + instance.Config.Drives = append(instance.Config.Drives, driveCfg.Drives...) + instance.Config.IOThreads = append(instance.Config.IOThreads, driveCfg.IOThreads...) + instance.Config.Devices = append(instance.Config.Devices, driveCfg.Devices...) + instance.Config.Blockdevs = append(instance.Config.Blockdevs, driveCfg.Blockdevs...) } if instance.DirectBoot { @@ -1102,49 +1103,115 @@ func (m *Manager) initENIRequests(instance *VM) { } } -// buildDrives converts EBS requests into QEMU drive/iothread/device configs. -// Returns an error if any volume is missing its NBDURI. EFI volumes emit -// pflash unit=1; the readonly CODE blob (unit=0) is added by Config.Execute. -func buildDrives(requests []types.EBSRequest, cpuCount int, machineType string) ([]Drive, []IOThread, []Device, error) { - var drives []Drive - var iothreads []IOThread - var devices []Device +// driveConfig is buildDrives' output: the QEMU drive/iothread/device/blockdev +// entries derived from one instance's EBS requests. +type driveConfig struct { + Drives []Drive + IOThreads []IOThread + Devices []Device + Blockdevs []Blockdev +} - for _, v := range requests { +// buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev +// configs. Returns an error if any volume is missing its NBDURI, or if the +// EBS hot-plug port pool is exhausted while allocating one for a data volume. +// +// Boot volumes keep the existing anonymous if=none -drive shape exactly as +// before — they are not detachable (see ErrVolumeNotDetachable), so the +// anonymous-node problem this function otherwise fixes does not apply, and +// changing the boot drive risks boot order. EFI volumes emit pflash unit=1; +// the readonly CODE blob (unit=0) is added by Config.Execute. +// +// Every other (data) volume is given the same named node-name/device-id/ +// iothread-id shape AttachVolume's hot-plug path uses, via a -blockdev entry +// and a matching -device, instead of the old bare -drive with no id and no +// node-name. Without this a relaunched (stop/start, reboot, resize) data +// volume was undetachable: device_del/blockdev-del had nothing to address, +// so DetachVolume could never complete and the NBD socket stayed open until +// the unmount burned its full kill grace. +// +// requests is mutated in place: a data volume with no persisted HotplugPort +// (state written before hot-plug port accounting existed, or a volume +// attached at RunInstances time) gets one allocated and written back, so a +// later relaunch reuses the same port instead of reallocating it. +func buildDrives(requests []types.EBSRequest, cpuCount int, machineType string) (driveConfig, error) { + var cfg driveConfig + + for i := range requests { + v := &requests[i] if v.NBDURI == "" { - return nil, nil, nil, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name) + return driveConfig{}, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name) } - drive := Drive{File: v.NBDURI} + switch { + case v.Boot: + drive := Drive{ + File: v.NBDURI, + Format: "raw", + If: "none", + Media: "disk", + ID: "os", + Cache: "none", - if v.Boot { - drive.Format = "raw" - drive.If = "none" - drive.Media = "disk" - drive.ID = "os" - drive.Cache = "none" - - // Report backend write errors to the guest rather than letting - // QEMU's default werror=enospc pause the VM: when the storage pool - // exhausts, the guest fs then returns a clean ENOSPC to userspace - // and the VM stays reachable instead of freezing. - drive.Werror = "report" - drive.Rerror = "report" + // Report backend write errors to the guest rather than letting + // QEMU's default werror=enospc pause the VM: when the storage + // pool exhausts, the guest fs then returns a clean ENOSPC to + // userspace and the VM stays reachable instead of freezing. + Werror: "report", + Rerror: "report", + } iothreadID := "ioth-os" - iothreads = append(iothreads, IOThread{ID: iothreadID}) - devices = append(devices, BlkDevice(machineType, drive.ID, iothreadID, cpuCount, 1)) - } + cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID}) + cfg.Devices = append(cfg.Devices, BlkDevice(machineType, drive.ID, iothreadID, cpuCount, 1)) + cfg.Drives = append(cfg.Drives, drive) - if v.EFI { - drive.Format = "raw" - drive.If = "pflash" - drive.Unit = 1 + case v.EFI: + cfg.Drives = append(cfg.Drives, Drive{ + File: v.NBDURI, + Format: "raw", + If: "pflash", + Unit: 1, + }) + + default: + // A data (non-boot, non-EFI) volume. Allocate a stable hot-plug + // port when the persisted state predates port accounting, or the + // volume was attached at RunInstances time without one; + // freeHotplugEBSPort re-scans requests as mutated so far in this + // loop, so two unset data volumes in one call cannot collide. + if v.HotplugPort == 0 { + port := freeHotplugEBSPort(requests) + if port == 0 { + return driveConfig{}, fmt.Errorf("no free EBS hot-plug port for volume %s", v.Name) + } + v.HotplugPort = port + } + + serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(v.NBDURI) + if err != nil { + return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err) + } + + nodeName := VolumeNodeName(v.Name) + iothreadID := VolumeIOThreadID(v.Name) + bus := HotplugEBSBus(v.HotplugPort) + + cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID}) + cfg.Blockdevs = append(cfg.Blockdevs, VolumeBlockdev(nodeName, NBDServerOpts{ + Type: serverType, + Path: socketPath, + Host: nbdHost, + Port: nbdPort, + })) + cfg.Devices = append(cfg.Devices, VolumeBlkDevice(v.Name, nodeName, iothreadID, bus)) + // Deliberately no entry in cfg.Drives: the bare anonymous -drive + // this case used to emit (no id=, no node-name=) is exactly what + // left the volume undetachable. } slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI) - drives = append(drives, drive) } - return drives, iothreads, devices, nil + return cfg, nil } diff --git a/spinifex/vm/lifecycle_test.go b/spinifex/vm/lifecycle_test.go index 41c5b03b4..9d9d8a27f 100644 --- a/spinifex/vm/lifecycle_test.go +++ b/spinifex/vm/lifecycle_test.go @@ -113,7 +113,12 @@ func TestBuildDrives(t *testing.T) { wantDrives []Drive wantIOThreads []IOThread wantDevices []Device - wantErr string + wantBlockdevs []Blockdev + // wantHotplugPorts, when non-nil, asserts requests[i].HotplugPort + // after the call — pinning both the "already recorded, left alone" + // and "unset, allocated and written back" behaviours. + wantHotplugPorts []int + wantErr string }{ { name: "boot volume", @@ -156,20 +161,95 @@ func TestBuildDrives(t *testing.T) { wantErr: "NBDURI not set for volume vol-efi-bad", }, { - name: "mixed boot + EFI", + name: "data volume with a recorded HotplugPort keeps it and emits a named blockdev+device", + requests: []types.EBSRequest{ + {Name: "vol-data-a", NBDURI: "nbd:unix:/tmp/data-a.sock", HotplugPort: 3}, + }, + cpuCount: 2, + wantIOThreads: []IOThread{{ID: "ioth-vol-data-a"}}, + wantBlockdevs: []Blockdev{ + {Value: "driver=nbd,node-name=nbd-vol-data-a,server.type=unix,server.path=/tmp/data-a.sock,export="}, + }, + wantDevices: []Device{ + {Value: "virtio-blk-pci,id=vdisk-vol-data-a,drive=nbd-vol-data-a,iothread=ioth-vol-data-a,serial=voldataa,bus=hotplug-ebs3"}, + }, + wantHotplugPorts: []int{3}, + }, + { + name: "data volume with HotplugPort unset gets the lowest free port allocated and written back", + requests: []types.EBSRequest{ + {Name: "vol-data-b", NBDURI: "nbd:unix:/tmp/data-b.sock"}, + }, + cpuCount: 2, + wantIOThreads: []IOThread{{ID: "ioth-vol-data-b"}}, + wantBlockdevs: []Blockdev{ + {Value: "driver=nbd,node-name=nbd-vol-data-b,server.type=unix,server.path=/tmp/data-b.sock,export="}, + }, + wantDevices: []Device{ + {Value: "virtio-blk-pci,id=vdisk-vol-data-b,drive=nbd-vol-data-b,iothread=ioth-vol-data-b,serial=voldatab,bus=hotplug-ebs1"}, + }, + wantHotplugPorts: []int{1}, + }, + { + name: "two data volumes with HotplugPort unset in one call get distinct ports", + requests: []types.EBSRequest{ + {Name: "vol-data-c", NBDURI: "nbd:unix:/tmp/data-c.sock"}, + {Name: "vol-data-d", NBDURI: "nbd:unix:/tmp/data-d.sock"}, + }, + cpuCount: 2, + wantIOThreads: []IOThread{ + {ID: "ioth-vol-data-c"}, + {ID: "ioth-vol-data-d"}, + }, + wantBlockdevs: []Blockdev{ + {Value: "driver=nbd,node-name=nbd-vol-data-c,server.type=unix,server.path=/tmp/data-c.sock,export="}, + {Value: "driver=nbd,node-name=nbd-vol-data-d,server.type=unix,server.path=/tmp/data-d.sock,export="}, + }, + wantDevices: []Device{ + {Value: "virtio-blk-pci,id=vdisk-vol-data-c,drive=nbd-vol-data-c,iothread=ioth-vol-data-c,serial=voldatac,bus=hotplug-ebs1"}, + {Value: "virtio-blk-pci,id=vdisk-vol-data-d,drive=nbd-vol-data-d,iothread=ioth-vol-data-d,serial=voldatad,bus=hotplug-ebs2"}, + }, + wantHotplugPorts: []int{1, 2}, + }, + { + name: "data volume errors when the hot-plug port pool is exhausted", + requests: func() []types.EBSRequest { + reqs := make([]types.EBSRequest, 0, EBSHotPlugSlotCount+1) + for i := 1; i <= EBSHotPlugSlotCount; i++ { + reqs = append(reqs, types.EBSRequest{ + Name: fmt.Sprintf("vol-filler-%d", i), NBDURI: "nbd:unix:/tmp/filler.sock", HotplugPort: i, + }) + } + reqs = append(reqs, types.EBSRequest{Name: "vol-overflow", NBDURI: "nbd:unix:/tmp/overflow.sock"}) + return reqs + }(), + cpuCount: 2, + wantErr: "no free EBS hot-plug port for volume vol-overflow", + }, + { + name: "mixed boot + EFI + data: boot and EFI stay byte-identical to the legacy shape", requests: []types.EBSRequest{ {Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true}, {Name: "vol-efi", NBDURI: "nbd:unix:/tmp/efi.sock", EFI: true}, + {Name: "vol-data-a", NBDURI: "nbd:unix:/tmp/data-a.sock", HotplugPort: 3}, }, cpuCount: 4, wantDrives: []Drive{ {File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "report", Rerror: "report"}, {File: "nbd:unix:/tmp/efi.sock", Format: "raw", If: "pflash", Unit: 1}, }, - wantIOThreads: []IOThread{{ID: "ioth-os"}}, + wantIOThreads: []IOThread{ + {ID: "ioth-os"}, + {ID: "ioth-vol-data-a"}, + }, + wantBlockdevs: []Blockdev{ + {Value: "driver=nbd,node-name=nbd-vol-data-a,server.type=unix,server.path=/tmp/data-a.sock,export="}, + }, wantDevices: []Device{ {Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1"}, + {Value: "virtio-blk-pci,id=vdisk-vol-data-a,drive=nbd-vol-data-a,iothread=ioth-vol-data-a,serial=voldataa,bus=hotplug-ebs3"}, }, + wantHotplugPorts: []int{0, 0, 3}, }, { name: "empty requests", @@ -184,7 +264,7 @@ func TestBuildDrives(t *testing.T) { if machineType == "" { machineType = "q35" } - drives, iothreads, devices, err := buildDrives(tt.requests, tt.cpuCount, machineType) + cfg, err := buildDrives(tt.requests, tt.cpuCount, machineType) if tt.wantErr != "" { require.Error(t, err) @@ -193,9 +273,18 @@ func TestBuildDrives(t *testing.T) { } require.NoError(t, err) - assert.Equal(t, tt.wantDrives, drives) - assert.Equal(t, tt.wantIOThreads, iothreads) - assert.Equal(t, tt.wantDevices, devices) + assert.Equal(t, tt.wantDrives, cfg.Drives) + assert.Equal(t, tt.wantIOThreads, cfg.IOThreads) + assert.Equal(t, tt.wantDevices, cfg.Devices) + assert.Equal(t, tt.wantBlockdevs, cfg.Blockdevs) + + if tt.wantHotplugPorts != nil { + require.Len(t, tt.requests, len(tt.wantHotplugPorts)) + for i, want := range tt.wantHotplugPorts { + assert.Equal(t, want, tt.requests[i].HotplugPort, + "request %d (%s) HotplugPort", i, tt.requests[i].Name) + } + } }) } } diff --git a/spinifex/vm/vm.go b/spinifex/vm/vm.go index 705e7f4b7..e4bfe30a3 100644 --- a/spinifex/vm/vm.go +++ b/spinifex/vm/vm.go @@ -224,6 +224,16 @@ type IOThread struct { ID string `json:"id"` } +// Blockdev is a QEMU -blockdev entry: the command-line equivalent of the +// blockdev-add QMP command. Unlike a -drive with if=none, it produces a +// monitor-owned node addressable by node-name, which is what makes +// blockdev-del able to remove it later. Used for data (non-boot, non-EFI) +// volumes so a cold-booted one has the same detachable block-graph shape as +// a hot-attached one; boot/EFI volumes keep the legacy -drive shape. +type Blockdev struct { + Value string `json:"value"` +} + type Config struct { Name string `json:"name"` PIDFile string `json:"pid_file"` @@ -243,6 +253,9 @@ type Config struct { Drives []Drive `json:"drives"` IOThreads []IOThread `json:"io_threads,omitempty"` + // Blockdevs are rendered as -blockdev, before Devices, so a -device's + // drive= reference resolves against an already-declared node. + Blockdevs []Blockdev `json:"blockdevs,omitempty"` Devices []Device `json:"devices"` NetDevs []NetDev `json:"net_devs"` @@ -417,6 +430,13 @@ func (cfg *Config) Execute() (*exec.Cmd, error) { args = append(args, "-drive", strings.Join(opts, ",")) } + // -blockdev entries must precede the -device entries that reference them + // via drive= so QEMU resolves the node on first pass (QEMU itself + // tolerates either order; this just matches how the file already reads). + for _, blockdev := range cfg.Blockdevs { + args = append(args, "-blockdev", blockdev.Value) + } + for _, device := range cfg.Devices { args = append(args, "-device", device.Value) } diff --git a/spinifex/vm/vm_test.go b/spinifex/vm/vm_test.go index cbc5076ef..559af39cc 100644 --- a/spinifex/vm/vm_test.go +++ b/spinifex/vm/vm_test.go @@ -220,6 +220,91 @@ func TestExecute_MultipleIOThreads(t *testing.T) { assert.Equal(t, 2, iothreadCount) } +// TestExecute_Blockdev_UnixServer asserts a unix-socket NBD blockdev renders +// as a -blockdev argument, ordered before its referencing -device, and that +// -drive is untouched when Drives is empty — a data volume no longer gets an +// anonymous -drive. +func TestExecute_Blockdev_UnixServer(t *testing.T) { + cfg := Config{ + CPUCount: 2, + Memory: 1024, + Architecture: "x86_64", + // Execute requires a drive or kernel image; a kernel keeps this test + // isolated to blockdev/device rendering without an unrelated -drive. + KernelImage: "/boot/vmlinuz", + IOThreads: []IOThread{{ID: "ioth-vol-data-a"}}, + Blockdevs: []Blockdev{ + VolumeBlockdev("nbd-vol-data-a", NBDServerOpts{Type: "unix", Path: "/run/spinifex/nbd/nbd-vol-data-a.sock"}), + }, + Devices: []Device{ + VolumeBlkDevice("vol-data-a", "nbd-vol-data-a", "ioth-vol-data-a", "hotplug-ebs3"), + }, + } + + cmd, err := cfg.Execute() + require.NoError(t, err) + + args := cmd.Args[1:] + assert.False(t, argExists(args, "-drive"), "a config with only a data volume must emit no -drive") + + blockdevIdx := -1 + for i, a := range args { + if a == "-blockdev" { + blockdevIdx = i + break + } + } + require.Greater(t, blockdevIdx, -1, "-blockdev must be emitted") + assert.Equal(t, "driver=nbd,node-name=nbd-vol-data-a,server.type=unix,server.path=/run/spinifex/nbd/nbd-vol-data-a.sock,export=", args[blockdevIdx+1]) + + deviceIdx := -1 + for i, a := range args { + if a == "-device" { + deviceIdx = i + break + } + } + require.Greater(t, deviceIdx, -1) + assert.Greater(t, deviceIdx, blockdevIdx, "-blockdev must precede the -device referencing it") + assert.Equal(t, "virtio-blk-pci,id=vdisk-vol-data-a,drive=nbd-vol-data-a,iothread=ioth-vol-data-a,serial=voldataa,bus=hotplug-ebs3", args[deviceIdx+1]) +} + +// TestExecute_Blockdev_InetServer asserts the TCP NBD form renders +// server.type=inet with host/port instead of a unix socket path. +func TestExecute_Blockdev_InetServer(t *testing.T) { + cfg := Config{ + CPUCount: 2, + Memory: 1024, + Architecture: "x86_64", + KernelImage: "/boot/vmlinuz", + Blockdevs: []Blockdev{ + VolumeBlockdev("nbd-vol-data-b", NBDServerOpts{Type: "inet", Host: "10.0.0.5", Port: 10809}), + }, + } + + cmd, err := cfg.Execute() + require.NoError(t, err) + + assert.Equal(t, "driver=nbd,node-name=nbd-vol-data-b,server.type=inet,server.host=10.0.0.5,server.port=10809,export=", + argValue(cmd.Args[1:], "-blockdev")) +} + +// TestExecute_NoBlockdevWhenEmpty asserts a config with no Blockdevs emits no +// -blockdev flag at all — the common case (boot-only, or no EBS requests). +func TestExecute_NoBlockdevWhenEmpty(t *testing.T) { + cfg := Config{ + CPUCount: 1, + Memory: 512, + Architecture: "x86_64", + Drives: []Drive{{File: "disk.img", Format: "raw"}}, + } + + cmd, err := cfg.Execute() + require.NoError(t, err) + + assert.False(t, argExists(cmd.Args[1:], "-blockdev")) +} + // argValue returns the value following flag in args, or "" if not found. func argValue(args []string, flag string) string { for i, a := range args { diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index 713aa6378..a233f9844 100644 --- a/spinifex/vm/volumes.go +++ b/spinifex/vm/volumes.go @@ -7,7 +7,6 @@ import ( "fmt" "log/slog" "net" - "strconv" "strings" "time" @@ -100,17 +99,13 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) m.rollbackUnmount(ebsRequest) return "", fmt.Errorf("parse NBDURI: %w", err) } + serverArg := NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg() - var serverArg map[string]any - if serverType == "unix" { - serverArg = map[string]any{"type": "unix", "path": socketPath} - } else { - serverArg = map[string]any{"type": "inet", "host": nbdHost, "port": strconv.Itoa(nbdPort)} - } - - nodeName := fmt.Sprintf("nbd-%s", volumeID) - deviceID := fmt.Sprintf("vdisk-%s", volumeID) - iothreadID := fmt.Sprintf("ioth-%s", volumeID) + // Shared with buildDrives' cold-boot path so a relaunched volume's block + // graph uses exactly the same names DetachVolume addresses. + nodeName := VolumeNodeName(volumeID) + deviceID := VolumeDeviceID(volumeID) + iothreadID := VolumeIOThreadID(volumeID) // Allocate the PCIe hot-plug port from in-memory accounting. QEMU reports // block devices by id (/machine/peripheral//virtio-backend), not by bus, @@ -160,20 +155,13 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) // The virtio-blk-pci device must land on a free hot-plug PCIe root port // (hotplug-ebs{N}); pcie.0 rejects hot-plug. The port is allocated from // live QEMU state above, independent of the AWS device name. - hotplugBus := fmt.Sprintf("hotplug-ebs%d", hotplugPort) + hotplugBus := HotplugEBSBus(hotplugPort) - // serial is the volume-id with dashes stripped ("vol" + 17 hex = 20 bytes, - // the virtio-blk serial limit). It surfaces in-guest as the block device - // serial so the EBS CSI node plugin can locate /dev/disk/by-id and match - // `lsblk -o SERIAL` against the volume-id. - deviceAddArgs := map[string]any{ - "driver": "virtio-blk-pci", - "id": deviceID, - "drive": nodeName, - "iothread": iothreadID, - "serial": strings.ReplaceAll(volumeID, "-", ""), - "bus": hotplugBus, - } + // serial surfaces in-guest as the block device serial so the EBS CSI node + // plugin can locate /dev/disk/by-id and match `lsblk -o SERIAL` against + // the volume-id. Shared with buildDrives so a relaunched volume keeps the + // same serial. + deviceAddArgs := VolumeBlkDeviceQMPArgs(volumeID, nodeName, iothreadID, hotplugBus) if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{ Execute: "device_add", @@ -316,9 +304,12 @@ func (m *Manager) DetachVolume(ctx context.Context, id, volumeID, device string, ErrVolumeDeviceMismatch, device, ebsReq.DeviceName) } - deviceID := fmt.Sprintf("vdisk-%s", volumeID) - nodeName := fmt.Sprintf("nbd-%s", volumeID) - iothreadID := fmt.Sprintf("ioth-%s", volumeID) + // Shared with buildDrives' cold-boot path: a relaunched data volume gets + // exactly these names, so these addresses resolve whether the volume was + // hot-attached or cold-booted. + deviceID := VolumeDeviceID(volumeID) + nodeName := VolumeNodeName(volumeID) + iothreadID := VolumeIOThreadID(volumeID) // device_del is idempotent on DeviceNotFound so a second AWS-CLI // retry can drive blockdev-del to completion when a prior detach left diff --git a/tests/e2e/harness/ec2helpers.go b/tests/e2e/harness/ec2helpers.go index 0d7727064..858444956 100644 --- a/tests/e2e/harness/ec2helpers.go +++ b/tests/e2e/harness/ec2helpers.go @@ -4,6 +4,7 @@ package harness import ( "bytes" + "context" "fmt" "os/exec" "strconv" @@ -327,11 +328,24 @@ func teardownVolume(ec2c ec2iface.EC2API, volID string) error { // terminating and there is nothing left to acknowledge a graceful // detach. Issued once, then we wait for the state to settle. if !forced { - _, _ = ec2c.DetachVolume(&ec2.DetachVolumeInput{ + // Bound the call itself: the SDK's built-in retry (4 attempts + // against the gateway's 30s NATS budget) can burn ~119s on a + // slow detach, more than the whole deadline, leaving no time + // for the state poll below and misreporting a slow detach as + // an unexplained leak. Deriving the context from the same + // deadline this loop already honours means a bounded-out + // detach still leaves the poll loop below a chance to observe + // the volume settle. + detachCtx, cancel := context.WithDeadline(context.Background(), deadline) + _, _ = ec2c.DetachVolumeWithContext(detachCtx, &ec2.DetachVolumeInput{ VolumeId: aws.String(volID), Force: aws.Bool(true), }) + cancel() forced = true + if time.Now().After(deadline) { + return fmt.Errorf("volume %s still %s after %s, not deleted", volID, state, volumeTeardownTimeout) + } continue } if time.Now().After(deadline) { diff --git a/tests/e2e/harness/fixtures_test.go b/tests/e2e/harness/fixtures_test.go index c0cbf6977..ae9c336bc 100644 --- a/tests/e2e/harness/fixtures_test.go +++ b/tests/e2e/harness/fixtures_test.go @@ -13,6 +13,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" + "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/ec2/ec2iface" "github.com/aws/aws-sdk-go/service/elbv2/elbv2iface" @@ -62,6 +63,12 @@ func (f *fakeEC2) DetachVolume(*ec2.DetachVolumeInput) (*ec2.VolumeAttachment, e return &ec2.VolumeAttachment{}, nil } +// DetachVolumeWithContext is what teardownVolume actually calls (bounded by +// the teardown deadline); it shares DetachVolume's behaviour and ignores ctx. +func (f *fakeEC2) DetachVolumeWithContext(_ aws.Context, in *ec2.DetachVolumeInput, _ ...request.Option) (*ec2.VolumeAttachment, error) { + return f.DetachVolume(in) +} + func (f *fakeEC2) DeleteVolume(*ec2.DeleteVolumeInput) (*ec2.DeleteVolumeOutput, error) { f.deleteVolume.Add(1) if f.deleteVolumeErr != nil {