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 {