gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-22 04:07:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(vm): move AttachVolume/DetachVolume/Reboot onto Manager (#189)

Browse Source
This commit is contained in:
Julian Sommer
2026-05-05 14:27:59 +10:00
committed by GitHub
parent cfe56c49d1
commit c2be7f047c
15 changed files with 1365 additions and 717 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/daemon/daemon.go
-110
View File
@@ -1687,82 +1687,6 @@ func (d *Daemon) ebsTopic(action string) string {
return fmt.Sprintf("ebs.%s.%s", d.node, action)
}
// MountVolumes mounts the volumes for an instance
func (d *Daemon) MountVolumes(instance *vm.VM) error {
instance.EBSRequests.Mu.Lock()
defer instance.EBSRequests.Mu.Unlock()
for k, v := range instance.EBSRequests.Requests {
// Send the volume payload as JSON
ebsMountRequest, err := json.Marshal(v)
if err != nil {
slog.Error("Failed to marshal volume payload", "err", err)
return err
}
reply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountRequest, 30*time.Second)
slog.Info("Mounting volume", "Vol", v.Name, "NBDURI", v.NBDURI)
// TODO: Improve timeout handling
if err != nil {
slog.Error("Failed to request EBS mount", "err", err)
return err
}
// Unmarshal the response
var ebsMountResponse types.EBSMountResponse
err = json.Unmarshal(reply.Data, &ebsMountResponse)
if err != nil {
slog.Error("Failed to unmarshal volume response:", "err", err)
return err
}
if ebsMountResponse.Error == "" {
slog.Debug("Mounted volume successfully", "response", ebsMountResponse.URI)
// Append the NBD URI to the request
instance.EBSRequests.Requests[k].NBDURI = ebsMountResponse.URI
} else {
slog.Error("Failed to mount volume", "error", ebsMountResponse.Error)
return fmt.Errorf("failed to mount volume: %s", ebsMountResponse.Error)
}
}
return nil
}
// rollbackEBSMount sends an ebs.unmount request to undo a previously successful ebs.mount.
// Rollback failures are logged but not propagated; callers treat this as best-effort cleanup.
func (d *Daemon) rollbackEBSMount(req types.EBSRequest) {
data, err := json.Marshal(req)
if err != nil {
slog.Error("rollbackEBSMount: failed to marshal unmount request", "volume", req.Name, "err", err)
return
}
msg, err := d.natsConn.Request(d.ebsTopic("unmount"), data, 10*time.Second)
if err != nil {
slog.Error("rollbackEBSMount: ebs.unmount NATS request failed", "volume", req.Name, "err", err)
return
}
var resp types.EBSUnMountResponse
if err := json.Unmarshal(msg.Data, &resp); err != nil {
slog.Error("rollbackEBSMount: failed to unmarshal response", "volume", req.Name, "err", err)
return
}
if resp.Error != "" {
slog.Error("rollbackEBSMount: ebs.unmount returned error", "volume", req.Name, "err", resp.Error)
return
}
if resp.Mounted {
slog.Error("rollbackEBSMount: volume still mounted after unmount", "volume", req.Name)
return
}
slog.Info("rollbackEBSMount: volume unmounted successfully", "volume", req.Name)
}
// respondWithVolumeAttachment builds an ec2.VolumeAttachment, marshals it to JSON, and
// responds on the NATS message. Used by both AttachVolume and DetachVolume handlers.
func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID, device, state string) {
@@ -1787,40 +1711,6 @@ func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID
}
}
// nextAvailableDevice finds the next available /dev/sd[f-p] device name for an instance.
// It checks both EBSRequests and BlockDeviceMappings to avoid conflicts.
func nextAvailableDevice(instance *vm.VM) string {
usedDevices := make(map[string]bool)
// Collect devices from existing BlockDeviceMappings
if instance.Instance != nil {
for _, bdm := range instance.Instance.BlockDeviceMappings {
if bdm.DeviceName != nil {
usedDevices[*bdm.DeviceName] = true
}
}
}
// Collect devices from EBSRequests (may not yet be in BlockDeviceMappings)
instance.EBSRequests.Mu.Lock()
for _, req := range instance.EBSRequests.Requests {
if req.DeviceName != "" {
usedDevices[req.DeviceName] = true
}
}
instance.EBSRequests.Mu.Unlock()
// AWS convention: /dev/sd[f-p] for attached volumes
for c := 'f'; c <= 'p'; c++ {
dev := fmt.Sprintf("/dev/sd%c", c)
if !usedDevices[dev] {
return dev
}
}
return ""
}
// canAllocate checks how many instances of the given type can be allocated.
// Returns the count that can actually be allocated (0 to count).
func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo, count int) int {
spinifex/daemon/daemon_handlers_instance.go
+16 -17
View File
@@ -3,6 +3,7 @@ package daemon
import (
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
@@ -12,7 +13,6 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/filterutil"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/vm"
@@ -391,7 +391,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
}
// Discover actual guest device names via QMP query-block
d.updateGuestDeviceNames(instance)
d.vmMgr.UpdateGuestDeviceNames(instance)
successCount++
slog.Info("handleEC2RunInstances launched instance", "instanceId", instance.ID)
@@ -403,19 +403,18 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
func (d *Daemon) handleRebootInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
slog.Info("Rebooting instance", "id", command.ID)
var status vm.InstanceState
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
if status != vm.StateRunning {
slog.Error("RebootInstance: instance not in running state", "instanceId", command.ID, "status", status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
_, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_reset"}, command.ID)
if err != nil {
slog.Error("RebootInstance: QMP system_reset failed", "instanceId", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
if err := d.vmMgr.Reboot(instance.ID); err != nil {
switch {
case errors.Is(err, vm.ErrInstanceNotFound):
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
case errors.Is(err, vm.ErrInvalidTransition):
slog.Error("RebootInstance: instance not in running state",
"instanceId", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
default:
slog.Error("RebootInstance: reboot failed", "instanceId", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
}
return
}
@@ -465,7 +464,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom
}
// Discover actual guest device names via QMP query-block
d.updateGuestDeviceNames(instance)
d.vmMgr.UpdateGuestDeviceNames(instance)
slog.Info("Instance started", "instanceId", instance.ID)
@@ -878,7 +877,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
}
// Discover actual guest device names via QMP query-block
d.updateGuestDeviceNames(instance)
d.vmMgr.UpdateGuestDeviceNames(instance)
// Remove from shared KV now that it's running locally.
// Retry once on failure — a stale KV entry risks duplicate starts.
spinifex/daemon/daemon_handlers_volume.go
+49 -427
View File
@@ -2,30 +2,26 @@ package daemon
import (
"encoding/json"
"fmt"
"errors"
"log/slog"
"strconv"
"strings"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/nats-io/nats.go"
)
// handleAttachVolume performs a three-phase hot-plug:
//
// Phase 1: ebs.mount via NATS (rolls back with ebs.unmount)
// Phase 2: QMP blockdev-add (rolls back Phase 1)
// Phase 3: QMP device_add (rolls back Phase 2 + Phase 1)
// handleAttachVolume validates the volume request against the volume
// service (existence, ownership, availability, AZ) and dispatches the
// QMP/state-machine pipeline to vm.Manager.AttachVolume. The manager owns
// every QMP and persistence side-effect; the daemon only emits the AWS API
// response.
func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
slog.Info("Attaching volume to instance", "instanceId", command.ID)
// Validate AttachVolumeData
if command.AttachVolumeData == nil || command.AttachVolumeData.VolumeID == "" {
slog.Error("AttachVolume: missing attach volume data")
respondWithError(msg, awserrors.ErrorInvalidParameterValue)
@@ -33,19 +29,18 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm
}
volumeID := command.AttachVolumeData.VolumeID
device := command.AttachVolumeData.Device
// Validate instance is running
// Surface IncorrectInstanceState before any volume-side lookups so the
// AWS-SDK error matches the pre-2d ordering.
var status vm.InstanceState
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
if status != vm.StateRunning {
slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status)
slog.Error("AttachVolume: instance not running",
"instanceId", command.ID, "status", status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Validate volume exists and is available
volCfg, err := d.volumeService.GetVolumeConfig(volumeID)
if err != nil {
slog.Error("AttachVolume: failed to get volume config", "volumeId", volumeID, "err", err)
@@ -53,400 +48,86 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm
return
}
// Account scoping: verify the caller owns this volume.
// Pre-Phase4 volumes (empty TenantID) are root-only; otherwise the caller
// must match the recorded tenant exactly.
callerAccountID := utils.AccountIDFromMsg(msg)
if !volumeVisibleTo(volCfg.VolumeMetadata.TenantID, callerAccountID) {
slog.Warn("AttachVolume: account does not own volume", "volumeId", volumeID, "callerAccount", callerAccountID, "ownerAccount", volCfg.VolumeMetadata.TenantID)
slog.Warn("AttachVolume: account does not own volume",
"volumeId", volumeID,
"callerAccount", callerAccountID,
"ownerAccount", volCfg.VolumeMetadata.TenantID)
respondWithError(msg, awserrors.ErrorInvalidVolumeNotFound)
return
}
if volCfg.VolumeMetadata.State != "available" {
slog.Error("AttachVolume: volume not available", "volumeId", volumeID, "state", volCfg.VolumeMetadata.State)
slog.Error("AttachVolume: volume not available",
"volumeId", volumeID, "state", volCfg.VolumeMetadata.State)
respondWithError(msg, awserrors.ErrorVolumeInUse)
return
}
// Check AZ compatibility — volume and instance must be in the same AZ
if volCfg.VolumeMetadata.AvailabilityZone != "" && d.config.AZ != "" &&
volCfg.VolumeMetadata.AvailabilityZone != d.config.AZ {
slog.Error("AttachVolume: volume and instance are in different AZs",
"volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone,
"volumeId", volumeID,
"volumeAZ", volCfg.VolumeMetadata.AvailabilityZone,
"instanceAZ", d.config.AZ)
respondWithError(msg, awserrors.ErrorInvalidVolumeZoneMismatch)
return
}
// Determine device name
if device == "" {
d.vmMgr.Inspect(instance, func(v *vm.VM) { device = nextAvailableDevice(v) })
if device == "" {
slog.Error("AttachVolume: no available device names")
respondWithError(msg, awserrors.ErrorAttachmentLimitExceeded)
return
}
}
// Create EBS mount request
ebsRequest := types.EBSRequest{
Name: volumeID,
DeviceName: device,
}
ebsMountData, err := json.Marshal(ebsRequest)
res, err := d.vmMgr.AttachVolume(instance.ID, volumeID, command.AttachVolumeData.Device)
if err != nil {
slog.Error("AttachVolume: failed to marshal ebs.mount request", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
respondWithError(msg, attachDetachErrorCode(err))
return
}
mountReply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountData, 30*time.Second)
if err != nil {
slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
var mountResp types.EBSMountResponse
if err := json.Unmarshal(mountReply.Data, &mountResp); err != nil {
slog.Error("AttachVolume: failed to unmarshal mount response", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if mountResp.Error != "" {
slog.Error("AttachVolume: mount returned error", "volumeId", volumeID, "err", mountResp.Error)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
nbdURI := mountResp.URI
if nbdURI == "" {
slog.Error("AttachVolume: mount response has empty URI", "volumeId", volumeID)
d.rollbackEBSMount(ebsRequest)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
ebsRequest.NBDURI = nbdURI
// Parse NBDURI for QMP blockdev-add
serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(nbdURI)
if err != nil {
slog.Error("AttachVolume: failed to parse NBDURI", "uri", nbdURI, "err", err)
// Rollback: unmount
d.rollbackEBSMount(ebsRequest)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Build QMP server argument
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)
// QMP object-add: create iothread for this volume
iothreadCmd := qmp.QMPCommand{
Execute: "object-add",
Arguments: map[string]any{
"qom-type": "iothread",
"id": iothreadID,
},
}
_, err = d.SendQMPCommand(instance.QMPClient, iothreadCmd, instance.ID)
if err != nil {
slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err)
d.rollbackEBSMount(ebsRequest)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// QMP blockdev-add
blockdevCmd := qmp.QMPCommand{
Execute: "blockdev-add",
Arguments: map[string]any{
"node-name": nodeName,
"driver": "nbd",
"server": serverArg,
"export": "",
"read-only": false,
},
}
_, err = d.SendQMPCommand(instance.QMPClient, blockdevCmd, instance.ID)
if err != nil {
slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err)
d.rollbackEBSMount(ebsRequest)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Determine which hotplug root port to use based on device letter.
// /dev/sdf -> hotplug1, /dev/sdg -> hotplug2, etc.
hotplugBus := ""
if len(device) > 0 {
letter := device[len(device)-1]
if letter >= 'f' && letter <= 'p' {
hotplugBus = fmt.Sprintf("hotplug%d", letter-'f'+1)
}
}
// QMP device_add
deviceAddArgs := map[string]any{
"driver": "virtio-blk-pci",
"id": deviceID,
"drive": nodeName,
"iothread": iothreadID,
}
if hotplugBus != "" {
deviceAddArgs["bus"] = hotplugBus
}
deviceAddCmd := qmp.QMPCommand{
Execute: "device_add",
Arguments: deviceAddArgs,
}
_, err = d.SendQMPCommand(instance.QMPClient, deviceAddCmd, instance.ID)
if err != nil {
slog.Error("AttachVolume: QMP device_add failed, rolling back blockdev", "volumeId", volumeID, "err", err)
// Rollback: blockdev-del then unmount
if _, delErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-del",
Arguments: map[string]any{"node-name": nodeName},
}, instance.ID); delErr != nil {
slog.Error("AttachVolume: rollback blockdev-del failed, skipping EBS unmount", "volumeId", volumeID, "err", delErr)
} else {
d.rollbackEBSMount(ebsRequest)
}
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Discover actual guest device name via QMP query-block.
// Use the retry variant because query-block may not include the device
// immediately after device_add.
guestDevice := device // fallback to AWS API name
deviceMap, qmpErr := queryGuestDeviceMapWait(d, instance.QMPClient, instance.ID, deviceID)
if qmpErr != nil {
slog.Warn("AttachVolume: failed to query guest device map, using API device name", "volumeId", volumeID, "err", qmpErr)
} else if gd, ok := deviceMap[deviceID]; ok {
guestDevice = gd
slog.Info("AttachVolume: discovered guest device", "volumeId", volumeID, "qemuDevice", deviceID, "guestDevice", guestDevice)
} else {
slog.Error("AttachVolume: device not found in QMP device map after retries, using API device name",
"volumeId", volumeID, "qemuDevice", deviceID, "deviceMap", deviceMap)
}
// Update instance state: replace existing entry for this volume (handles
// stop/start cycles that keep stale entries) or append a new one.
instance.EBSRequests.Mu.Lock()
replaced := false
for i, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
instance.EBSRequests.Requests[i] = ebsRequest
replaced = true
break
}
}
if !replaced {
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest)
}
instance.EBSRequests.Mu.Unlock()
// Update BlockDeviceMappings on the ec2.Instance using actual guest device name
d.vmMgr.Inspect(instance, func(v *vm.VM) {
if v.Instance == nil {
return
}
now := time.Now()
mapping := &ec2.InstanceBlockDeviceMapping{}
mapping.SetDeviceName(guestDevice)
mapping.Ebs = &ec2.EbsInstanceBlockDevice{}
mapping.Ebs.SetVolumeId(volumeID)
mapping.Ebs.SetAttachTime(now)
mapping.Ebs.SetDeleteOnTermination(false)
mapping.Ebs.SetStatus("attached")
v.Instance.BlockDeviceMappings = append(v.Instance.BlockDeviceMappings, mapping)
})
// Update volume metadata in S3
if err := d.volumeService.UpdateVolumeState(volumeID, "in-use", command.ID, guestDevice); err != nil {
slog.Error("AttachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err)
}
// Persist state
if err := d.WriteState(); err != nil {
slog.Error("AttachVolume: failed to write state", "err", err)
}
d.respondWithVolumeAttachment(msg, volumeID, command.ID, guestDevice, "attached")
slog.Info("Volume attached successfully", "volumeId", volumeID, "instanceId", command.ID, "apiDevice", device, "guestDevice", guestDevice)
d.respondWithVolumeAttachment(msg, volumeID, command.ID, res.GuestDevice, "attached")
}
// handleDetachVolume performs a three-phase hot-unplug (reverse of attach):
//
// Phase 1: QMP device_del (remove guest device)
// Phase 2: QMP blockdev-del (remove block node)
// Phase 3: ebs.unmount NATS (stop NBD server)
// handleDetachVolume dispatches the QMP/state-machine pipeline to
// vm.Manager.DetachVolume and emits the AWS API response.
func (d *Daemon) handleDetachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
slog.Info("Detaching volume from instance", "instanceId", command.ID)
// Validate DetachVolumeData
if command.DetachVolumeData == nil || command.DetachVolumeData.VolumeID == "" {
slog.Error("DetachVolume: missing detach volume data")
respondWithError(msg, awserrors.ErrorInvalidParameterValue)
return
}
volumeID := command.DetachVolumeData.VolumeID
device := command.DetachVolumeData.Device
force := command.DetachVolumeData.Force
// Validate instance is running
var status vm.InstanceState
d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status })
if status != vm.StateRunning {
slog.Error("DetachVolume: instance not running", "instanceId", command.ID, "status", status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
deviceName, err := d.vmMgr.DetachVolume(
instance.ID,
command.DetachVolumeData.VolumeID,
command.DetachVolumeData.Device,
command.DetachVolumeData.Force,
)
if err != nil {
respondWithError(msg, attachDetachErrorCode(err))
return
}
// Find the volume in EBSRequests
instance.EBSRequests.Mu.Lock()
var ebsReq types.EBSRequest
found := false
for _, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
ebsReq = req
found = true
break
}
}
instance.EBSRequests.Mu.Unlock()
d.respondWithVolumeAttachment(msg, command.DetachVolumeData.VolumeID, command.ID, deviceName, "detaching")
}
if !found {
slog.Error("DetachVolume: volume not attached to instance", "volumeId", volumeID, "instanceId", command.ID)
respondWithError(msg, awserrors.ErrorIncorrectState)
return
}
// Reject detaching boot/EFI/CloudInit volumes
if ebsReq.Boot || ebsReq.EFI || ebsReq.CloudInit {
slog.Error("DetachVolume: cannot detach boot/EFI/CloudInit volume", "volumeId", volumeID)
respondWithError(msg, awserrors.ErrorOperationNotPermitted)
return
}
// Optional device cross-check
if device != "" && ebsReq.DeviceName != "" && device != ebsReq.DeviceName {
slog.Error("DetachVolume: device mismatch", "requested", device, "actual", ebsReq.DeviceName)
respondWithError(msg, awserrors.ErrorInvalidParameterValue)
return
}
deviceID := fmt.Sprintf("vdisk-%s", volumeID)
nodeName := fmt.Sprintf("nbd-%s", volumeID)
iothreadID := fmt.Sprintf("ioth-%s", volumeID)
// Phase 1: QMP device_del (remove guest device).
// Idempotent: a prior detach may have already removed the device but
// failed at blockdev-del, leaving the volume's external state intact.
// Treat DeviceNotFound as success so the retry can drive blockdev-del
// to completion without bailing on the now-absent guest device.
_, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "device_del",
Arguments: map[string]any{"id": deviceID},
}, instance.ID)
// attachDetachErrorCode maps a vm.Manager error returned by AttachVolume
// or DetachVolume to the AWS API error code that the SDK expects.
func attachDetachErrorCode(err error) string {
switch {
case err == nil:
case isQMPDeviceNotFound(err):
slog.Info("DetachVolume: guest device already removed (resuming detach)", "volumeId", volumeID, "err", err)
case force:
slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err)
case errors.Is(err, vm.ErrInstanceNotFound):
return awserrors.ErrorInvalidInstanceIDNotFound
case errors.Is(err, vm.ErrInvalidTransition):
return awserrors.ErrorIncorrectInstanceState
case errors.Is(err, vm.ErrAttachmentLimitExceeded):
return awserrors.ErrorAttachmentLimitExceeded
case errors.Is(err, vm.ErrVolumeNotAttached):
return awserrors.ErrorIncorrectState
case errors.Is(err, vm.ErrVolumeNotDetachable):
return awserrors.ErrorOperationNotPermitted
case errors.Is(err, vm.ErrVolumeDeviceMismatch):
return awserrors.ErrorInvalidParameterValue
default:
slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
return awserrors.ErrorServerInternal
}
// Brief pause for guest to acknowledge PCI removal
time.Sleep(d.detachDelay)
// Phase 2: QMP blockdev-del (remove block node).
// Poll-retry on "node is in use": after device_del, NBD client teardown
// and any in-flight I/O can hold the block node briefly. Bailing here
// without retry leaves the guest device gone but the block node intact,
// and the next AWS-CLI retry hits DeviceNotFound on device_del with no
// path forward (mulga-siv-41).
blockdevErr := d.tryBlockdevDel(instance, nodeName)
if blockdevErr != nil {
// Block node still referenced after retry budget exhausted; do not
// clean up state or unmount — tearing down the NBD server would
// crash the VM, and removing metadata would allow the volume to be
// double-attached.
slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", "volumeId", volumeID, "err", blockdevErr)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Phase 2b: QMP object-del (remove iothread, best-effort)
_, iothreadErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "object-del",
Arguments: map[string]any{"id": iothreadID},
}, instance.ID)
if iothreadErr != nil {
slog.Warn("DetachVolume: QMP object-del iothread failed (non-fatal)", "volumeId", volumeID, "err", iothreadErr)
}
// Phase 3: ebs.unmount via NATS (best-effort)
d.rollbackEBSMount(ebsReq)
// State cleanup: remove volume from EBSRequests (search by name to avoid stale index)
instance.EBSRequests.Mu.Lock()
for i, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests[:i], instance.EBSRequests.Requests[i+1:]...)
break
}
}
instance.EBSRequests.Mu.Unlock()
// Remove from BlockDeviceMappings
d.vmMgr.Inspect(instance, func(v *vm.VM) {
if v.Instance == nil {
return
}
filtered := make([]*ec2.InstanceBlockDeviceMapping, 0, len(v.Instance.BlockDeviceMappings))
for _, bdm := range v.Instance.BlockDeviceMappings {
if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil && *bdm.Ebs.VolumeId == volumeID {
continue
}
filtered = append(filtered, bdm)
}
v.Instance.BlockDeviceMappings = filtered
})
// Update volume metadata to "available"
if err := d.volumeService.UpdateVolumeState(volumeID, "available", "", ""); err != nil {
slog.Error("DetachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err)
}
// Persist state
if err := d.WriteState(); err != nil {
slog.Error("DetachVolume: failed to write state", "err", err)
}
d.respondWithVolumeAttachment(msg, volumeID, command.ID, ebsReq.DeviceName, "detaching")
slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", command.ID)
}
func (d *Daemon) handleEC2CreateVolume(msg *nats.Msg) {
@@ -515,62 +196,3 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) {
func (d *Daemon) handleEC2DeleteVolume(msg *nats.Msg) {
handleNATSRequest(msg, d.volumeService.DeleteVolume)
}
// tryBlockdevDel issues blockdev-del with bounded retry on "is in use"
// errors. After device_del, NBD client teardown and any in-flight guest
// I/O can briefly hold the block node; QEMU surfaces this as a
// GenericError carrying "Node <name> is in use". Polling at d.detachDelay
// gives the I/O drain enough time without a hardcoded long sleep.
func (d *Daemon) tryBlockdevDel(instance *vm.VM, nodeName string) error {
const maxAttempts = 20 // 20 × detachDelay (default 1s) = 20s budget
var lastErr error
for attempt := 1; attempt <= maxAttempts; attempt++ {
_, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-del",
Arguments: map[string]any{"node-name": nodeName},
}, instance.ID)
if err == nil {
if attempt > 1 {
slog.Info("DetachVolume: blockdev-del succeeded after retry",
"nodeName", nodeName, "attempts", attempt)
}
return nil
}
lastErr = err
if !isQMPNodeInUse(err) {
return err
}
if attempt == maxAttempts {
break
}
slog.Debug("DetachVolume: blockdev-del busy, retrying",
"nodeName", nodeName, "attempt", attempt, "err", err)
time.Sleep(d.detachDelay)
}
return lastErr
}
// isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class
// error from device_del. Used to make device_del idempotent across detach
// retries — the second AWS-CLI call must not bail when QEMU reports the
// device is already gone (mulga-siv-41).
func isQMPDeviceNotFound(err error) bool {
if err == nil {
return false
}
s := err.Error()
return strings.Contains(s, "DeviceNotFound")
}
// isQMPNodeInUse returns true when err is a QMP GenericError reporting a
// block node still in use. blockdev-del fails this way while NBD client
// teardown and queued I/O drain after device_del; the caller polls until
// the node is released or the budget expires. Match "in use" rather than
// "is in use" — QEMU phrases this as both "X is in use" and "X is still
// in use" depending on the path.
func isQMPNodeInUse(err error) bool {
if err == nil {
return false
}
return strings.Contains(err.Error(), "in use")
}
spinifex/daemon/daemon_handlers_volume_test.go
+88
View File
@@ -0,0 +1,88 @@
package daemon
import (
"errors"
"fmt"
"testing"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/stretchr/testify/assert"
)
// TestAttachDetachErrorCode locks down the manager-error → AWS-API-code
// mapping that handleAttachVolume and handleDetachVolume both call. Wrong
// mapping silently breaks AWS-SDK retry semantics: clients expect 4xx
// codes for caller-fixable problems and 5xx for server faults. A future
// edit that drops a sentinel branch would otherwise pass the mechanical
// "tests still compile" bar with no signal.
func TestAttachDetachErrorCode(t *testing.T) {
tests := []struct {
name string
err error
want string
}{
{
name: "ErrInstanceNotFound maps to InvalidInstanceID.NotFound",
err: vm.ErrInstanceNotFound,
want: awserrors.ErrorInvalidInstanceIDNotFound,
},
{
name: "wrapped ErrInstanceNotFound still matches via errors.Is",
err: fmt.Errorf("manager: %w", vm.ErrInstanceNotFound),
want: awserrors.ErrorInvalidInstanceIDNotFound,
},
{
name: "ErrInvalidTransition maps to IncorrectInstanceState",
err: vm.ErrInvalidTransition,
want: awserrors.ErrorIncorrectInstanceState,
},
{
name: "wrapped ErrInvalidTransition still matches",
err: fmt.Errorf("cannot attach in state stopped: %w", vm.ErrInvalidTransition),
want: awserrors.ErrorIncorrectInstanceState,
},
{
name: "ErrAttachmentLimitExceeded maps to AttachmentLimitExceeded",
err: vm.ErrAttachmentLimitExceeded,
want: awserrors.ErrorAttachmentLimitExceeded,
},
{
name: "ErrVolumeNotAttached maps to IncorrectState",
err: vm.ErrVolumeNotAttached,
want: awserrors.ErrorIncorrectState,
},
{
name: "wrapped ErrVolumeNotAttached still matches",
err: fmt.Errorf("%w: vol-1", vm.ErrVolumeNotAttached),
want: awserrors.ErrorIncorrectState,
},
{
name: "ErrVolumeNotDetachable maps to OperationNotPermitted",
err: vm.ErrVolumeNotDetachable,
want: awserrors.ErrorOperationNotPermitted,
},
{
name: "ErrVolumeDeviceMismatch maps to InvalidParameterValue",
err: vm.ErrVolumeDeviceMismatch,
want: awserrors.ErrorInvalidParameterValue,
},
{
name: "unknown error falls through to ServerInternal",
err: errors.New("QMP blockdev-add: connection refused"),
want: awserrors.ErrorServerInternal,
},
{
name: "wrapped unknown error falls through to ServerInternal",
err: fmt.Errorf("manager: %w", errors.New("nbdkit timeout")),
want: awserrors.ErrorServerInternal,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := attachDetachErrorCode(tt.err)
assert.Equal(t, tt.want, got)
})
}
}
spinifex/daemon/daemon_test.go
+36 -141
View File
@@ -104,6 +104,18 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, objectstore.NewMemoryObjectStore())
daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc)
// Wire the minimum vm.Deps that handler tests rely on. Lifecycle (Run/Start/
// Stop/Terminate) tests still set up their own deps; this gives the
// post-2d AttachVolume / DetachVolume manager methods enough plumbing to
// drive ebs.mount/unmount over NATS using the test's connection.
volState := newVolumeStateUpdaterAdapter(daemon.volumeService)
daemon.vmMgr.SetDeps(vm.Deps{
NodeID: daemon.node,
VolumeMounter: newVolumeMounterAdapter(daemon.natsConn, daemon.node, volState),
VolumeStateUpdater: volState,
DetachDelay: daemon.detachDelay,
})
t.Cleanup(func() {
if daemon.natsConn != nil {
drainAndClose(t, daemon.natsConn)
@@ -3136,16 +3148,18 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) {
}, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated")
}
// TestRollbackEBSMount_Success verifies that rollbackEBSMount sends an ebs.unmount
// request and handles a successful unmount response.
func TestRollbackEBSMount_Success(t *testing.T) {
// TestVolumeMounterAdapter_UnmountOne_Success verifies that the adapter's
// UnmountOne sends an ebs.unmount NATS request and handles a successful
// response. UnmountOne is the post-2d successor to Daemon.rollbackEBSMount;
// it is the AttachVolume rollback path and DetachVolume Phase 3.
func TestVolumeMounterAdapter_UnmountOne_Success(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
unmountCalled := make(chan string, 1)
// Mock ebs.unmount subscriber that returns success
sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) {
var req types.EBSRequest
json.Unmarshal(msg.Data, &req)
@@ -3157,12 +3171,10 @@ func TestRollbackEBSMount_Success(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
ebsReq := types.EBSRequest{
adapter.UnmountOne(types.EBSRequest{
Name: "vol-rollback-test",
DeviceName: "/dev/sdf",
}
daemon.rollbackEBSMount(ebsReq)
})
select {
case volName := <-unmountCalled:
@@ -3172,14 +3184,14 @@ func TestRollbackEBSMount_Success(t *testing.T) {
}
}
// TestRollbackEBSMount_UnmountError verifies that rollbackEBSMount handles
// an error response from ebs.unmount gracefully (no panic, just logs).
func TestRollbackEBSMount_UnmountError(t *testing.T) {
// TestVolumeMounterAdapter_UnmountOne_UnmountError verifies that UnmountOne
// tolerates an error response from ebs.unmount (logs only, no propagation).
func TestVolumeMounterAdapter_UnmountOne_UnmountError(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
// Mock ebs.unmount subscriber that returns an error
sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) {
resp := types.EBSUnMountResponse{Error: "unmount failed: device busy"}
data, _ := json.Marshal(resp)
@@ -3188,45 +3200,37 @@ func TestRollbackEBSMount_UnmountError(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
ebsReq := types.EBSRequest{Name: "vol-rollback-err"}
// Should not panic — errors are logged but not propagated
daemon.rollbackEBSMount(ebsReq)
adapter.UnmountOne(types.EBSRequest{Name: "vol-rollback-err"})
}
// TestRollbackEBSMount_StillMounted verifies that rollbackEBSMount handles
// the case where the unmount response says the volume is still mounted.
func TestRollbackEBSMount_StillMounted(t *testing.T) {
// TestVolumeMounterAdapter_UnmountOne_StillMounted verifies that UnmountOne
// tolerates an unmount response that reports the volume still mounted.
func TestVolumeMounterAdapter_UnmountOne_StillMounted(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) {
resp := types.EBSUnMountResponse{Mounted: true} // still mounted
resp := types.EBSUnMountResponse{Mounted: true}
data, _ := json.Marshal(resp)
msg.Respond(data)
})
require.NoError(t, err)
defer sub.Unsubscribe()
ebsReq := types.EBSRequest{Name: "vol-still-mounted"}
// Should not panic
daemon.rollbackEBSMount(ebsReq)
adapter.UnmountOne(types.EBSRequest{Name: "vol-still-mounted"})
}
// TestRollbackEBSMount_NATSTimeout verifies that rollbackEBSMount handles
// NATS request timeout gracefully (no subscriber on ebs.unmount).
func TestRollbackEBSMount_NATSTimeout(t *testing.T) {
// TestVolumeMounterAdapter_UnmountOne_NATSTimeout verifies that UnmountOne
// tolerates NATS request timeout (no subscriber on ebs.unmount).
func TestVolumeMounterAdapter_UnmountOne_NATSTimeout(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
// No ebs.unmount subscriber — will timeout
ebsReq := types.EBSRequest{Name: "vol-timeout"}
// Should not panic, just log the timeout
daemon.rollbackEBSMount(ebsReq)
adapter.UnmountOne(types.EBSRequest{Name: "vol-timeout"})
}
// TestDescribeInstances_InvalidInstanceIDMalformed verifies that DescribeInstances
@@ -3328,115 +3332,6 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
})
}
// TestNextAvailableDevice tests the device name auto-assignment logic
func TestNextAvailableDevice(t *testing.T) {
tests := []struct {
name string
instance *vm.VM
wantDevice string
}{
{
name: "empty instance returns first device",
instance: &vm.VM{
Instance: &ec2.Instance{},
},
wantDevice: "/dev/sdf",
},
{
name: "nil Instance returns first device",
instance: &vm.VM{},
wantDevice: "/dev/sdf",
},
{
name: "existing BlockDeviceMappings skipped",
instance: &vm.VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: aws.String("/dev/sdf")},
{DeviceName: aws.String("/dev/sdg")},
},
},
},
wantDevice: "/dev/sdh",
},
{
name: "existing EBSRequests skipped",
instance: &vm.VM{
Instance: &ec2.Instance{},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{Name: "vol-1", DeviceName: "/dev/sdf"},
},
},
},
wantDevice: "/dev/sdg",
},
{
name: "mixed sources all skipped",
instance: &vm.VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: aws.String("/dev/sdf")},
},
},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{Name: "vol-1", DeviceName: "/dev/sdg"},
},
},
},
wantDevice: "/dev/sdh",
},
{
name: "all devices f-p used returns empty",
instance: func() *vm.VM {
var bdms []*ec2.InstanceBlockDeviceMapping
for c := 'f'; c <= 'p'; c++ {
dev := fmt.Sprintf("/dev/sd%c", c)
bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{
DeviceName: aws.String(dev),
})
}
return &vm.VM{
Instance: &ec2.Instance{BlockDeviceMappings: bdms},
}
}(),
wantDevice: "",
},
{
name: "nil DeviceName in BlockDeviceMappings ignored",
instance: &vm.VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: nil},
{DeviceName: aws.String("/dev/sdf")},
},
},
},
wantDevice: "/dev/sdg",
},
{
name: "empty DeviceName in EBSRequests ignored",
instance: &vm.VM{
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{DeviceName: ""},
{DeviceName: "/dev/sdf"},
},
},
},
wantDevice: "/dev/sdg",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := nextAvailableDevice(tt.instance)
assert.Equal(t, tt.wantDevice, got)
})
}
}
func TestCanAllocate(t *testing.T) {
makeInstanceType := func(vCPUs int64, memoryMiB int64) *ec2.InstanceTypeInfo {
return &ec2.InstanceTypeInfo{
spinifex/daemon/vm_adapters.go
+63
View File
@@ -159,6 +159,68 @@ func (a *volumeMounterAdapter) Unmount(instance *vm.VM) error {
return nil
}
// MountOne sends ebs.mount for a single request and writes the resolved
// NBDURI back into req.NBDURI. Used by hot-attach (Manager.AttachVolume).
func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error {
payload, err := json.Marshal(req)
if err != nil {
return fmt.Errorf("marshal ebs.mount request: %w", err)
}
reply, err := a.nc.Request(a.topic("mount"), payload, 30*time.Second)
if err != nil {
return fmt.Errorf("ebs.mount NATS request: %w", err)
}
var resp types.EBSMountResponse
if err := json.Unmarshal(reply.Data, &resp); err != nil {
return fmt.Errorf("unmarshal ebs.mount response: %w", err)
}
if resp.Error != "" {
return fmt.Errorf("ebs.mount returned error: %s", resp.Error)
}
if resp.URI == "" {
return vm.ErrMountAmbiguous
}
req.NBDURI = resp.URI
return nil
}
// UnmountOne sends ebs.unmount for a single request. Best-effort: errors
// are logged. Mirrors the pre-2d Daemon.rollbackEBSMount semantics so
// AttachVolume rollback and DetachVolume Phase 3 share one code path.
func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) {
payload, err := json.Marshal(req)
if err != nil {
slog.Error("UnmountOne: failed to marshal unmount request",
"volume", req.Name, "err", err)
return
}
msg, err := a.nc.Request(a.topic("unmount"), payload, 10*time.Second)
if err != nil {
slog.Error("UnmountOne: ebs.unmount NATS request failed",
"volume", req.Name, "err", err)
return
}
var resp types.EBSUnMountResponse
if err := json.Unmarshal(msg.Data, &resp); err != nil {
slog.Error("UnmountOne: failed to unmarshal response",
"volume", req.Name, "err", err)
return
}
if resp.Error != "" {
slog.Error("UnmountOne: ebs.unmount returned error",
"volume", req.Name, "err", resp.Error)
return
}
if resp.Mounted {
slog.Error("UnmountOne: volume still mounted after unmount", "volume", req.Name)
return
}
slog.Info("UnmountOne: volume unmounted successfully", "volume", req.Name)
}
// qmpClientFactoryAdapter satisfies vm.QMPClientFactory. It performs only
// the connect + qmp_capabilities handshake; the manager owns starting the
// heartbeat goroutine on the returned client.
@@ -409,6 +471,7 @@ func (d *Daemon) buildVMManagerDeps() vm.Deps {
TransitionState: d.TransitionState,
DevNetworking: d.config.Daemon.DevNetworking,
BindHost: d.config.Host,
DetachDelay: d.detachDelay,
}
}
spinifex/daemon/device_map.go → spinifex/vm/device_map.go
+51 -19
View File
@@ -1,4 +1,4 @@
package daemon
package vm
import (
"encoding/json"
@@ -11,7 +11,6 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/vm"
)
// pciAddrRegexp extracts the device index from a boot-time QDev path.
@@ -33,8 +32,8 @@ var hotplugPortRegexp = regexp.MustCompile(`hotplug(\d+)`)
// The mapping is derived from PCI address order: virtio-blk-pci devices are
// enumerated by the guest kernel in PCI bus order, which corresponds to the
// device index in the QDev path.
func queryGuestDeviceMap(d *Daemon, qmpClient *qmp.QMPClient, instanceID string) (map[string]string, error) {
resp, err := d.SendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID)
func queryGuestDeviceMap(qmpClient *qmp.QMPClient, instanceID string) (map[string]string, error) {
resp, err := sendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID)
if err != nil {
return nil, fmt.Errorf("query-block failed: %w", err)
}
@@ -50,12 +49,12 @@ func queryGuestDeviceMap(d *Daemon, qmpClient *qmp.QMPClient, instanceID string)
// queryGuestDeviceMapWait retries queryGuestDeviceMap until expectedDevice
// appears in the result. This handles the race where query-block is called
// immediately after device_add, before QEMU has registered the new device.
func queryGuestDeviceMapWait(d *Daemon, qmpClient *qmp.QMPClient, instanceID, expectedDevice string) (map[string]string, error) {
func queryGuestDeviceMapWait(qmpClient *qmp.QMPClient, instanceID, expectedDevice string) (map[string]string, error) {
const maxAttempts = 5
const retryDelay = 200 * time.Millisecond
for attempt := range maxAttempts {
deviceMap, err := queryGuestDeviceMap(d, qmpClient, instanceID)
deviceMap, err := queryGuestDeviceMap(qmpClient, instanceID)
if err != nil {
return nil, err
}
@@ -70,7 +69,7 @@ func queryGuestDeviceMapWait(d *Daemon, qmpClient *qmp.QMPClient, instanceID, ex
}
// Return the last result even though the device wasn't found — caller decides fallback.
return queryGuestDeviceMap(d, qmpClient, instanceID)
return queryGuestDeviceMap(qmpClient, instanceID)
}
// buildDeviceMap takes a list of BlockDevices from QMP and returns a map
@@ -140,16 +139,18 @@ func buildDeviceMap(devices []qmp.BlockDevice) map[string]string {
return result
}
// updateGuestDeviceNames queries the running VM's QMP to discover actual guest device
// paths and updates the instance's BlockDeviceMappings accordingly.
func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) {
// UpdateGuestDeviceNames queries the running VM's QMP to discover actual
// guest device paths and updates the instance's BlockDeviceMappings
// accordingly. Persists running state on success. No-op when QMPClient is
// nil. Errors are logged; callers do not need to react.
func (m *Manager) UpdateGuestDeviceNames(instance *VM) {
if instance.QMPClient == nil {
slog.Warn("updateGuestDeviceNames: QMPClient is nil, cannot discover guest device names",
slog.Warn("UpdateGuestDeviceNames: QMPClient is nil, cannot discover guest device names",
"instanceId", instance.ID)
return
}
deviceMap, err := queryGuestDeviceMap(d, instance.QMPClient, instance.ID)
deviceMap, err := queryGuestDeviceMap(instance.QMPClient, instance.ID)
if err != nil {
slog.Warn("Failed to query guest device map, BlockDeviceMappings will use API names",
"instanceId", instance.ID, "err", err)
@@ -157,8 +158,8 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) {
}
// Build volume ID → guest device path mapping from EBSRequests.
// Collect under EBSRequests lock, then release before acquiring Instances lock
// to maintain consistent lock ordering.
// Collect under EBSRequests lock, then release before acquiring the
// manager lock to maintain consistent lock ordering.
instance.EBSRequests.Mu.Lock()
volToGuest := make(map[string]string, len(instance.EBSRequests.Requests))
for _, req := range instance.EBSRequests.Requests {
@@ -176,8 +177,7 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) {
}
instance.EBSRequests.Mu.Unlock()
// Update BlockDeviceMappings using the mapping
d.vmMgr.Inspect(instance, func(v *vm.VM) {
m.Inspect(instance, func(v *VM) {
if v.Instance == nil {
return
}
@@ -191,8 +191,9 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) {
}
})
if err := d.WriteState(); err != nil {
slog.Error("Failed to persist state after guest device name update", "instanceId", instance.ID, "err", err)
if err := m.writeRunningState(); err != nil {
slog.Error("Failed to persist state after guest device name update",
"instanceId", instance.ID, "err", err)
}
slog.Info("Updated guest device names", "instanceId", instance.ID, "deviceMap", deviceMap)
@@ -215,7 +216,6 @@ func extractPCIIndex(qdev string) int {
// extractPeripheralName extracts the device ID from a hot-plugged QDev path.
// For example: "/machine/peripheral/vdisk-vol-abc123/virtio-backend" → "vdisk-vol-abc123"
func extractPeripheralName(qdev string) string {
// Path format: /machine/peripheral/<name>/virtio-backend
const prefix = "/machine/peripheral/"
if !strings.HasPrefix(qdev, prefix) {
return ""
@@ -241,3 +241,35 @@ func extractHotplugPort(qdev string) int {
}
return idx
}
// nextAvailableDevice finds the next available /dev/sd[f-p] device name
// for instance. AWS convention reserves f-p for attached volumes. Returns
// an empty string when every slot is occupied.
func nextAvailableDevice(instance *VM) string {
usedDevices := make(map[string]bool)
if instance.Instance != nil {
for _, bdm := range instance.Instance.BlockDeviceMappings {
if bdm.DeviceName != nil {
usedDevices[*bdm.DeviceName] = true
}
}
}
instance.EBSRequests.Mu.Lock()
for _, req := range instance.EBSRequests.Requests {
if req.DeviceName != "" {
usedDevices[req.DeviceName] = true
}
}
instance.EBSRequests.Mu.Unlock()
for c := 'f'; c <= 'p'; c++ {
dev := fmt.Sprintf("/dev/sd%c", c)
if !usedDevices[dev] {
return dev
}
}
return ""
}
spinifex/daemon/device_map_test.go → spinifex/vm/device_map_test.go
+1 -1
View File
@@ -1,4 +1,4 @@
package daemon
package vm
import (
"testing"
spinifex/vm/errors.go
+23
View File
@@ -10,3 +10,26 @@ var ErrInstanceNotFound = errors.New("instance not found")
// because the instance's current state does not permit the target
// transition (e.g. Stop on an already-stopped instance).
var ErrInvalidTransition = errors.New("invalid state transition")
// ErrAttachmentLimitExceeded is returned by AttachVolume when the
// instance has no free /dev/sd[f-p] slot to assign.
var ErrAttachmentLimitExceeded = errors.New("attachment limit exceeded")
// ErrVolumeNotAttached is returned by DetachVolume when the supplied
// volumeID is not present in the instance's EBSRequests.
var ErrVolumeNotAttached = errors.New("volume not attached to instance")
// ErrVolumeNotDetachable is returned by DetachVolume when the target
// volume is a boot, EFI, or cloud-init volume that cannot be hot-unplugged.
var ErrVolumeNotDetachable = errors.New("volume cannot be detached")
// ErrVolumeDeviceMismatch is returned by DetachVolume when the caller
// supplies a device name that does not match the recorded attachment.
var ErrVolumeDeviceMismatch = errors.New("volume device name mismatch")
// ErrMountAmbiguous is returned by VolumeMounter.MountOne when the
// backend acknowledged the mount but returned an empty URI. The mount may
// or may not have started serving NBD; AttachVolume must invoke UnmountOne
// defensively so a half-started backend mount is not orphaned. Mirrors the
// pre-2d daemon-side rollback that triggered on this exact response shape.
var ErrMountAmbiguous = errors.New("ebs.mount succeeded with empty URI")
spinifex/vm/lifecycle.go
+21
View File
@@ -38,6 +38,27 @@ func (m *Manager) Start(id string) error {
return m.launch(instance)
}
// Reboot issues a QMP system_reset to a running instance. The VM stays
// in StateRunning across the reset; QEMU re-runs firmware and the guest
// kernel reboots in place. Returns ErrInstanceNotFound when id is unknown
// and ErrInvalidTransition when the instance is not Running.
func (m *Manager) Reboot(id string) error {
instance, ok := m.Get(id)
if !ok {
return ErrInstanceNotFound
}
var status InstanceState
m.Inspect(instance, func(v *VM) { status = v.Status })
if status != StateRunning {
return fmt.Errorf("%w: cannot reboot instance %s in state %s",
ErrInvalidTransition, id, status)
}
if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_reset"}, id); err != nil {
return fmt.Errorf("QMP system_reset: %w", err)
}
return nil
}
// launchStillValid returns true while the launch pipeline may continue
// setting up resources for instance. Returns false if a concurrent terminate
// has flipped status out of pending/stopped/provisioning — at that point the
spinifex/vm/manager.go
+6
View File
@@ -3,6 +3,7 @@ package vm
import (
"maps"
"sync"
"time"
)
// ManagerHooks are callbacks the manager fires synchronously on every
@@ -48,6 +49,11 @@ type Deps struct {
// BindHost is the daemon's listen IP, used when wiring user-mode
// hostfwd rules. Empty / "0.0.0.0" falls back to 127.0.0.1.
BindHost string
// DetachDelay is the pause between QMP device_del and blockdev-del
// during DetachVolume, and the polling interval for blockdev-del retry.
// Zero is acceptable in tests; production uses 1s.
DetachDelay time.Duration
}
// Manager owns the in-memory map of running VMs on this node and every
spinifex/vm/manager_deps_test.go
+22 -2
View File
@@ -4,6 +4,8 @@ import (
"errors"
"maps"
"testing"
"github.com/mulgadc/spinifex/spinifex/types"
)
// fakeStateStore is a minimal in-memory StateStore used to verify Deps wiring.
@@ -135,8 +137,11 @@ func TestManagerHooks_InitiallyNil(t *testing.T) {
// fakeVolumeMounter records every call so lifecycle tests can assert ordering.
type fakeVolumeMounter struct {
mounted, unmounted []string
mountErr error
mounted, unmounted []string
mountedOne, unmountedOne []string
mountErr error
mountOneErr error
mountOneURI string
}
func (f *fakeVolumeMounter) Mount(v *VM) error {
@@ -149,6 +154,21 @@ func (f *fakeVolumeMounter) Unmount(v *VM) error {
return nil
}
func (f *fakeVolumeMounter) MountOne(req *types.EBSRequest) error {
f.mountedOne = append(f.mountedOne, req.Name)
if f.mountOneErr != nil {
return f.mountOneErr
}
if f.mountOneURI != "" {
req.NBDURI = f.mountOneURI
}
return nil
}
func (f *fakeVolumeMounter) UnmountOne(req types.EBSRequest) {
f.unmountedOne = append(f.unmountedOne, req.Name)
}
var _ VolumeMounter = (*fakeVolumeMounter)(nil)
func TestDeps_VolumeMounterSatisfiesInterface(t *testing.T) {
spinifex/vm/volume_mounter.go
+9
View File
@@ -1,5 +1,7 @@
package vm
import "github.com/mulgadc/spinifex/spinifex/types"
// VolumeMounter mounts and unmounts the EBS volumes attached to a VM. The
// real implementation routes ebs.mount / ebs.unmount NATS requests; the
// abstraction keeps NATS out of the manager.
@@ -10,4 +12,11 @@ type VolumeMounter interface {
// Unmount sends ebs.unmount for each attached volume. Errors are logged
// per volume and aggregated; partial failure is tolerated.
Unmount(v *VM) error
// MountOne sends ebs.mount for a single request and writes the resolved
// NBDURI back into req.NBDURI on success. Used by hot-attach.
MountOne(req *types.EBSRequest) error
// UnmountOne sends ebs.unmount for a single request. Best-effort: errors
// are logged and tolerated.
UnmountOne(req types.EBSRequest)
}
spinifex/vm/volumes.go
+416
View File
@@ -0,0 +1,416 @@
package vm
import (
"errors"
"fmt"
"log/slog"
"strconv"
"strings"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
)
// blockdevDelMaxAttempts caps the bounded retry on "node is in use" while
// QEMU drains pending I/O after device_del. 20 × DetachDelay (default 1s)
// gives a 20s budget that matches the pre-2d daemon-side helper.
const blockdevDelMaxAttempts = 20
// AttachVolumeResult carries the device names produced by AttachVolume:
// the AWS-API name (/dev/sdf, possibly auto-allocated) and the discovered
// guest device path (/dev/vdb…). Daemon handlers use GuestDevice in the
// AWS API response; the discovered name eventually appears in
// DescribeInstances.
type AttachVolumeResult struct {
DeviceName string
GuestDevice string
}
// AttachVolume hot-plugs a volume into a running instance via the
// three-phase QMP pipeline (mount → blockdev-add → device_add). On
// failure mid-pipeline, partial state is rolled back. The instance must
// be in StateRunning. If device is empty, the next free /dev/sd[f-p]
// slot is allocated.
//
// Daemon handlers retain volume-side validation (existence, ownership,
// availability, AZ) and emit the AWS API response; the manager owns
// every QMP and persistence side-effect.
func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, error) {
instance, ok := m.Get(id)
if !ok {
return AttachVolumeResult{}, ErrInstanceNotFound
}
var status InstanceState
m.Inspect(instance, func(v *VM) { status = v.Status })
if status != StateRunning {
return AttachVolumeResult{}, fmt.Errorf("%w: cannot attach to instance %s in state %s",
ErrInvalidTransition, id, status)
}
if device == "" {
m.Inspect(instance, func(v *VM) { device = nextAvailableDevice(v) })
if device == "" {
return AttachVolumeResult{}, ErrAttachmentLimitExceeded
}
}
ebsRequest := types.EBSRequest{
Name: volumeID,
DeviceName: device,
}
if m.deps.VolumeMounter == nil {
return AttachVolumeResult{}, fmt.Errorf("VolumeMounter not wired")
}
if err := m.deps.VolumeMounter.MountOne(&ebsRequest); err != nil {
slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err)
// Empty-URI response leaves backend NBD state ambiguous; unmount
// defensively to avoid orphaning a half-started mount.
if errors.Is(err, ErrMountAmbiguous) {
m.deps.VolumeMounter.UnmountOne(ebsRequest)
}
return AttachVolumeResult{}, fmt.Errorf("mount volume %s: %w", volumeID, err)
}
serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(ebsRequest.NBDURI)
if err != nil {
slog.Error("AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err)
m.deps.VolumeMounter.UnmountOne(ebsRequest)
return AttachVolumeResult{}, fmt.Errorf("parse NBDURI: %w", err)
}
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)
if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "object-add",
Arguments: map[string]any{
"qom-type": "iothread",
"id": iothreadID,
},
}, instance.ID); err != nil {
slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err)
m.deps.VolumeMounter.UnmountOne(ebsRequest)
return AttachVolumeResult{}, fmt.Errorf("QMP object-add iothread: %w", err)
}
if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-add",
Arguments: map[string]any{
"node-name": nodeName,
"driver": "nbd",
"server": serverArg,
"export": "",
"read-only": false,
},
}, instance.ID); err != nil {
slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err)
m.deps.VolumeMounter.UnmountOne(ebsRequest)
return AttachVolumeResult{}, fmt.Errorf("QMP blockdev-add: %w", err)
}
// /dev/sdf -> hotplug1, /dev/sdg -> hotplug2, etc.
hotplugBus := ""
if len(device) > 0 {
letter := device[len(device)-1]
if letter >= 'f' && letter <= 'p' {
hotplugBus = fmt.Sprintf("hotplug%d", letter-'f'+1)
}
}
deviceAddArgs := map[string]any{
"driver": "virtio-blk-pci",
"id": deviceID,
"drive": nodeName,
"iothread": iothreadID,
}
if hotplugBus != "" {
deviceAddArgs["bus"] = hotplugBus
}
if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "device_add",
Arguments: deviceAddArgs,
}, instance.ID); err != nil {
slog.Error("AttachVolume: QMP device_add failed, rolling back blockdev",
"volumeId", volumeID, "err", err)
if _, delErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-del",
Arguments: map[string]any{"node-name": nodeName},
}, instance.ID); delErr != nil {
slog.Error("AttachVolume: rollback blockdev-del failed, skipping EBS unmount",
"volumeId", volumeID, "err", delErr)
} else {
m.deps.VolumeMounter.UnmountOne(ebsRequest)
}
return AttachVolumeResult{}, fmt.Errorf("QMP device_add: %w", err)
}
// Discover guest device. query-block may not include the device
// immediately after device_add; queryGuestDeviceMapWait retries.
guestDevice := device // fallback to AWS API name
deviceMap, qmpErr := queryGuestDeviceMapWait(instance.QMPClient, instance.ID, deviceID)
if qmpErr != nil {
slog.Warn("AttachVolume: failed to query guest device map, using API device name",
"volumeId", volumeID, "err", qmpErr)
} else if gd, ok := deviceMap[deviceID]; ok {
guestDevice = gd
slog.Info("AttachVolume: discovered guest device",
"volumeId", volumeID, "qemuDevice", deviceID, "guestDevice", guestDevice)
} else {
slog.Error("AttachVolume: device not found in QMP device map after retries, using API device name",
"volumeId", volumeID, "qemuDevice", deviceID, "deviceMap", deviceMap)
}
// Replace stale entry for volumeID (covers stop/start cycles that keep
// the original request) or append a new one.
instance.EBSRequests.Mu.Lock()
replaced := false
for i, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
instance.EBSRequests.Requests[i] = ebsRequest
replaced = true
break
}
}
if !replaced {
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest)
}
instance.EBSRequests.Mu.Unlock()
m.Inspect(instance, func(v *VM) {
if v.Instance == nil {
return
}
now := time.Now()
mapping := &ec2.InstanceBlockDeviceMapping{}
mapping.SetDeviceName(guestDevice)
mapping.Ebs = &ec2.EbsInstanceBlockDevice{}
mapping.Ebs.SetVolumeId(volumeID)
mapping.Ebs.SetAttachTime(now)
mapping.Ebs.SetDeleteOnTermination(false)
mapping.Ebs.SetStatus("attached")
v.Instance.BlockDeviceMappings = append(v.Instance.BlockDeviceMappings, mapping)
})
if m.deps.VolumeStateUpdater != nil {
if err := m.deps.VolumeStateUpdater.UpdateVolumeState(volumeID, "in-use", instance.ID, guestDevice); err != nil {
slog.Error("AttachVolume: failed to update volume metadata",
"volumeId", volumeID, "err", err)
}
}
if err := m.writeRunningState(); err != nil {
slog.Error("AttachVolume: failed to write state", "err", err)
}
slog.Info("Volume attached successfully",
"volumeId", volumeID, "instanceId", instance.ID,
"apiDevice", device, "guestDevice", guestDevice)
return AttachVolumeResult{DeviceName: device, GuestDevice: guestDevice}, nil
}
// DetachVolume hot-unplugs a volume from a running instance via the
// three-phase QMP pipeline (device_del → blockdev-del → object-del →
// ebs.unmount). Returns the recorded AWS-API device name on success so
// the daemon handler can echo it in the response.
//
// device may be empty (no cross-check) or must equal the recorded
// DeviceName. force=true tolerates a failing device_del; blockdev-del
// is always required because tearing down the NBD server underneath a
// live block node would crash the VM.
func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, error) {
instance, ok := m.Get(id)
if !ok {
return "", ErrInstanceNotFound
}
var status InstanceState
m.Inspect(instance, func(v *VM) { status = v.Status })
if status != StateRunning {
return "", fmt.Errorf("%w: cannot detach from instance %s in state %s",
ErrInvalidTransition, id, status)
}
instance.EBSRequests.Mu.Lock()
var ebsReq types.EBSRequest
found := false
for _, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
ebsReq = req
found = true
break
}
}
instance.EBSRequests.Mu.Unlock()
if !found {
return "", fmt.Errorf("%w: %s", ErrVolumeNotAttached, volumeID)
}
if ebsReq.Boot || ebsReq.EFI || ebsReq.CloudInit {
return "", fmt.Errorf("%w: %s", ErrVolumeNotDetachable, volumeID)
}
if device != "" && ebsReq.DeviceName != "" && device != ebsReq.DeviceName {
return "", fmt.Errorf("%w: requested %s, recorded %s",
ErrVolumeDeviceMismatch, device, ebsReq.DeviceName)
}
deviceID := fmt.Sprintf("vdisk-%s", volumeID)
nodeName := fmt.Sprintf("nbd-%s", volumeID)
iothreadID := fmt.Sprintf("ioth-%s", volumeID)
// Phase 1: device_del. Idempotent on DeviceNotFound so the second
// AWS-CLI retry can drive blockdev-del to completion when a prior
// detach left the guest device gone but the block node intact
// (mulga-siv-41).
_, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "device_del",
Arguments: map[string]any{"id": deviceID},
}, instance.ID)
switch {
case err == nil:
case isQMPDeviceNotFound(err):
slog.Info("DetachVolume: guest device already removed (resuming detach)",
"volumeId", volumeID, "err", err)
case force:
slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)",
"volumeId", volumeID, "err", err)
default:
slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err)
return "", fmt.Errorf("QMP device_del: %w", err)
}
if m.deps.DetachDelay > 0 {
time.Sleep(m.deps.DetachDelay)
}
// Phase 2: blockdev-del with bounded retry on "node is in use".
if blockdevErr := m.tryBlockdevDel(instance, nodeName); blockdevErr != nil {
slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact",
"volumeId", volumeID, "err", blockdevErr)
return "", fmt.Errorf("QMP blockdev-del: %w", blockdevErr)
}
// Phase 2b: object-del (best-effort).
if _, iothreadErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "object-del",
Arguments: map[string]any{"id": iothreadID},
}, instance.ID); iothreadErr != nil {
slog.Warn("DetachVolume: QMP object-del iothread failed (non-fatal)",
"volumeId", volumeID, "err", iothreadErr)
}
// Phase 3: ebs.unmount (best-effort).
if m.deps.VolumeMounter != nil {
m.deps.VolumeMounter.UnmountOne(ebsReq)
}
// State cleanup.
instance.EBSRequests.Mu.Lock()
for i, req := range instance.EBSRequests.Requests {
if req.Name == volumeID {
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests[:i], instance.EBSRequests.Requests[i+1:]...)
break
}
}
instance.EBSRequests.Mu.Unlock()
m.Inspect(instance, func(v *VM) {
if v.Instance == nil {
return
}
filtered := make([]*ec2.InstanceBlockDeviceMapping, 0, len(v.Instance.BlockDeviceMappings))
for _, bdm := range v.Instance.BlockDeviceMappings {
if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil && *bdm.Ebs.VolumeId == volumeID {
continue
}
filtered = append(filtered, bdm)
}
v.Instance.BlockDeviceMappings = filtered
})
if m.deps.VolumeStateUpdater != nil {
if err := m.deps.VolumeStateUpdater.UpdateVolumeState(volumeID, "available", "", ""); err != nil {
slog.Error("DetachVolume: failed to update volume metadata",
"volumeId", volumeID, "err", err)
}
}
if err := m.writeRunningState(); err != nil {
slog.Error("DetachVolume: failed to write state", "err", err)
}
slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", instance.ID)
return ebsReq.DeviceName, nil
}
// tryBlockdevDel issues blockdev-del with bounded retry on "is in use"
// errors. After device_del, NBD client teardown and any in-flight guest
// I/O can briefly hold the block node; QEMU surfaces this as a
// GenericError carrying "Node <name> is in use". Polling at DetachDelay
// gives the I/O drain enough time without a hardcoded long sleep.
func (m *Manager) tryBlockdevDel(instance *VM, nodeName string) error {
var lastErr error
for attempt := 1; attempt <= blockdevDelMaxAttempts; attempt++ {
_, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-del",
Arguments: map[string]any{"node-name": nodeName},
}, instance.ID)
if err == nil {
if attempt > 1 {
slog.Info("DetachVolume: blockdev-del succeeded after retry",
"nodeName", nodeName, "attempts", attempt)
}
return nil
}
lastErr = err
if !isQMPNodeInUse(err) {
return err
}
if attempt == blockdevDelMaxAttempts {
break
}
slog.Debug("DetachVolume: blockdev-del busy, retrying",
"nodeName", nodeName, "attempt", attempt, "err", err)
if m.deps.DetachDelay > 0 {
time.Sleep(m.deps.DetachDelay)
}
}
return lastErr
}
// isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class
// error from device_del. Used to make device_del idempotent across detach
// retries (mulga-siv-41).
func isQMPDeviceNotFound(err error) bool {
if err == nil {
return false
}
return strings.Contains(err.Error(), "DeviceNotFound")
}
// isQMPNodeInUse returns true when err is a QMP GenericError reporting a
// block node still in use. blockdev-del fails this way while NBD client
// teardown and queued I/O drain after device_del. QEMU phrases this as
// both "X is in use" and "X is still in use" — match "in use".
func isQMPNodeInUse(err error) bool {
if err == nil {
return false
}
return strings.Contains(err.Error(), "in use")
}
spinifex/vm/volumes_test.go
+564
View File
@@ -0,0 +1,564 @@
package vm
import (
"encoding/json"
"errors"
"fmt"
"net"
"sync"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newMockQMPClient creates a QMPClient backed by an in-memory pipe. The
// responder is invoked for every command and returns the JSON object the
// server should reply with (e.g. {"return": {}} or {"error": {...}}). A
// nil responder reply defaults to an empty success.
func newMockQMPClient(t *testing.T, responder func(qmp.QMPCommand) map[string]any) (*qmp.QMPClient, func()) {
t.Helper()
clientConn, serverConn := net.Pipe()
client := &qmp.QMPClient{
Conn: clientConn,
Decoder: json.NewDecoder(clientConn),
Encoder: json.NewEncoder(clientConn),
}
done := make(chan struct{})
go func() {
defer close(done)
dec := json.NewDecoder(serverConn)
enc := json.NewEncoder(serverConn)
for {
var cmd qmp.QMPCommand
if err := dec.Decode(&cmd); err != nil {
return
}
var resp map[string]any
if responder != nil {
resp = responder(cmd)
}
if resp == nil {
resp = map[string]any{"return": map[string]any{}}
}
if err := enc.Encode(resp); err != nil {
return
}
}
}()
cancel := func() {
_ = clientConn.Close()
_ = serverConn.Close()
<-done
}
return client, cancel
}
// qmpRecorder records the order of QMP commands for assertion in rollback
// tests. Safe for concurrent use because sendQMPCommand serializes via the
// QMPClient mutex, but we lock anyway for clarity.
type qmpRecorder struct {
mu sync.Mutex
commands []qmp.QMPCommand
}
func (r *qmpRecorder) record(cmd qmp.QMPCommand) {
r.mu.Lock()
defer r.mu.Unlock()
r.commands = append(r.commands, cmd)
}
func (r *qmpRecorder) executes() []string {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]string, len(r.commands))
for i, c := range r.commands {
out[i] = c.Execute
}
return out
}
func TestNextAvailableDevice(t *testing.T) {
tests := []struct {
name string
instance *VM
wantDevice string
}{
{
name: "empty instance returns first device",
instance: &VM{
Instance: &ec2.Instance{},
},
wantDevice: "/dev/sdf",
},
{
name: "nil Instance returns first device",
instance: &VM{},
wantDevice: "/dev/sdf",
},
{
name: "existing BlockDeviceMappings skipped",
instance: &VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: aws.String("/dev/sdf")},
{DeviceName: aws.String("/dev/sdg")},
},
},
},
wantDevice: "/dev/sdh",
},
{
name: "existing EBSRequests skipped",
instance: &VM{
Instance: &ec2.Instance{},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{Name: "vol-1", DeviceName: "/dev/sdf"},
},
},
},
wantDevice: "/dev/sdg",
},
{
name: "mixed sources all skipped",
instance: &VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: aws.String("/dev/sdf")},
},
},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{Name: "vol-1", DeviceName: "/dev/sdg"},
},
},
},
wantDevice: "/dev/sdh",
},
{
name: "all devices f-p used returns empty",
instance: func() *VM {
var bdms []*ec2.InstanceBlockDeviceMapping
for c := 'f'; c <= 'p'; c++ {
dev := fmt.Sprintf("/dev/sd%c", c)
bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{
DeviceName: aws.String(dev),
})
}
return &VM{
Instance: &ec2.Instance{BlockDeviceMappings: bdms},
}
}(),
wantDevice: "",
},
{
name: "nil DeviceName in BlockDeviceMappings ignored",
instance: &VM{
Instance: &ec2.Instance{
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
{DeviceName: nil},
{DeviceName: aws.String("/dev/sdf")},
},
},
},
wantDevice: "/dev/sdg",
},
{
name: "empty DeviceName in EBSRequests ignored",
instance: &VM{
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{
{DeviceName: ""},
{DeviceName: "/dev/sdf"},
},
},
},
wantDevice: "/dev/sdg",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := nextAvailableDevice(tt.instance)
assert.Equal(t, tt.wantDevice, got)
})
}
}
func TestIsQMPDeviceNotFound(t *testing.T) {
tests := []struct {
name string
err error
want bool
}{
{name: "nil", err: nil, want: false},
{name: "matches DeviceNotFound", err: errors.New("QMP error: DeviceNotFound: device 'vdisk-x' not found"), want: true},
{name: "other QMP error", err: errors.New("QMP error: GenericError: bad command"), want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, isQMPDeviceNotFound(tt.err))
})
}
}
func TestIsQMPNodeInUse(t *testing.T) {
tests := []struct {
name string
err error
want bool
}{
{name: "nil", err: nil, want: false},
{name: "matches 'is in use'", err: errors.New("QMP error: GenericError: Node 'nbd-x' is in use"), want: true},
{name: "matches 'still in use'", err: errors.New("QMP error: GenericError: Node 'nbd-x' is still in use"), want: true},
{name: "other QMP error", err: errors.New("QMP error: GenericError: not found"), want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, isQMPNodeInUse(tt.err))
})
}
}
func TestAttachVolume_InstanceNotFound(t *testing.T) {
m := NewManager()
_, err := m.AttachVolume("i-missing", "vol-1", "")
require.Error(t, err)
assert.ErrorIs(t, err, ErrInstanceNotFound)
}
func TestAttachVolume_NotRunning(t *testing.T) {
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateStopped, Instance: &ec2.Instance{}})
_, err := m.AttachVolume("i-1", "vol-1", "")
require.Error(t, err)
assert.ErrorIs(t, err, ErrInvalidTransition)
}
func TestAttachVolume_AttachmentLimitExceeded(t *testing.T) {
bdms := make([]*ec2.InstanceBlockDeviceMapping, 0, 11)
for c := 'f'; c <= 'p'; c++ {
dev := fmt.Sprintf("/dev/sd%c", c)
bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{DeviceName: aws.String(dev)})
}
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateRunning, Instance: &ec2.Instance{BlockDeviceMappings: bdms}})
_, err := m.AttachVolume("i-1", "vol-1", "")
require.Error(t, err)
assert.ErrorIs(t, err, ErrAttachmentLimitExceeded)
}
func TestDetachVolume_InstanceNotFound(t *testing.T) {
m := NewManager()
_, err := m.DetachVolume("i-missing", "vol-1", "", false)
require.Error(t, err)
assert.ErrorIs(t, err, ErrInstanceNotFound)
}
func TestDetachVolume_NotRunning(t *testing.T) {
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateStopped})
_, err := m.DetachVolume("i-1", "vol-1", "", false)
require.Error(t, err)
assert.ErrorIs(t, err, ErrInvalidTransition)
}
func TestDetachVolume_VolumeNotAttached(t *testing.T) {
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateRunning, Instance: &ec2.Instance{}})
_, err := m.DetachVolume("i-1", "vol-missing", "", false)
require.Error(t, err)
assert.ErrorIs(t, err, ErrVolumeNotAttached)
}
func TestDetachVolume_BootVolumeRejected(t *testing.T) {
m := NewManager()
m.Insert(&VM{
ID: "i-1",
Status: StateRunning,
Instance: &ec2.Instance{},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{{Name: "vol-boot", Boot: true}},
},
})
_, err := m.DetachVolume("i-1", "vol-boot", "", false)
require.Error(t, err)
assert.ErrorIs(t, err, ErrVolumeNotDetachable)
}
func TestDetachVolume_DeviceMismatch(t *testing.T) {
m := NewManager()
m.Insert(&VM{
ID: "i-1",
Status: StateRunning,
Instance: &ec2.Instance{},
EBSRequests: types.EBSRequests{
Requests: []types.EBSRequest{{Name: "vol-1", DeviceName: "/dev/sdf"}},
},
})
_, err := m.DetachVolume("i-1", "vol-1", "/dev/sdg", false)
require.Error(t, err)
assert.ErrorIs(t, err, ErrVolumeDeviceMismatch)
}
func TestReboot_InstanceNotFound(t *testing.T) {
m := NewManager()
err := m.Reboot("i-missing")
require.Error(t, err)
assert.ErrorIs(t, err, ErrInstanceNotFound)
}
func TestReboot_NotRunning(t *testing.T) {
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateStopped})
err := m.Reboot("i-1")
require.Error(t, err)
assert.ErrorIs(t, err, ErrInvalidTransition)
}
// TestReboot_DoesNotFireHooks asserts that Manager.Reboot fires neither
// OnInstanceUp nor OnInstanceDown. The plan's "Hook firing contract" requires
// hooks to fire only on Pending→Running and terminal transitions; reboot keeps
// the VM in StateRunning across system_reset, so firing OnInstanceDown +
// OnInstanceUp would tear down per-instance NATS subs on every reboot.
func TestReboot_DoesNotFireHooks(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
return nil // success
})
defer cancel()
var upCalls, downCalls int
hooks := ManagerHooks{
OnInstanceUp: func(*VM) { upCalls++ },
OnInstanceDown: func(string) { downCalls++ },
}
m := NewManagerWithDeps(Deps{Hooks: hooks})
m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
require.NoError(t, m.Reboot("i-1"))
assert.Equal(t, 0, upCalls, "Reboot must not fire OnInstanceUp")
assert.Equal(t, 0, downCalls, "Reboot must not fire OnInstanceDown")
assert.Equal(t, []string{"system_reset"}, recorder.executes(),
"Reboot must issue exactly one system_reset QMP command")
}
// TestReboot_DoesNotChangeStatus asserts the VM stays in StateRunning across
// reboot. Pairs with TestReboot_DoesNotFireHooks to lock down the hook
// contract: status doesn't transition, so hooks shouldn't fire.
func TestReboot_DoesNotChangeStatus(t *testing.T) {
qmpClient, cancel := newMockQMPClient(t, nil)
defer cancel()
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
require.NoError(t, m.Reboot("i-1"))
v, ok := m.Get("i-1")
require.True(t, ok)
var status InstanceState
m.Inspect(v, func(v *VM) { status = v.Status })
assert.Equal(t, StateRunning, status)
}
func TestReboot_QMPFailureSurfacesError(t *testing.T) {
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
return map[string]any{
"error": map[string]any{
"class": "GenericError",
"desc": "guest is not running",
},
}
})
defer cancel()
m := NewManager()
m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
err := m.Reboot("i-1")
require.Error(t, err)
assert.Contains(t, err.Error(), "QMP system_reset")
}
// attachVolumeRunningInstance returns a manager with a single running VM
// wired with the supplied QMP client. Used by AttachVolume rollback tests.
func attachVolumeRunningInstance(t *testing.T, qmpClient *qmp.QMPClient, mounter VolumeMounter) (*Manager, *VM) {
t.Helper()
m := NewManagerWithDeps(Deps{VolumeMounter: mounter})
v := &VM{
ID: "i-1",
Status: StateRunning,
Instance: &ec2.Instance{},
QMPClient: qmpClient,
}
m.Insert(v)
return m, v
}
// TestAttachVolume_MountAmbiguous_TriggersUnmountOne covers the regression
// from issue #1: pre-2d the daemon called rollbackEBSMount when the backend
// acknowledged the mount with an empty URI. 2d collapsed the empty-URI case
// into an opaque MountOne error and dropped the rollback. The fix introduces
// ErrMountAmbiguous so AttachVolume can detect this specific case and call
// UnmountOne defensively.
func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) {
mounter := &fakeVolumeMounter{mountOneErr: ErrMountAmbiguous}
m, _ := attachVolumeRunningInstance(t, nil, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.ErrorIs(t, err, ErrMountAmbiguous)
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne,
"AttachVolume must call UnmountOne to clean up the ambiguous backend mount")
}
// TestAttachVolume_GenericMountError_DoesNotUnmount confirms that other
// MountOne failures (NATS timeout, backend explicit Error, marshal errors)
// do NOT trigger UnmountOne — matches pre-2d behaviour where the daemon
// only rolled back on the empty-URI case, not on mountResp.Error != "".
func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) {
mounter := &fakeVolumeMounter{mountOneErr: errors.New("ebs.mount NATS request: timeout")}
m, _ := attachVolumeRunningInstance(t, nil, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.NotErrorIs(t, err, ErrMountAmbiguous)
assert.Empty(t, mounter.unmountedOne,
"non-ambiguous MountOne errors must not trigger UnmountOne (matches pre-2d)")
}
// TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne covers the rollback
// after a successful mount that returns a malformed NBDURI. AttachVolume must
// unmount the backend before bailing.
func TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne(t *testing.T) {
mounter := &fakeVolumeMounter{mountOneURI: "not-a-valid-nbd-uri"}
m, _ := attachVolumeRunningInstance(t, nil, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "parse NBDURI")
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).
func TestAttachVolume_ObjectAddFailure_TriggersUnmountOne(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
if cmd.Execute == "object-add" {
return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "iothread limit"}}
}
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "object-add")
assert.Equal(t, []string{"object-add"}, recorder.executes(),
"object-add failure must short-circuit before blockdev-add")
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
}
// TestAttachVolume_BlockdevAddFailure_TriggersUnmountOne covers QMP
// blockdev-add failure: AttachVolume must unmount and bail before issuing
// device_add.
func TestAttachVolume_BlockdevAddFailure_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": "nbd connect failed"}}
}
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume("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"}, recorder.executes())
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
}
// TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts covers the
// device_add-fail-then-rollback path where blockdev-del succeeds. The
// rollback chain must run blockdev-del then UnmountOne in order.
func TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
if cmd.Execute == "device_add" {
return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "no PCI slot"}}
}
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "device_add")
assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes())
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne,
"successful blockdev-del rollback must be followed by UnmountOne")
}
// TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne
// preserves the load-bearing invariant in the rollback chain: when
// blockdev-del fails, AttachVolume must NOT call UnmountOne. Tearing down
// the NBD server while QEMU still holds the block node would crash the VM.
// A future refactor that "fixes" the apparent bug by always unmounting
// would crash production VMs — this test is the regression guard.
func TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
switch cmd.Execute {
case "device_add":
return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "no PCI slot"}}
case "blockdev-del":
return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "node still in use"}}
}
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "device_add")
assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes())
assert.Empty(t, mounter.unmountedOne,
"failed blockdev-del must skip UnmountOne — unmounting under a live block node crashes QEMU")
}