diff --git a/contracts/viperblockd/legacy/v1/volume.go b/contracts/viperblockd/legacy/v1/volume.go index 9e519efd1..57020d83c 100644 --- a/contracts/viperblockd/legacy/v1/volume.go +++ b/contracts/viperblockd/legacy/v1/volume.go @@ -48,6 +48,11 @@ type EBSRequest struct { NBDURI string `json:"NBDURI"` DeviceName string `json:"DeviceName"` HotplugPort int `json:"HotplugPort,omitempty"` + // BlockDevice and BlockSizeBytes describe a host block device export, the + // alternative to NBDURI; the guest sees BlockSizeBytes of it. Exactly one + // of NBDURI and BlockDevice may be set. + BlockDevice string `json:"BlockDevice,omitempty"` + BlockSizeBytes uint64 `json:"BlockSizeBytes,omitempty"` } // EBSMountResponse is the response to an EBSRequest on MountSubject. diff --git a/contracts/viperblockd/legacy/v1/volume_test.go b/contracts/viperblockd/legacy/v1/volume_test.go index 16f310d87..2571afb03 100644 --- a/contracts/viperblockd/legacy/v1/volume_test.go +++ b/contracts/viperblockd/legacy/v1/volume_test.go @@ -35,6 +35,18 @@ func TestEBSRequestJSONContract(t *testing.T) { "NBDURI":"nbd:unix:/run/spinifex/vol-123.sock", "DeviceName":"/dev/sdf", "HotplugPort":3 }`, string(data)) + + block, err := json.Marshal(EBSRequest{ + Name: "vol-456", + BlockDevice: "/dev/sdx", + BlockSizeBytes: 10 << 30, + }) + require.NoError(t, err) + require.JSONEq(t, `{ + "Name":"vol-456", "VolType":"", "Boot":false, "EFI":false, + "DeleteOnTermination":false, "NBDURI":"", "DeviceName":"", + "BlockDevice":"/dev/sdx", "BlockSizeBytes":10737418240 + }`, string(block)) } func TestVolumeResponseJSONContracts(t *testing.T) { diff --git a/spinifex/runtime/compute/vm/devices.go b/spinifex/runtime/compute/vm/devices.go index 13735a021..97b56096c 100644 --- a/spinifex/runtime/compute/vm/devices.go +++ b/spinifex/runtime/compute/vm/devices.go @@ -355,3 +355,35 @@ func VolumeBlockdev(nodeName string, server NBDServerOpts) Blockdev { opts = append(opts, "export=", fmt.Sprintf("reconnect-delay=%d", NBDReconnectDelaySeconds)) return Blockdev{Value: strings.Join(opts, ",")} } + +// HostDeviceOpts is a host block device export. QEMU opens Path through the +// host_device protocol driver with O_DIRECT and native AIO, and a raw format +// node over it exposes only the first SizeBytes to the guest. +type HostDeviceOpts struct { + Path string + SizeBytes uint64 +} + +// QMPArgs renders the blockdev-add arguments for the raw node and its nested +// host_device file node. +func (o HostDeviceOpts) QMPArgs(nodeName string) map[string]any { + return map[string]any{ + "node-name": nodeName, + "driver": "raw", + "size": o.SizeBytes, + "file": map[string]any{ + "driver": "host_device", + "filename": o.Path, + "aio": "native", + "cache": map[string]any{"direct": true}, + }, + } +} + +// VolumeHostDeviceBlockdev renders the -blockdev equivalent of QMPArgs for a +// data volume backed by a host block device. +func VolumeHostDeviceBlockdev(nodeName string, dev HostDeviceOpts) Blockdev { + return Blockdev{Value: fmt.Sprintf( + "driver=raw,node-name=%s,size=%d,file.driver=host_device,file.filename=%s,file.aio=native,file.cache.direct=on", + nodeName, dev.SizeBytes, dev.Path)} +} diff --git a/spinifex/runtime/compute/vm/ebs_requests.go b/spinifex/runtime/compute/vm/ebs_requests.go index cf1711b6b..258978aa3 100644 --- a/spinifex/runtime/compute/vm/ebs_requests.go +++ b/spinifex/runtime/compute/vm/ebs_requests.go @@ -1,6 +1,10 @@ package vm import ( + "fmt" + "os" + "path/filepath" + "strings" "sync" viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" @@ -20,3 +24,45 @@ type EBSRequests struct { // VM's persisted attachment list. The alias keeps the VM state tied to the // deployed wire shape without making the old catch-all types package an owner. type EBSRequest = viperblocklegacyv1.EBSRequest + +// volumeExportKind is how a volume's backing store reaches QEMU. +type volumeExportKind int + +const ( + volumeExportInvalid volumeExportKind = iota + volumeExportNBD + volumeExportBlock +) + +// classifyVolumeExport reports which export a request carries. A block device +// must be absolute, comma-free (QEMU key=value options would split it) and +// sized; both exports set, neither, or a stray size is invalid. +func classifyVolumeExport(req EBSRequest) (volumeExportKind, error) { + switch { + case req.NBDURI != "" && req.BlockDevice != "": + return volumeExportInvalid, fmt.Errorf("volume %s has both NBDURI and BlockDevice set", req.Name) + case req.BlockSizeBytes != 0 && req.BlockDevice == "": + return volumeExportInvalid, fmt.Errorf("volume %s has BlockSizeBytes set without BlockDevice", req.Name) + case req.NBDURI != "": + return volumeExportNBD, nil + case req.BlockDevice != "": + if !filepath.IsAbs(req.BlockDevice) { + return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q is not an absolute path", req.Name, req.BlockDevice) + } + if strings.Contains(req.BlockDevice, ",") { + return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q contains a comma", req.Name, req.BlockDevice) + } + if req.BlockSizeBytes == 0 { + return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q has zero BlockSizeBytes", req.Name, req.BlockDevice) + } + return volumeExportBlock, nil + default: + return volumeExportInvalid, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", req.Name) + } +} + +// checkBlockDevice reports whether a block export's device node is present. +func checkBlockDevice(path string) error { + _, err := os.Stat(path) + return err +} diff --git a/spinifex/runtime/compute/vm/lifecycle.go b/spinifex/runtime/compute/vm/lifecycle.go index 47447b3aa..b96ce6cf5 100644 --- a/spinifex/runtime/compute/vm/lifecycle.go +++ b/spinifex/runtime/compute/vm/lifecycle.go @@ -607,11 +607,18 @@ func (m *Manager) startQEMU(instance *VM) error { refreshTelemetryMeta(instance) } + // An NBD export is waited on until it accepts connections; a block export + // is local to the host and only has to exist. An invalid request has no + // endpoint to probe (buildDrives rejects it on the command-line path). instance.EBSRequests.Mu.Lock() nbdEndpoints := make([]struct{ name, uri string }, 0, len(instance.EBSRequests.Requests)) + var blockDevices []struct{ name, path string } for _, req := range instance.EBSRequests.Requests { - if req.NBDURI != "" { + switch kind, _ := classifyVolumeExport(req); kind { + case volumeExportNBD: nbdEndpoints = append(nbdEndpoints, struct{ name, uri string }{req.Name, req.NBDURI}) + case volumeExportBlock: + blockDevices = append(blockDevices, struct{ name, path string }{req.Name, req.BlockDevice}) } } instance.EBSRequests.Mu.Unlock() @@ -620,6 +627,11 @@ func (m *Manager) startQEMU(instance *VM) error { return fmt.Errorf("nbd endpoint not ready for %s: %w", ep.name, err) } } + for _, dev := range blockDevices { + if err := checkBlockDevice(dev.path); err != nil { + return fmt.Errorf("block device not present for %s: %w", dev.name, err) + } + } processChan := make(chan int, 1) exitChan := make(chan int, 1) @@ -900,11 +912,11 @@ func qmpGreetingTimeout(v *VM) time.Duration { // hasNetworkBackedDrive reports whether any of the guest's drives is served // over NBD, and so has to be opened across the object store before QEMU's -// monitor answers. Every production guest boots from one; a VM without is a -// unit test. +// monitor answers. A host block device opens locally and does not count. func hasNetworkBackedDrive(v *VM) bool { return slices.ContainsFunc(v.EBSRequests.Requests, func(req EBSRequest) bool { - return req.NBDURI != "" + kind, _ := classifyVolumeExport(req) + return kind == volumeExportNBD }) } @@ -1445,8 +1457,9 @@ type driveConfig struct { } // 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. +// configs. Returns an error if any volume has no valid export (see +// classifyVolumeExport), 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 @@ -1471,11 +1484,17 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive for i := range requests { v := &requests[i] - if v.NBDURI == "" { - return driveConfig{}, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name) + kind, err := classifyVolumeExport(*v) + if err != nil { + return driveConfig{}, err } switch { + case kind == volumeExportBlock && (v.Boot || v.EFI): + // Boot and EFI keep the legacy -drive shape, which has no + // host_device form here; only data volumes take a block export. + return driveConfig{}, fmt.Errorf("volume %s: block device export is only supported for data volumes", v.Name) + case v.Boot: drive := Drive{ File: v.NBDURI, @@ -1529,29 +1548,39 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive v.HotplugPort = port } - serverType, socketPath, nbdHost, nbdPort, err := nbd.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) + var blockdev Blockdev + if kind == volumeExportBlock { + blockdev = VolumeHostDeviceBlockdev(nodeName, HostDeviceOpts{Path: v.BlockDevice, SizeBytes: v.BlockSizeBytes}) + } else { + serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(v.NBDURI) + if err != nil { + return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err) + } + blockdev = VolumeBlockdev(nodeName, NBDServerOpts{ + Type: serverType, + Path: socketPath, + Host: nbdHost, + Port: nbdPort, + }) + } + 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.Blockdevs = append(cfg.Blockdevs, blockdev) 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) + if kind == volumeExportBlock { + slog.Info("Using block device for drive", "volume", v.Name, "device", v.BlockDevice, "size_bytes", v.BlockSizeBytes) + } else { + slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI) + } } return cfg, nil diff --git a/spinifex/runtime/compute/vm/lifecycle_test.go b/spinifex/runtime/compute/vm/lifecycle_test.go index 04ff0e759..a10e519cc 100644 --- a/spinifex/runtime/compute/vm/lifecycle_test.go +++ b/spinifex/runtime/compute/vm/lifecycle_test.go @@ -273,6 +273,83 @@ func TestBuildDrives(t *testing.T) { }, wantHotplugPorts: []int{0, 0, 3}, }, + { + name: "block data volume emits a raw node over host_device with the same device wiring", + requests: []EBSRequest{ + {Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 10 << 30, HotplugPort: 2}, + }, + cpuCount: 2, + wantIOThreads: []IOThread{{ID: "ioth-vol-blk"}}, + wantBlockdevs: []Blockdev{ + {Value: "driver=raw,node-name=nbd-vol-blk,size=10737418240,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"}, + }, + wantDevices: []Device{ + {Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs2,werror=stop,rerror=stop"}, + }, + wantHotplugPorts: []int{2}, + }, + { + name: "NBD boot volume alongside a block data volume", + requests: []EBSRequest{ + {Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true}, + {Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20}, + }, + cpuCount: 2, + wantDrives: []Drive{ + {File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds}, + }, + wantIOThreads: []IOThread{{ID: "ioth-os"}, {ID: "ioth-vol-blk"}}, + wantBlockdevs: []Blockdev{ + {Value: "driver=raw,node-name=nbd-vol-blk,size=1048576,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"}, + }, + wantDevices: []Device{ + {Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"}, + {Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs1,werror=stop,rerror=stop"}, + }, + wantHotplugPorts: []int{0, 1}, + }, + { + name: "block boot volume is rejected", + requests: []EBSRequest{{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, Boot: true}}, + cpuCount: 2, + wantErr: "volume vol-boot: block device export is only supported for data volumes", + }, + { + name: "block EFI volume is rejected", + requests: []EBSRequest{{Name: "vol-efi", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, EFI: true}}, + cpuCount: 2, + wantErr: "volume vol-efi: block device export is only supported for data volumes", + }, + { + name: "both NBDURI and BlockDevice set is rejected", + requests: []EBSRequest{{Name: "vol-both", NBDURI: "nbd:unix:/tmp/x.sock", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20}}, + cpuCount: 2, + wantErr: "volume vol-both has both NBDURI and BlockDevice set", + }, + { + name: "NBDURI with a stray BlockSizeBytes is rejected", + requests: []EBSRequest{{Name: "vol-stray", NBDURI: "nbd:unix:/tmp/x.sock", BlockSizeBytes: 1 << 20}}, + cpuCount: 2, + wantErr: "volume vol-stray has BlockSizeBytes set without BlockDevice", + }, + { + name: "block device with zero size is rejected", + requests: []EBSRequest{{Name: "vol-nosize", BlockDevice: "/dev/sdx"}}, + cpuCount: 2, + wantErr: `volume vol-nosize BlockDevice "/dev/sdx" has zero BlockSizeBytes`, + }, + { + name: "relative block device path is rejected", + requests: []EBSRequest{{Name: "vol-rel", BlockDevice: "dev/sdx", BlockSizeBytes: 1 << 20}}, + cpuCount: 2, + wantErr: `volume vol-rel BlockDevice "dev/sdx" is not an absolute path`, + }, + { + name: "block device path with a comma is rejected", + requests: []EBSRequest{{Name: "vol-comma", BlockDevice: "/dev/sdx,file.locking=off", BlockSizeBytes: 1 << 20}}, + cpuCount: 2, + wantErr: `volume vol-comma BlockDevice "/dev/sdx,file.locking=off" contains a comma`, + }, { name: "empty requests", requests: []EBSRequest{}, @@ -1008,6 +1085,28 @@ func TestStartQEMU_DirectBoot_InstanceTypeNotFound(t *testing.T) { assert.Empty(t, plumber.setupCalls) } +// TestStartQEMU_BlockDeviceMissing verifies a block volume's launch readiness +// is the device's presence, not an NBD connect. +func TestStartQEMU_BlockDeviceMissing(t *testing.T) { + m := directBootManager(t, &fakeNetworkPlumber{}) + missing := filepath.Join(t.TempDir(), "missing-dev") + + instance := &VM{ + ID: "i-blk-missing", + InstanceType: "t3.nano", + DirectBoot: true, + } + instance.EBSRequests.Requests = []EBSRequest{ + {Name: "vol-blk", BlockDevice: missing, BlockSizeBytes: 1 << 20}, + } + + err := m.startQEMU(instance) + + require.Error(t, err) + assert.Contains(t, err.Error(), "block device not present for vol-blk") + assert.ErrorIs(t, err, os.ErrNotExist) +} + // scriptedNetworkPlumber returns errors from a pre-loaded slice, one per call. type scriptedNetworkPlumber struct { calls []TapSpec @@ -1271,6 +1370,12 @@ func TestQMPGreetingTimeout(t *testing.T) { assert.Greater(t, qmpNBDGreetingTimeout, 35*time.Second, "the deadline must exceed the 35s cold open measured with one node of three away") + block := &VM{EBSRequests: EBSRequests{Requests: []EBSRequest{ + {Name: "vol-data", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20}, + }}} + assert.Equal(t, qmp.DefaultGreetingTimeout, qmpGreetingTimeout(block), + "a host block device opens locally, so it keeps the plain default") + // A small GPU guest (<1 GiB scaling contribution) sits at the floor. smallGPU := &VM{ GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:5e:00.0"}}, diff --git a/spinifex/runtime/compute/vm/manager_deps_test.go b/spinifex/runtime/compute/vm/manager_deps_test.go index c66070afb..a5dcf8431 100644 --- a/spinifex/runtime/compute/vm/manager_deps_test.go +++ b/spinifex/runtime/compute/vm/manager_deps_test.go @@ -142,6 +142,10 @@ type fakeVolumeMounter struct { unmountErr error unmountOneErr error mountOneURI string + // mountOneBlockDevice and mountOneBlockSize, when set, are written onto + // the request by MountOne as a block device export. + mountOneBlockDevice string + mountOneBlockSize uint64 // onMount fires synchronously inside Mount before the configured // mountErr is returned. Used by lifecycle tests to simulate a // concurrent terminate flipping VM.Status while Mount is in flight. @@ -181,6 +185,7 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques f.mountedOne = append(f.mountedOne, req.Name) mountOneErr := f.mountOneErr uri := f.mountOneURI + blockDevice, blockSize := f.mountOneBlockDevice, f.mountOneBlockSize f.mu.Unlock() if mountOneErr != nil { return mountOneErr @@ -188,6 +193,10 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques if uri != "" { req.NBDURI = uri } + if blockDevice != "" { + req.BlockDevice = blockDevice + req.BlockSizeBytes = blockSize + } return nil } diff --git a/spinifex/runtime/compute/vm/restore.go b/spinifex/runtime/compute/vm/restore.go index 73de81bf5..aea43eb2a 100644 --- a/spinifex/runtime/compute/vm/restore.go +++ b/spinifex/runtime/compute/vm/restore.go @@ -523,14 +523,22 @@ func isInstanceProcessRunning(instance *VM) bool { } // AreVolumeSocketsValid dials each NBD Unix socket to confirm viperblock is -// still listening. A stat-only check is insufficient since viperblock may restart -// leaving stale socket files. TCP and unparseable URIs are treated as valid. +// still listening, and checks each block export's device still exists. TCP, +// unparseable and invalid exports are treated as valid. func AreVolumeSocketsValid(instance *VM) bool { instance.EBSRequests.Mu.Lock() defer instance.EBSRequests.Mu.Unlock() for _, req := range instance.EBSRequests.Requests { - if req.NBDURI == "" { + kind, _ := classifyVolumeExport(req) + if kind == volumeExportBlock { + if err := checkBlockDevice(req.BlockDevice); err != nil { + slog.Debug("Block device missing", "volume", req.Name, "device", req.BlockDevice, "err", err) + return false + } + continue + } + if kind != volumeExportNBD { continue } serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI) diff --git a/spinifex/runtime/compute/vm/restore_test.go b/spinifex/runtime/compute/vm/restore_test.go index 349172dae..94be518f4 100644 --- a/spinifex/runtime/compute/vm/restore_test.go +++ b/spinifex/runtime/compute/vm/restore_test.go @@ -82,6 +82,22 @@ func TestAreVolumeSocketsValid_OneOfManyMissing(t *testing.T) { "a single dead socket is enough to invalidate the whole instance") } +// TestAreVolumeSocketsValid_BlockDevice checks a block export is judged by its +// device node's presence, with no NBD probe. +func TestAreVolumeSocketsValid_BlockDevice(t *testing.T) { + present := filepath.Join(t.TempDir(), "dev-present") + require.NoError(t, os.WriteFile(present, nil, 0o600)) + missing := filepath.Join(t.TempDir(), "dev-missing") + + blockVM := func(path string) *VM { + v := &VM{ID: "i-blk", Status: StateRunning} + v.EBSRequests.Requests = []EBSRequest{{Name: "vol-blk", BlockDevice: path, BlockSizeBytes: 1 << 20}} + return v + } + assert.True(t, AreVolumeSocketsValid(blockVM(present)), "a present block device is valid") + assert.False(t, AreVolumeSocketsValid(blockVM(missing)), "a vanished block device is the orphan-QEMU signal") +} + // classifyTestManager wires a manager with an in-memory StateStore plus // the InstanceTypes resolver and Resources controller that // classifyRestoredInstances consults. Returns the manager and the diff --git a/spinifex/runtime/compute/vm/volumes.go b/spinifex/runtime/compute/vm/volumes.go index 40b101eca..00b0c3acf 100644 --- a/spinifex/runtime/compute/vm/volumes.go +++ b/spinifex/runtime/compute/vm/volumes.go @@ -119,17 +119,36 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) return "", fmt.Errorf("mount volume %s: %w", volumeID, err) } - serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI) - if err != nil { - slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err) - _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) - return "", fmt.Errorf("parse NBDURI: %w", err) - } - serverArg := NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg() - // Shared with buildDrives' cold-boot path so a relaunched volume's block // graph uses exactly the same names DetachVolume addresses. nodeName := VolumeNodeName(volumeID) + + kind, err := classifyVolumeExport(ebsRequest) + if err != nil { + slog.ErrorContext(ctx, "AttachVolume: invalid volume export", "volumeId", volumeID, "err", err) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) + return "", fmt.Errorf("volume export: %w", err) + } + var blockdevArgs map[string]any + if kind == volumeExportBlock { + blockdevArgs = HostDeviceOpts{Path: ebsRequest.BlockDevice, SizeBytes: ebsRequest.BlockSizeBytes}.QMPArgs(nodeName) + } else { + serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI) + if err != nil { + slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err) + _ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest) + return "", fmt.Errorf("parse NBDURI: %w", err) + } + blockdevArgs = map[string]any{ + "node-name": nodeName, + "driver": "nbd", + "server": NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg(), + "export": "", + "read-only": false, + "reconnect-delay": NBDReconnectDelaySeconds, + } + } + deviceID := VolumeDeviceID(volumeID) iothreadID := VolumeIOThreadID(volumeID) @@ -163,15 +182,8 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string) } if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{ - Execute: "blockdev-add", - Arguments: map[string]any{ - "node-name": nodeName, - "driver": "nbd", - "server": serverArg, - "export": "", - "read-only": false, - "reconnect-delay": NBDReconnectDelaySeconds, - }, + Execute: "blockdev-add", + Arguments: blockdevArgs, }, instance.ID); err != nil { slog.ErrorContext(ctx, "AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) m.delIothreadBestEffort(ctx, instance, iothreadID, volumeID) diff --git a/spinifex/runtime/compute/vm/volumes_test.go b/spinifex/runtime/compute/vm/volumes_test.go index 348e86c0f..e52d7cff8 100644 --- a/spinifex/runtime/compute/vm/volumes_test.go +++ b/spinifex/runtime/compute/vm/volumes_test.go @@ -943,6 +943,91 @@ func TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne(t *testing.T) { assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) } +// TestAttachVolume_BlockDeviceBlockdevAdd pins the blockdev-add a block export +// sends: a sized raw node over a host_device file node, under the same node +// name the NBD path uses, followed by the usual device_add. +func TestAttachVolume_BlockDeviceBlockdevAdd(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 10 << 30} + m, v := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf") + require.NoError(t, err) + + recorder.mu.Lock() + var blockdevAdd *qmp.QMPCommand + for i := range recorder.commands { + if recorder.commands[i].Execute == "blockdev-add" { + blockdevAdd = &recorder.commands[i] + } + } + recorder.mu.Unlock() + require.NotNil(t, blockdevAdd) + got, err := json.Marshal(blockdevAdd.Arguments) + require.NoError(t, err) + assert.JSONEq(t, `{ + "node-name":"nbd-vol-1", "driver":"raw", "size":10737418240, + "file":{"driver":"host_device", "filename":"/dev/sdx", "aio":"native", "cache":{"direct":true}} + }`, string(got)) + assert.Equal(t, []string{"object-add", "blockdev-add", "device_add"}, recorder.executes()[:3]) + + v.EBSRequests.Mu.Lock() + defer v.EBSRequests.Mu.Unlock() + require.Len(t, v.EBSRequests.Requests, 1) + assert.Equal(t, "/dev/sdx", v.EBSRequests.Requests[0].BlockDevice) + assert.Equal(t, uint64(10<<30), v.EBSRequests.Requests[0].BlockSizeBytes) + assert.Empty(t, v.EBSRequests.Requests[0].NBDURI) +} + +// TestAttachVolume_InvalidExport_TriggersUnmountOne covers a mount that +// returns both exports: AttachVolume must unmount before any QMP traffic. +func TestAttachVolume_InvalidExport_TriggersUnmountOne(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock", mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "both NBDURI and BlockDevice set") + assert.Empty(t, recorder.executes()) + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) +} + +// TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne checks a +// block export unwinds a failed blockdev-add exactly as an NBD one does. +func TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + if cmd.Execute == "blockdev-add" { + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "Could not open '/dev/sdx'"}} + } + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "blockdev-add") + assert.Equal(t, []string{"object-add", "blockdev-add", "object-del"}, recorder.executes()) + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) +} + // TestAttachVolume_ObjectAddFailure_TriggersUnmountOne covers QMP iothread // object-add failure: AttachVolume must unmount and bail before issuing // blockdev-add (no further QMP traffic past the failed object-add).